detached 2795b89

feat(app,client): attach flow, CLI dispatch, daemon auto-spawn with mkdir lock

commit
2795b89f1ad8d43b2ac0b2c9301c94949f62ea6a
author
brice · · 2026-07-02 11:29 UTC
parent
36aa808
signature
GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email
9 files · +1337 -3
Unified
UnifiedSplit
A internal/adapters/ipc/dial.go +22 -0
@@ -0,0 +1,22 @@
1+package ipc
2+
3+import (
4+ "fmt"
5+ "net"
6+ "path/filepath"
7+
8+ "github.com/bnema/vev/internal/ports"
9+)
10+
11+// Dial connects to the daemon's socket in dir (dir/daemon.sock) and returns
12+// a Transport speaking vev's framed protocol. A dial failure (no daemon
13+// listening, or a stale/absent socket) is reported as a wrapped error so
14+// callers can decide whether to spawn a daemon and retry.
15+func Dial(dir string) (ports.Transport, error) {
16+ sockPath := filepath.Join(dir, socketFileName)
17+ conn, err := net.Dial("unix", sockPath)
18+ if err != nil {
19+ return nil, fmt.Errorf("ipc: dial %s: %w", sockPath, err)
20+ }
21+ return NewTransport(conn), nil
22+}
@@ -0,0 +1,22 @@
1+package ipc
2+
3+import (
4+ "fmt"
5+ "net"
6+ "path/filepath"
7+
8+ "github.com/bnema/vev/internal/ports"
9+)
10+
11+// Dial connects to the daemon's socket in dir (dir/daemon.sock) and returns
12+// a Transport speaking vev's framed protocol. A dial failure (no daemon
13+// listening, or a stale/absent socket) is reported as a wrapped error so
14+// callers can decide whether to spawn a daemon and retry.
15+func Dial(dir string) (ports.Transport, error) {
16+ sockPath := filepath.Join(dir, socketFileName)
17+ conn, err := net.Dial("unix", sockPath)
18+ if err != nil {
19+ return nil, fmt.Errorf("ipc: dial %s: %w", sockPath, err)
20+ }
21+ return NewTransport(conn), nil
22+}
A internal/app/logging.go +62 -0
@@ -0,0 +1,62 @@
1+package app
2+
3+import (
4+ "fmt"
5+ "log/slog"
6+ "os"
7+ "path/filepath"
8+ "strings"
9+)
10+
11+// logFileName is the fixed name of vev's shared log file, used by both the
12+// daemon and the raw-attached client (which must never log to the console).
13+const logFileName = "vev.log"
14+
15+// stateDir returns the directory vev writes its log to: $XDG_STATE_HOME/vev
16+// if set, else ~/.local/state/vev.
17+func stateDir() string {
18+ if xdg := os.Getenv("XDG_STATE_HOME"); xdg != "" {
19+ return filepath.Join(xdg, "vev")
20+ }
21+ home, err := os.UserHomeDir()
22+ if err != nil || home == "" {
23+ // Last resort: a per-uid temp path so logging never aborts startup.
24+ return filepath.Join(os.TempDir(), fmt.Sprintf("vev-state-%d", os.Getuid()))
25+ }
26+ return filepath.Join(home, ".local", "state", "vev")
27+}
28+
29+// parseLogLevel maps VEV_LOG to a slog level, defaulting to info.
30+func parseLogLevel(s string) slog.Level {
31+ switch strings.ToLower(strings.TrimSpace(s)) {
32+ case "debug":
33+ return slog.LevelDebug
34+ case "warn", "warning":
35+ return slog.LevelWarn
36+ case "error":
37+ return slog.LevelError
38+ default:
39+ return slog.LevelInfo
40+ }
41+}
42+
43+// setupLogging installs a JSON slog handler writing to the shared log file
44+// and returns the open file so the caller can close it on exit. The level
45+// comes from VEV_LOG (default info). Both the daemon and the client route
46+// diagnostics here — the client because console output while the terminal is
47+// raw would corrupt the display.
48+//
49+// No log rotation is performed for the MVP (future work).
50+func setupLogging() (*os.File, error) {
51+ dir := stateDir()
52+ if err := os.MkdirAll(dir, 0o700); err != nil {
53+ return nil, fmt.Errorf("vev: creating state dir: %w", err)
54+ }
55+ f, err := os.OpenFile(filepath.Join(dir, logFileName), os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0o600)
56+ if err != nil {
57+ return nil, fmt.Errorf("vev: opening log file: %w", err)
58+ }
59+ handler := slog.NewJSONHandler(f, &slog.HandlerOptions{Level: parseLogLevel(os.Getenv("VEV_LOG"))})
60+ slog.SetDefault(slog.New(handler))
61+ return f, nil
62+}
@@ -0,0 +1,62 @@
1+package app
2+
3+import (
4+ "fmt"
5+ "log/slog"
6+ "os"
7+ "path/filepath"
8+ "strings"
9+)
10+
11+// logFileName is the fixed name of vev's shared log file, used by both the
12+// daemon and the raw-attached client (which must never log to the console).
13+const logFileName = "vev.log"
14+
15+// stateDir returns the directory vev writes its log to: $XDG_STATE_HOME/vev
16+// if set, else ~/.local/state/vev.
17+func stateDir() string {
18+ if xdg := os.Getenv("XDG_STATE_HOME"); xdg != "" {
19+ return filepath.Join(xdg, "vev")
20+ }
21+ home, err := os.UserHomeDir()
22+ if err != nil || home == "" {
23+ // Last resort: a per-uid temp path so logging never aborts startup.
24+ return filepath.Join(os.TempDir(), fmt.Sprintf("vev-state-%d", os.Getuid()))
25+ }
26+ return filepath.Join(home, ".local", "state", "vev")
27+}
28+
29+// parseLogLevel maps VEV_LOG to a slog level, defaulting to info.
30+func parseLogLevel(s string) slog.Level {
31+ switch strings.ToLower(strings.TrimSpace(s)) {
32+ case "debug":
33+ return slog.LevelDebug
34+ case "warn", "warning":
35+ return slog.LevelWarn
36+ case "error":
37+ return slog.LevelError
38+ default:
39+ return slog.LevelInfo
40+ }
41+}
42+
43+// setupLogging installs a JSON slog handler writing to the shared log file
44+// and returns the open file so the caller can close it on exit. The level
45+// comes from VEV_LOG (default info). Both the daemon and the client route
46+// diagnostics here — the client because console output while the terminal is
47+// raw would corrupt the display.
48+//
49+// No log rotation is performed for the MVP (future work).
50+func setupLogging() (*os.File, error) {
51+ dir := stateDir()
52+ if err := os.MkdirAll(dir, 0o700); err != nil {
53+ return nil, fmt.Errorf("vev: creating state dir: %w", err)
54+ }
55+ f, err := os.OpenFile(filepath.Join(dir, logFileName), os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0o600)
56+ if err != nil {
57+ return nil, fmt.Errorf("vev: opening log file: %w", err)
58+ }
59+ handler := slog.NewJSONHandler(f, &slog.HandlerOptions{Level: parseLogLevel(os.Getenv("VEV_LOG"))})
60+ slog.SetDefault(slog.New(handler))
61+ return f, nil
62+}
M internal/app/run.go +249 -3
@@ -1,8 +1,254 @@
1-// Package app wires vev's runtime together via dependency injection.
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.
25 package app
36
4-// Run is the entry point invoked by main. It parses no flags yet and
5-// performs no work; later milestones will wire the application here.
7+import (
8+ "context"
9+ "errors"
10+ "fmt"
11+ "io"
12+ "log/slog"
13+ "os"
14+ "text/tabwriter"
15+
16+ "github.com/bnema/vev/internal/adapters/ipc"
17+ "github.com/bnema/vev/internal/adapters/term"
18+ "github.com/bnema/vev/internal/ports"
19+ "github.com/bnema/vev/internal/usecase/client"
20+ "github.com/bnema/vev/internal/usecase/daemon"
21+)
22+
23+// cmdKind identifies which sub-command the CLI parsed.
24+type cmdKind int
25+
26+const (
27+ kindAttach cmdKind = iota // ephemeral/new/attach — distinguished by intent
28+ kindList
29+ kindKill
30+ kindDaemon
31+ kindHelp
32+ kindVersion
33+)
34+
35+// command is the parsed CLI invocation: what to do, plus the attach intent
36+// and session name where relevant.
37+type command struct {
38+ kind cmdKind
39+ intent uint8
40+ name string
41+}
42+
43+// usageError is a user-facing argument error; the app prints it (with usage)
44+// rather than a stack of wrapped internals.
45+type usageError struct{ msg string }
46+
47+func (e *usageError) Error() string { return "vev: " + e.msg }
48+
49+func usagef(format string, args ...any) error {
50+ return &usageError{msg: fmt.Sprintf(format, args...)}
51+}
52+
53+const usageText = `vev — a terminal multiplexer
54+
55+usage:
56+ vev attach to (or create) an ephemeral session
57+ vev new <name> create and attach to a named session
58+ vev attach <name> attach to an existing named session (alias: a)
59+ vev ls list sessions
60+ vev kill <name> kill a named session
61+ vev --help show this help
62+ vev --version show version`
63+
64+// version is vev's reported build version. Wired here for the MVP; a build
65+// stamp can replace it later.
66+const version = "0.1.0-dev"
67+
68+// Run is the entry point invoked by main. It parses args into a command and
69+// dispatches it.
670 func Run(args []string) error {
71+ cmd, err := parseArgs(args)
72+ if err != nil {
73+ return err
74+ }
75+ return dispatch(context.Background(), cmd)
76+}
77+
78+// parseArgs turns the raw argv tail into a command. It is deliberately a
79+// pure function so the full dispatch table can be unit-tested without any
80+// I/O.
81+func parseArgs(args []string) (command, error) {
82+ if len(args) == 0 {
83+ return command{kind: kindAttach, intent: ipc.IntentEphemeral}, nil
84+ }
85+
86+ switch args[0] {
87+ case "--daemon":
88+ return command{kind: kindDaemon}, nil
89+ case "new":
90+ if len(args) < 2 || args[1] == "" {
91+ return command{}, usagef("`new` requires a session name")
92+ }
93+ return command{kind: kindAttach, intent: ipc.IntentNew, name: args[1]}, nil
94+ case "attach", "a":
95+ if len(args) < 2 || args[1] == "" {
96+ return command{}, usagef("`attach` requires a session name")
97+ }
98+ return command{kind: kindAttach, intent: ipc.IntentAttach, name: args[1]}, nil
99+ case "ls", "list":
100+ return command{kind: kindList}, nil
101+ case "kill":
102+ if len(args) < 2 || args[1] == "" {
103+ return command{}, usagef("`kill` requires a session name")
104+ }
105+ return command{kind: kindKill, name: args[1]}, nil
106+ case "-h", "--help", "help":
107+ return command{kind: kindHelp}, nil
108+ case "--version", "version":
109+ return command{kind: kindVersion}, nil
110+ default:
111+ return command{}, usagef("unknown command %q", args[0])
112+ }
113+}
114+
115+// dispatch runs the parsed command.
116+func dispatch(ctx context.Context, cmd command) error {
117+ switch cmd.kind {
118+ case kindHelp:
119+ fmt.Println(usageText)
120+ return nil
121+ case kindVersion:
122+ fmt.Println("vev", version)
123+ return nil
124+ case kindDaemon:
125+ return runDaemon()
126+ case kindList:
127+ return runList(ctx)
128+ case kindKill:
129+ return runKill(ctx, cmd.name)
130+ case kindAttach:
131+ return runAttach(ctx, cmd.intent, cmd.name)
132+ default:
133+ return usagef("unhandled command")
134+ }
135+}
136+
137+// runDaemon runs the daemon in the foreground (the hidden --daemon path,
138+// entered by an auto-spawned child). The daemon core lands in Task 11; here
139+// we set up logging, bind the socket, and hand off to daemon.Serve.
140+func runDaemon() error {
141+ logFile, err := setupLogging()
142+ if err != nil {
143+ return err
144+ }
145+ defer logFile.Close()
146+
147+ ln, err := ipc.Listen(ipc.SocketDir())
148+ if err != nil {
149+ return fmt.Errorf("vev: daemon listen: %w", err)
150+ }
151+ defer ln.Close()
152+
153+ slog.Info("daemon starting", "socket", ln.Addr())
154+ err = daemon.Serve(ln)
155+ slog.Error("daemon exited", "err", err)
156+ return err
157+}
158+
159+// runAttach dials (auto-spawning the daemon if needed) and runs the client
160+// attach loop. Logging goes to the shared file: the client must never write
161+// to the console while the terminal is raw.
162+func runAttach(ctx context.Context, intent uint8, name string) error {
163+ logFile, err := setupLogging()
164+ if err != nil {
165+ return err
166+ }
167+ defer logFile.Close()
168+
169+ transport, err := ensureDaemon(ctx, ipc.SocketDir(), realDial, realSpawn, defaultBackoff)
170+ if err != nil {
171+ return err
172+ }
173+ // client.Attach owns and closes transport.
174+ return client.Attach(ctx, transport, term.New(), intent, name)
175+}
176+
177+// runList prints the daemon's session listing. With no daemon running there
178+// are, by definition, no sessions.
179+func runList(_ context.Context) error {
180+ transport, err := realDial(ipc.SocketDir())
181+ if err != nil {
182+ fmt.Println("no sessions")
183+ return nil
184+ }
185+ defer transport.Close()
186+
187+ if err := transport.Send(ports.Frame{Type: ports.MsgList, Payload: ipc.MarshalList(ipc.List{})}); err != nil {
188+ return fmt.Errorf("vev: requesting session list: %w", err)
189+ }
190+ reply, err := transport.Recv()
191+ if err != nil {
192+ return fmt.Errorf("vev: reading session list: %w", err)
193+ }
194+ if reply.Type == ports.MsgError {
195+ em, _ := ipc.UnmarshalErrorMsg(reply.Payload)
196+ return fmt.Errorf("vev: %s", em.Text)
197+ }
198+ if reply.Type != ports.MsgSessions {
199+ return fmt.Errorf("vev: unexpected reply type %d to list", reply.Type)
200+ }
201+ sessions, err := ipc.UnmarshalSessions(reply.Payload)
202+ if err != nil {
203+ return fmt.Errorf("vev: decoding session list: %w", err)
204+ }
205+ printSessions(os.Stdout, sessions.Sessions)
206+ return nil
207+}
208+
209+// printSessions renders a session table (or a friendly note when empty).
210+func printSessions(w io.Writer, sessions []ipc.SessionInfo) {
211+ if len(sessions) == 0 {
212+ fmt.Fprintln(w, "no sessions")
213+ return
214+ }
215+ tw := tabwriter.NewWriter(w, 0, 4, 2, ' ', 0)
216+ fmt.Fprintln(tw, "NAME\tWINDOWS\tATTACHED")
217+ for _, s := range sessions {
218+ attached := "no"
219+ if s.Attached {
220+ attached = "yes"
221+ }
222+ fmt.Fprintf(tw, "%s\t%d\t%s\n", s.Name, s.Windows, attached)
223+ }
224+ _ = tw.Flush()
225+}
226+
227+// runKill asks the daemon to terminate a named session.
228+func runKill(_ context.Context, name string) error {
229+ transport, err := realDial(ipc.SocketDir())
230+ if err != nil {
231+ return fmt.Errorf("vev: no daemon running")
232+ }
233+ defer transport.Close()
234+
235+ if err := transport.Send(ports.Frame{Type: ports.MsgKill, Payload: ipc.MarshalKill(ipc.Kill{Name: name})}); err != nil {
236+ return fmt.Errorf("vev: requesting kill: %w", err)
237+ }
238+ reply, err := transport.Recv()
239+ if err != nil {
240+ // The daemon may close the connection after killing; treat a clean
241+ // EOF as success.
242+ if errors.Is(err, io.EOF) {
243+ fmt.Printf("killed %s\n", name)
244+ return nil
245+ }
246+ return fmt.Errorf("vev: reading kill reply: %w", err)
247+ }
248+ if reply.Type == ports.MsgError {
249+ em, _ := ipc.UnmarshalErrorMsg(reply.Payload)
250+ return fmt.Errorf("vev: %s", em.Text)
251+ }
252+ fmt.Printf("killed %s\n", name)
7253 return nil
8254 }
@@ -1,8 +1,254 @@
1-// Package app wires vev's runtime together via dependency injection.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.
2 package app5 package app
3 6
4-// Run is the entry point invoked by main. It parses no flags yet and7+import (
5-// performs no work; later milestones will wire the application here.8+ "context"
9+ "errors"
10+ "fmt"
11+ "io"
12+ "log/slog"
13+ "os"
14+ "text/tabwriter"
15+
16+ "github.com/bnema/vev/internal/adapters/ipc"
17+ "github.com/bnema/vev/internal/adapters/term"
18+ "github.com/bnema/vev/internal/ports"
19+ "github.com/bnema/vev/internal/usecase/client"
20+ "github.com/bnema/vev/internal/usecase/daemon"
21+)
22+
23+// cmdKind identifies which sub-command the CLI parsed.
24+type cmdKind int
25+
26+const (
27+ kindAttach cmdKind = iota // ephemeral/new/attach — distinguished by intent
28+ kindList
29+ kindKill
30+ kindDaemon
31+ kindHelp
32+ kindVersion
33+)
34+
35+// command is the parsed CLI invocation: what to do, plus the attach intent
36+// and session name where relevant.
37+type command struct {
38+ kind cmdKind
39+ intent uint8
40+ name string
41+}
42+
43+// usageError is a user-facing argument error; the app prints it (with usage)
44+// rather than a stack of wrapped internals.
45+type usageError struct{ msg string }
46+
47+func (e *usageError) Error() string { return "vev: " + e.msg }
48+
49+func usagef(format string, args ...any) error {
50+ return &usageError{msg: fmt.Sprintf(format, args...)}
51+}
52+
53+const usageText = `vev — a terminal multiplexer
54+
55+usage:
56+ vev attach to (or create) an ephemeral session
57+ vev new <name> create and attach to a named session
58+ vev attach <name> attach to an existing named session (alias: a)
59+ vev ls list sessions
60+ vev kill <name> kill a named session
61+ vev --help show this help
62+ vev --version show version`
63+
64+// version is vev's reported build version. Wired here for the MVP; a build
65+// stamp can replace it later.
66+const version = "0.1.0-dev"
67+
68+// Run is the entry point invoked by main. It parses args into a command and
69+// dispatches it.
6 func Run(args []string) error {70 func Run(args []string) error {
71+ cmd, err := parseArgs(args)
72+ if err != nil {
73+ return err
74+ }
75+ return dispatch(context.Background(), cmd)
76+}
77+
78+// parseArgs turns the raw argv tail into a command. It is deliberately a
79+// pure function so the full dispatch table can be unit-tested without any
80+// I/O.
81+func parseArgs(args []string) (command, error) {
82+ if len(args) == 0 {
83+ return command{kind: kindAttach, intent: ipc.IntentEphemeral}, nil
84+ }
85+
86+ switch args[0] {
87+ case "--daemon":
88+ return command{kind: kindDaemon}, nil
89+ case "new":
90+ if len(args) < 2 || args[1] == "" {
91+ return command{}, usagef("`new` requires a session name")
92+ }
93+ return command{kind: kindAttach, intent: ipc.IntentNew, name: args[1]}, nil
94+ case "attach", "a":
95+ if len(args) < 2 || args[1] == "" {
96+ return command{}, usagef("`attach` requires a session name")
97+ }
98+ return command{kind: kindAttach, intent: ipc.IntentAttach, name: args[1]}, nil
99+ case "ls", "list":
100+ return command{kind: kindList}, nil
101+ case "kill":
102+ if len(args) < 2 || args[1] == "" {
103+ return command{}, usagef("`kill` requires a session name")
104+ }
105+ return command{kind: kindKill, name: args[1]}, nil
106+ case "-h", "--help", "help":
107+ return command{kind: kindHelp}, nil
108+ case "--version", "version":
109+ return command{kind: kindVersion}, nil
110+ default:
111+ return command{}, usagef("unknown command %q", args[0])
112+ }
113+}
114+
115+// dispatch runs the parsed command.
116+func dispatch(ctx context.Context, cmd command) error {
117+ switch cmd.kind {
118+ case kindHelp:
119+ fmt.Println(usageText)
120+ return nil
121+ case kindVersion:
122+ fmt.Println("vev", version)
123+ return nil
124+ case kindDaemon:
125+ return runDaemon()
126+ case kindList:
127+ return runList(ctx)
128+ case kindKill:
129+ return runKill(ctx, cmd.name)
130+ case kindAttach:
131+ return runAttach(ctx, cmd.intent, cmd.name)
132+ default:
133+ return usagef("unhandled command")
134+ }
135+}
136+
137+// runDaemon runs the daemon in the foreground (the hidden --daemon path,
138+// entered by an auto-spawned child). The daemon core lands in Task 11; here
139+// we set up logging, bind the socket, and hand off to daemon.Serve.
140+func runDaemon() error {
141+ logFile, err := setupLogging()
142+ if err != nil {
143+ return err
144+ }
145+ defer logFile.Close()
146+
147+ ln, err := ipc.Listen(ipc.SocketDir())
148+ if err != nil {
149+ return fmt.Errorf("vev: daemon listen: %w", err)
150+ }
151+ defer ln.Close()
152+
153+ slog.Info("daemon starting", "socket", ln.Addr())
154+ err = daemon.Serve(ln)
155+ slog.Error("daemon exited", "err", err)
156+ return err
157+}
158+
159+// runAttach dials (auto-spawning the daemon if needed) and runs the client
160+// attach loop. Logging goes to the shared file: the client must never write
161+// to the console while the terminal is raw.
162+func runAttach(ctx context.Context, intent uint8, name string) error {
163+ logFile, err := setupLogging()
164+ if err != nil {
165+ return err
166+ }
167+ defer logFile.Close()
168+
169+ transport, err := ensureDaemon(ctx, ipc.SocketDir(), realDial, realSpawn, defaultBackoff)
170+ if err != nil {
171+ return err
172+ }
173+ // client.Attach owns and closes transport.
174+ return client.Attach(ctx, transport, term.New(), intent, name)
175+}
176+
177+// runList prints the daemon's session listing. With no daemon running there
178+// are, by definition, no sessions.
179+func runList(_ context.Context) error {
180+ transport, err := realDial(ipc.SocketDir())
181+ if err != nil {
182+ fmt.Println("no sessions")
183+ return nil
184+ }
185+ defer transport.Close()
186+
187+ if err := transport.Send(ports.Frame{Type: ports.MsgList, Payload: ipc.MarshalList(ipc.List{})}); err != nil {
188+ return fmt.Errorf("vev: requesting session list: %w", err)
189+ }
190+ reply, err := transport.Recv()
191+ if err != nil {
192+ return fmt.Errorf("vev: reading session list: %w", err)
193+ }
194+ if reply.Type == ports.MsgError {
195+ em, _ := ipc.UnmarshalErrorMsg(reply.Payload)
196+ return fmt.Errorf("vev: %s", em.Text)
197+ }
198+ if reply.Type != ports.MsgSessions {
199+ return fmt.Errorf("vev: unexpected reply type %d to list", reply.Type)
200+ }
201+ sessions, err := ipc.UnmarshalSessions(reply.Payload)
202+ if err != nil {
203+ return fmt.Errorf("vev: decoding session list: %w", err)
204+ }
205+ printSessions(os.Stdout, sessions.Sessions)
206+ return nil
207+}
208+
209+// printSessions renders a session table (or a friendly note when empty).
210+func printSessions(w io.Writer, sessions []ipc.SessionInfo) {
211+ if len(sessions) == 0 {
212+ fmt.Fprintln(w, "no sessions")
213+ return
214+ }
215+ tw := tabwriter.NewWriter(w, 0, 4, 2, ' ', 0)
216+ fmt.Fprintln(tw, "NAME\tWINDOWS\tATTACHED")
217+ for _, s := range sessions {
218+ attached := "no"
219+ if s.Attached {
220+ attached = "yes"
221+ }
222+ fmt.Fprintf(tw, "%s\t%d\t%s\n", s.Name, s.Windows, attached)
223+ }
224+ _ = tw.Flush()
225+}
226+
227+// runKill asks the daemon to terminate a named session.
228+func runKill(_ context.Context, name string) error {
229+ transport, err := realDial(ipc.SocketDir())
230+ if err != nil {
231+ return fmt.Errorf("vev: no daemon running")
232+ }
233+ defer transport.Close()
234+
235+ if err := transport.Send(ports.Frame{Type: ports.MsgKill, Payload: ipc.MarshalKill(ipc.Kill{Name: name})}); err != nil {
236+ return fmt.Errorf("vev: requesting kill: %w", err)
237+ }
238+ reply, err := transport.Recv()
239+ if err != nil {
240+ // The daemon may close the connection after killing; treat a clean
241+ // EOF as success.
242+ if errors.Is(err, io.EOF) {
243+ fmt.Printf("killed %s\n", name)
244+ return nil
245+ }
246+ return fmt.Errorf("vev: reading kill reply: %w", err)
247+ }
248+ if reply.Type == ports.MsgError {
249+ em, _ := ipc.UnmarshalErrorMsg(reply.Payload)
250+ return fmt.Errorf("vev: %s", em.Text)
251+ }
252+ fmt.Printf("killed %s\n", name)
7 return nil253 return nil
8 }254 }
A internal/app/run_test.go +65 -0
@@ -0,0 +1,65 @@
1+package app
2+
3+import (
4+ "errors"
5+ "testing"
6+
7+ "github.com/bnema/vev/internal/adapters/ipc"
8+)
9+
10+func TestParseArgs(t *testing.T) {
11+ tests := []struct {
12+ name string
13+ args []string
14+ wantKind cmdKind
15+ wantIntent uint8
16+ wantName string
17+ wantErr bool
18+ }{
19+ {name: "no args -> ephemeral attach", args: nil, wantKind: kindAttach, wantIntent: ipc.IntentEphemeral},
20+ {name: "empty slice -> ephemeral attach", args: []string{}, wantKind: kindAttach, wantIntent: ipc.IntentEphemeral},
21+ {name: "new named", args: []string{"new", "work"}, wantKind: kindAttach, wantIntent: ipc.IntentNew, wantName: "work"},
22+ {name: "new without name", args: []string{"new"}, wantErr: true},
23+ {name: "new empty name", args: []string{"new", ""}, wantErr: true},
24+ {name: "attach named", args: []string{"attach", "work"}, wantKind: kindAttach, wantIntent: ipc.IntentAttach, wantName: "work"},
25+ {name: "attach alias a", args: []string{"a", "work"}, wantKind: kindAttach, wantIntent: ipc.IntentAttach, wantName: "work"},
26+ {name: "attach without name", args: []string{"attach"}, wantErr: true},
27+ {name: "ls", args: []string{"ls"}, wantKind: kindList},
28+ {name: "list", args: []string{"list"}, wantKind: kindList},
29+ {name: "kill named", args: []string{"kill", "work"}, wantKind: kindKill, wantName: "work"},
30+ {name: "kill without name", args: []string{"kill"}, wantErr: true},
31+ {name: "daemon", args: []string{"--daemon"}, wantKind: kindDaemon},
32+ {name: "help", args: []string{"--help"}, wantKind: kindHelp},
33+ {name: "help subcommand", args: []string{"help"}, wantKind: kindHelp},
34+ {name: "version", args: []string{"--version"}, wantKind: kindVersion},
35+ {name: "unknown", args: []string{"frobnicate"}, wantErr: true},
36+ }
37+
38+ for _, tt := range tests {
39+ t.Run(tt.name, func(t *testing.T) {
40+ got, err := parseArgs(tt.args)
41+ if tt.wantErr {
42+ if err == nil {
43+ t.Fatalf("parseArgs(%q) = %+v, want error", tt.args, got)
44+ }
45+ var ue *usageError
46+ if !errors.As(err, &ue) {
47+ t.Fatalf("parseArgs(%q) error = %T, want *usageError", tt.args, err)
48+ }
49+ return
50+ }
51+ if err != nil {
52+ t.Fatalf("parseArgs(%q) unexpected error: %v", tt.args, err)
53+ }
54+ if got.kind != tt.wantKind {
55+ t.Errorf("kind = %v, want %v", got.kind, tt.wantKind)
56+ }
57+ if got.kind == kindAttach && got.intent != tt.wantIntent {
58+ t.Errorf("intent = %d, want %d", got.intent, tt.wantIntent)
59+ }
60+ if got.name != tt.wantName {
61+ t.Errorf("name = %q, want %q", got.name, tt.wantName)
62+ }
63+ })
64+ }
65+}
@@ -0,0 +1,65 @@
1+package app
2+
3+import (
4+ "errors"
5+ "testing"
6+
7+ "github.com/bnema/vev/internal/adapters/ipc"
8+)
9+
10+func TestParseArgs(t *testing.T) {
11+ tests := []struct {
12+ name string
13+ args []string
14+ wantKind cmdKind
15+ wantIntent uint8
16+ wantName string
17+ wantErr bool
18+ }{
19+ {name: "no args -> ephemeral attach", args: nil, wantKind: kindAttach, wantIntent: ipc.IntentEphemeral},
20+ {name: "empty slice -> ephemeral attach", args: []string{}, wantKind: kindAttach, wantIntent: ipc.IntentEphemeral},
21+ {name: "new named", args: []string{"new", "work"}, wantKind: kindAttach, wantIntent: ipc.IntentNew, wantName: "work"},
22+ {name: "new without name", args: []string{"new"}, wantErr: true},
23+ {name: "new empty name", args: []string{"new", ""}, wantErr: true},
24+ {name: "attach named", args: []string{"attach", "work"}, wantKind: kindAttach, wantIntent: ipc.IntentAttach, wantName: "work"},
25+ {name: "attach alias a", args: []string{"a", "work"}, wantKind: kindAttach, wantIntent: ipc.IntentAttach, wantName: "work"},
26+ {name: "attach without name", args: []string{"attach"}, wantErr: true},
27+ {name: "ls", args: []string{"ls"}, wantKind: kindList},
28+ {name: "list", args: []string{"list"}, wantKind: kindList},
29+ {name: "kill named", args: []string{"kill", "work"}, wantKind: kindKill, wantName: "work"},
30+ {name: "kill without name", args: []string{"kill"}, wantErr: true},
31+ {name: "daemon", args: []string{"--daemon"}, wantKind: kindDaemon},
32+ {name: "help", args: []string{"--help"}, wantKind: kindHelp},
33+ {name: "help subcommand", args: []string{"help"}, wantKind: kindHelp},
34+ {name: "version", args: []string{"--version"}, wantKind: kindVersion},
35+ {name: "unknown", args: []string{"frobnicate"}, wantErr: true},
36+ }
37+
38+ for _, tt := range tests {
39+ t.Run(tt.name, func(t *testing.T) {
40+ got, err := parseArgs(tt.args)
41+ if tt.wantErr {
42+ if err == nil {
43+ t.Fatalf("parseArgs(%q) = %+v, want error", tt.args, got)
44+ }
45+ var ue *usageError
46+ if !errors.As(err, &ue) {
47+ t.Fatalf("parseArgs(%q) error = %T, want *usageError", tt.args, err)
48+ }
49+ return
50+ }
51+ if err != nil {
52+ t.Fatalf("parseArgs(%q) unexpected error: %v", tt.args, err)
53+ }
54+ if got.kind != tt.wantKind {
55+ t.Errorf("kind = %v, want %v", got.kind, tt.wantKind)
56+ }
57+ if got.kind == kindAttach && got.intent != tt.wantIntent {
58+ t.Errorf("intent = %d, want %d", got.intent, tt.wantIntent)
59+ }
60+ if got.name != tt.wantName {
61+ t.Errorf("name = %q, want %q", got.name, tt.wantName)
62+ }
63+ })
64+ }
65+}
A internal/app/spawn.go +156 -0
@@ -0,0 +1,156 @@
1+package app
2+
3+import (
4+ "context"
5+ "errors"
6+ "fmt"
7+ "os"
8+ "os/exec"
9+ "path/filepath"
10+ "syscall"
11+ "time"
12+
13+ "github.com/bnema/vev/internal/adapters/ipc"
14+ "github.com/bnema/vev/internal/ports"
15+)
16+
17+// ErrDaemonUnreachable is returned when the daemon socket never becomes
18+// dialable within the retry budget, despite a spawn attempt.
19+var ErrDaemonUnreachable = errors.New("vev: daemon did not become reachable")
20+
21+// spawnLockName is the mkdir-based lock directory guarding daemon spawn, so
22+// concurrent first-clients elect a single spawner instead of racing to
23+// re-exec multiple daemons.
24+const spawnLockName = "spawn.lock"
25+
26+// staleLockAge bounds how long a spawn lock may persist before it is treated
27+// as abandoned (spawner crashed mid-flight) and forcibly taken over.
28+const staleLockAge = 10 * time.Second
29+
30+// dialFunc dials the daemon socket in dir. Injected so tests can drive the
31+// spawn/backoff logic without a real socket.
32+type dialFunc func(dir string) (ports.Transport, error)
33+
34+// spawnFunc launches a detached daemon process. Injected so tests never
35+// re-exec a real binary.
36+type spawnFunc func() error
37+
38+// backoffConfig parameterises retry-dial timing so tests can shrink it.
39+type backoffConfig struct {
40+ initial time.Duration
41+ max time.Duration
42+ total time.Duration
43+}
44+
45+// defaultBackoff is the production retry schedule: 10ms, 20ms, 40ms... capped
46+// at 500ms, for roughly 5s total.
47+var defaultBackoff = backoffConfig{
48+ initial: 10 * time.Millisecond,
49+ max: 500 * time.Millisecond,
50+ total: 5 * time.Second,
51+}
52+
53+// ensureDaemon returns a transport to a running daemon, spawning one if
54+// necessary. It first tries a plain dial; on failure it elects a single
55+// spawner via an mkdir lock (losers simply wait), then retry-dials with
56+// backoff until the socket is live or the budget expires.
57+func ensureDaemon(ctx context.Context, dir string, dial dialFunc, spawn spawnFunc, cfg backoffConfig) (ports.Transport, error) {
58+ if t, err := dial(dir); err == nil {
59+ return t, nil
60+ }
61+
62+ release, acquired, err := acquireSpawnLock(dir)
63+ if err != nil {
64+ return nil, fmt.Errorf("vev: acquiring spawn lock: %w", err)
65+ }
66+ if acquired {
67+ // We won the election: spawn, and hold the lock until the socket is
68+ // dialable (or spawn fails) so late-arriving clients keep waiting
69+ // rather than spawning a second daemon.
70+ defer release()
71+ if err := spawn(); err != nil {
72+ return nil, fmt.Errorf("vev: spawning daemon: %w", err)
73+ }
74+ }
75+
76+ return retryDial(ctx, dir, dial, cfg)
77+}
78+
79+// acquireSpawnLock attempts to create the spawn-lock directory. It returns
80+// acquired=true with a release closure when it wins, acquired=false (with a
81+// no-op release) when another process holds a fresh lock, and takes over a
82+// lock older than staleLockAge (crash resilience).
83+func acquireSpawnLock(dir string) (release func(), acquired bool, err error) {
84+ if err := os.MkdirAll(dir, 0o700); err != nil {
85+ return nil, false, err
86+ }
87+ path := filepath.Join(dir, spawnLockName)
88+
89+ if mkErr := os.Mkdir(path, 0o700); mkErr == nil {
90+ return func() { _ = os.Remove(path) }, true, nil
91+ } else if !errors.Is(mkErr, os.ErrExist) {
92+ return nil, false, mkErr
93+ }
94+
95+ // Lock exists: take it over only if it looks abandoned.
96+ if info, statErr := os.Stat(path); statErr == nil && time.Since(info.ModTime()) > staleLockAge {
97+ _ = os.Remove(path)
98+ if mkErr := os.Mkdir(path, 0o700); mkErr == nil {
99+ return func() { _ = os.Remove(path) }, true, nil
100+ }
101+ }
102+
103+ // A live peer is spawning; wait for it via retry-dial.
104+ return func() {}, false, nil
105+}
106+
107+// retryDial dials repeatedly with exponential backoff until the daemon
108+// answers, the context is cancelled, or the total budget is exhausted.
109+func retryDial(ctx context.Context, dir string, dial dialFunc, cfg backoffConfig) (ports.Transport, error) {
110+ deadline := time.Now().Add(cfg.total)
111+ backoff := cfg.initial
112+ for {
113+ if t, err := dial(dir); err == nil {
114+ return t, nil
115+ }
116+ if !time.Now().Before(deadline) {
117+ return nil, ErrDaemonUnreachable
118+ }
119+ select {
120+ case <-ctx.Done():
121+ return nil, ctx.Err()
122+ case <-time.After(backoff):
123+ }
124+ if backoff *= 2; backoff > cfg.max {
125+ backoff = cfg.max
126+ }
127+ }
128+}
129+
130+// realDial is the production dialer.
131+func realDial(dir string) (ports.Transport, error) {
132+ return ipc.Dial(dir)
133+}
134+
135+// realSpawn re-execs this binary as a detached daemon: a new session
136+// (Setsid) with stdio bound to /dev/null so it survives the client's exit
137+// and never writes to the client's terminal. It logs via slog to the shared
138+// log file instead.
139+func realSpawn() error {
140+ devNull, err := os.OpenFile(os.DevNull, os.O_RDWR, 0)
141+ if err != nil {
142+ return fmt.Errorf("opening %s: %w", os.DevNull, err)
143+ }
144+ defer devNull.Close()
145+
146+ cmd := exec.Command("/proc/self/exe", "--daemon")
147+ cmd.Stdin = devNull
148+ cmd.Stdout = devNull
149+ cmd.Stderr = devNull
150+ cmd.SysProcAttr = &syscall.SysProcAttr{Setsid: true}
151+ if err := cmd.Start(); err != nil {
152+ return err
153+ }
154+ // Detach: we neither wait for nor signal the daemon after launch.
155+ return cmd.Process.Release()
156+}
@@ -0,0 +1,156 @@
1+package app
2+
3+import (
4+ "context"
5+ "errors"
6+ "fmt"
7+ "os"
8+ "os/exec"
9+ "path/filepath"
10+ "syscall"
11+ "time"
12+
13+ "github.com/bnema/vev/internal/adapters/ipc"
14+ "github.com/bnema/vev/internal/ports"
15+)
16+
17+// ErrDaemonUnreachable is returned when the daemon socket never becomes
18+// dialable within the retry budget, despite a spawn attempt.
19+var ErrDaemonUnreachable = errors.New("vev: daemon did not become reachable")
20+
21+// spawnLockName is the mkdir-based lock directory guarding daemon spawn, so
22+// concurrent first-clients elect a single spawner instead of racing to
23+// re-exec multiple daemons.
24+const spawnLockName = "spawn.lock"
25+
26+// staleLockAge bounds how long a spawn lock may persist before it is treated
27+// as abandoned (spawner crashed mid-flight) and forcibly taken over.
28+const staleLockAge = 10 * time.Second
29+
30+// dialFunc dials the daemon socket in dir. Injected so tests can drive the
31+// spawn/backoff logic without a real socket.
32+type dialFunc func(dir string) (ports.Transport, error)
33+
34+// spawnFunc launches a detached daemon process. Injected so tests never
35+// re-exec a real binary.
36+type spawnFunc func() error
37+
38+// backoffConfig parameterises retry-dial timing so tests can shrink it.
39+type backoffConfig struct {
40+ initial time.Duration
41+ max time.Duration
42+ total time.Duration
43+}
44+
45+// defaultBackoff is the production retry schedule: 10ms, 20ms, 40ms... capped
46+// at 500ms, for roughly 5s total.
47+var defaultBackoff = backoffConfig{
48+ initial: 10 * time.Millisecond,
49+ max: 500 * time.Millisecond,
50+ total: 5 * time.Second,
51+}
52+
53+// ensureDaemon returns a transport to a running daemon, spawning one if
54+// necessary. It first tries a plain dial; on failure it elects a single
55+// spawner via an mkdir lock (losers simply wait), then retry-dials with
56+// backoff until the socket is live or the budget expires.
57+func ensureDaemon(ctx context.Context, dir string, dial dialFunc, spawn spawnFunc, cfg backoffConfig) (ports.Transport, error) {
58+ if t, err := dial(dir); err == nil {
59+ return t, nil
60+ }
61+
62+ release, acquired, err := acquireSpawnLock(dir)
63+ if err != nil {
64+ return nil, fmt.Errorf("vev: acquiring spawn lock: %w", err)
65+ }
66+ if acquired {
67+ // We won the election: spawn, and hold the lock until the socket is
68+ // dialable (or spawn fails) so late-arriving clients keep waiting
69+ // rather than spawning a second daemon.
70+ defer release()
71+ if err := spawn(); err != nil {
72+ return nil, fmt.Errorf("vev: spawning daemon: %w", err)
73+ }
74+ }
75+
76+ return retryDial(ctx, dir, dial, cfg)
77+}
78+
79+// acquireSpawnLock attempts to create the spawn-lock directory. It returns
80+// acquired=true with a release closure when it wins, acquired=false (with a
81+// no-op release) when another process holds a fresh lock, and takes over a
82+// lock older than staleLockAge (crash resilience).
83+func acquireSpawnLock(dir string) (release func(), acquired bool, err error) {
84+ if err := os.MkdirAll(dir, 0o700); err != nil {
85+ return nil, false, err
86+ }
87+ path := filepath.Join(dir, spawnLockName)
88+
89+ if mkErr := os.Mkdir(path, 0o700); mkErr == nil {
90+ return func() { _ = os.Remove(path) }, true, nil
91+ } else if !errors.Is(mkErr, os.ErrExist) {
92+ return nil, false, mkErr
93+ }
94+
95+ // Lock exists: take it over only if it looks abandoned.
96+ if info, statErr := os.Stat(path); statErr == nil && time.Since(info.ModTime()) > staleLockAge {
97+ _ = os.Remove(path)
98+ if mkErr := os.Mkdir(path, 0o700); mkErr == nil {
99+ return func() { _ = os.Remove(path) }, true, nil
100+ }
101+ }
102+
103+ // A live peer is spawning; wait for it via retry-dial.
104+ return func() {}, false, nil
105+}
106+
107+// retryDial dials repeatedly with exponential backoff until the daemon
108+// answers, the context is cancelled, or the total budget is exhausted.
109+func retryDial(ctx context.Context, dir string, dial dialFunc, cfg backoffConfig) (ports.Transport, error) {
110+ deadline := time.Now().Add(cfg.total)
111+ backoff := cfg.initial
112+ for {
113+ if t, err := dial(dir); err == nil {
114+ return t, nil
115+ }
116+ if !time.Now().Before(deadline) {
117+ return nil, ErrDaemonUnreachable
118+ }
119+ select {
120+ case <-ctx.Done():
121+ return nil, ctx.Err()
122+ case <-time.After(backoff):
123+ }
124+ if backoff *= 2; backoff > cfg.max {
125+ backoff = cfg.max
126+ }
127+ }
128+}
129+
130+// realDial is the production dialer.
131+func realDial(dir string) (ports.Transport, error) {
132+ return ipc.Dial(dir)
133+}
134+
135+// realSpawn re-execs this binary as a detached daemon: a new session
136+// (Setsid) with stdio bound to /dev/null so it survives the client's exit
137+// and never writes to the client's terminal. It logs via slog to the shared
138+// log file instead.
139+func realSpawn() error {
140+ devNull, err := os.OpenFile(os.DevNull, os.O_RDWR, 0)
141+ if err != nil {
142+ return fmt.Errorf("opening %s: %w", os.DevNull, err)
143+ }
144+ defer devNull.Close()
145+
146+ cmd := exec.Command("/proc/self/exe", "--daemon")
147+ cmd.Stdin = devNull
148+ cmd.Stdout = devNull
149+ cmd.Stderr = devNull
150+ cmd.SysProcAttr = &syscall.SysProcAttr{Setsid: true}
151+ if err := cmd.Start(); err != nil {
152+ return err
153+ }
154+ // Detach: we neither wait for nor signal the daemon after launch.
155+ return cmd.Process.Release()
156+}
A internal/app/spawn_test.go +189 -0
@@ -0,0 +1,189 @@
1+package app
2+
3+import (
4+ "context"
5+ "errors"
6+ "os"
7+ "path/filepath"
8+ "sync"
9+ "sync/atomic"
10+ "testing"
11+ "time"
12+
13+ "github.com/bnema/vev/internal/ports"
14+ portsmocks "github.com/bnema/vev/internal/ports/mocks"
15+)
16+
17+var errDialFailed = errors.New("dial failed")
18+
19+// fastBackoff keeps the retry loop snappy for tests.
20+var fastBackoff = backoffConfig{initial: time.Millisecond, max: 5 * time.Millisecond, total: 100 * time.Millisecond}
21+
22+func TestAcquireSpawnLockSingleWinnerUnderRace(t *testing.T) {
23+ dir := t.TempDir()
24+ const racers = 16
25+
26+ var wins atomic.Int32
27+ var wg sync.WaitGroup
28+ start := make(chan struct{})
29+ releases := make(chan func(), racers)
30+
31+ for i := 0; i < racers; i++ {
32+ wg.Add(1)
33+ go func() {
34+ defer wg.Done()
35+ <-start
36+ release, acquired, err := acquireSpawnLock(dir)
37+ if err != nil {
38+ t.Errorf("acquireSpawnLock: %v", err)
39+ return
40+ }
41+ if acquired {
42+ wins.Add(1)
43+ releases <- release
44+ }
45+ }()
46+ }
47+
48+ close(start)
49+ wg.Wait()
50+ close(releases)
51+
52+ if got := wins.Load(); got != 1 {
53+ t.Fatalf("expected exactly one winner, got %d", got)
54+ }
55+ for release := range releases {
56+ release()
57+ }
58+ // After release, the lock directory is gone and can be re-acquired.
59+ release, acquired, err := acquireSpawnLock(dir)
60+ if err != nil || !acquired {
61+ t.Fatalf("re-acquire after release: acquired=%v err=%v", acquired, err)
62+ }
63+ release()
64+}
65+
66+func TestAcquireSpawnLockTakesOverStaleLock(t *testing.T) {
67+ dir := t.TempDir()
68+ lockPath := filepath.Join(dir, spawnLockName)
69+ if err := os.Mkdir(lockPath, 0o700); err != nil {
70+ t.Fatalf("seeding lock: %v", err)
71+ }
72+
73+ // A fresh lock must NOT be taken over.
74+ if _, acquired, err := acquireSpawnLock(dir); err != nil || acquired {
75+ t.Fatalf("fresh lock taken over: acquired=%v err=%v", acquired, err)
76+ }
77+
78+ // Backdate the lock beyond the stale threshold; now it is taken over.
79+ old := time.Now().Add(-2 * staleLockAge)
80+ if err := os.Chtimes(lockPath, old, old); err != nil {
81+ t.Fatalf("backdating lock: %v", err)
82+ }
83+ release, acquired, err := acquireSpawnLock(dir)
84+ if err != nil || !acquired {
85+ t.Fatalf("stale lock not taken over: acquired=%v err=%v", acquired, err)
86+ }
87+ release()
88+}
89+
90+func TestRetryDialSucceedsAfterTransientFailures(t *testing.T) {
91+ want := portsmocks.NewMockTransport(t)
92+ var calls atomic.Int32
93+ dial := func(string) (ports.Transport, error) {
94+ if calls.Add(1) <= 3 {
95+ return nil, errDialFailed
96+ }
97+ return want, nil
98+ }
99+
100+ got, err := retryDial(context.Background(), t.TempDir(), dial, fastBackoff)
101+ if err != nil {
102+ t.Fatalf("retryDial: %v", err)
103+ }
104+ if got != want {
105+ t.Fatalf("retryDial returned unexpected transport")
106+ }
107+ if calls.Load() != 4 {
108+ t.Fatalf("dial called %d times, want 4", calls.Load())
109+ }
110+}
111+
112+func TestRetryDialPermanentFailure(t *testing.T) {
113+ dial := func(string) (ports.Transport, error) { return nil, errDialFailed }
114+ _, err := retryDial(context.Background(), t.TempDir(), dial, fastBackoff)
115+ if !errors.Is(err, ErrDaemonUnreachable) {
116+ t.Fatalf("retryDial error = %v, want ErrDaemonUnreachable", err)
117+ }
118+}
119+
120+func TestRetryDialContextCancelled(t *testing.T) {
121+ ctx, cancel := context.WithCancel(context.Background())
122+ cancel()
123+ dial := func(string) (ports.Transport, error) { return nil, errDialFailed }
124+ _, err := retryDial(ctx, t.TempDir(), dial, fastBackoff)
125+ if !errors.Is(err, context.Canceled) {
126+ t.Fatalf("retryDial error = %v, want context.Canceled", err)
127+ }
128+}
129+
130+func TestEnsureDaemonSpawnsThenDials(t *testing.T) {
131+ dir := t.TempDir()
132+ want := portsmocks.NewMockTransport(t)
133+ var dialCalls, spawnCalls atomic.Int32
134+ dial := func(string) (ports.Transport, error) {
135+ // First (pre-spawn) dial fails; the post-spawn retry succeeds.
136+ if dialCalls.Add(1) == 1 {
137+ return nil, errDialFailed
138+ }
139+ return want, nil
140+ }
141+ spawn := func() error {
142+ spawnCalls.Add(1)
143+ return nil
144+ }
145+
146+ got, err := ensureDaemon(context.Background(), dir, dial, spawn, fastBackoff)
147+ if err != nil {
148+ t.Fatalf("ensureDaemon: %v", err)
149+ }
150+ if got != want {
151+ t.Fatalf("ensureDaemon returned unexpected transport")
152+ }
153+ if spawnCalls.Load() != 1 {
154+ t.Fatalf("spawn called %d times, want 1", spawnCalls.Load())
155+ }
156+ // The spawn lock must be released once ensureDaemon returns.
157+ if _, err := os.Stat(filepath.Join(dir, spawnLockName)); !os.IsNotExist(err) {
158+ t.Fatalf("spawn lock not released: stat err=%v", err)
159+ }
160+}
161+
162+func TestEnsureDaemonReturnsExistingDaemon(t *testing.T) {
163+ want := portsmocks.NewMockTransport(t)
164+ var spawnCalls atomic.Int32
165+ dial := func(string) (ports.Transport, error) { return want, nil }
166+ spawn := func() error { spawnCalls.Add(1); return nil }
167+
168+ got, err := ensureDaemon(context.Background(), t.TempDir(), dial, spawn, fastBackoff)
169+ if err != nil {
170+ t.Fatalf("ensureDaemon: %v", err)
171+ }
172+ if got != want {
173+ t.Fatalf("ensureDaemon returned unexpected transport")
174+ }
175+ if spawnCalls.Load() != 0 {
176+ t.Fatalf("spawn should not be called when daemon already up, got %d", spawnCalls.Load())
177+ }
178+}
179+
180+func TestEnsureDaemonSpawnFailure(t *testing.T) {
181+ spawnErr := errors.New("exec boom")
182+ dial := func(string) (ports.Transport, error) { return nil, errDialFailed }
183+ spawn := func() error { return spawnErr }
184+
185+ _, err := ensureDaemon(context.Background(), t.TempDir(), dial, spawn, fastBackoff)
186+ if !errors.Is(err, spawnErr) {
187+ t.Fatalf("ensureDaemon error = %v, want wrapped %v", err, spawnErr)
188+ }
189+}
@@ -0,0 +1,189 @@
1+package app
2+
3+import (
4+ "context"
5+ "errors"
6+ "os"
7+ "path/filepath"
8+ "sync"
9+ "sync/atomic"
10+ "testing"
11+ "time"
12+
13+ "github.com/bnema/vev/internal/ports"
14+ portsmocks "github.com/bnema/vev/internal/ports/mocks"
15+)
16+
17+var errDialFailed = errors.New("dial failed")
18+
19+// fastBackoff keeps the retry loop snappy for tests.
20+var fastBackoff = backoffConfig{initial: time.Millisecond, max: 5 * time.Millisecond, total: 100 * time.Millisecond}
21+
22+func TestAcquireSpawnLockSingleWinnerUnderRace(t *testing.T) {
23+ dir := t.TempDir()
24+ const racers = 16
25+
26+ var wins atomic.Int32
27+ var wg sync.WaitGroup
28+ start := make(chan struct{})
29+ releases := make(chan func(), racers)
30+
31+ for i := 0; i < racers; i++ {
32+ wg.Add(1)
33+ go func() {
34+ defer wg.Done()
35+ <-start
36+ release, acquired, err := acquireSpawnLock(dir)
37+ if err != nil {
38+ t.Errorf("acquireSpawnLock: %v", err)
39+ return
40+ }
41+ if acquired {
42+ wins.Add(1)
43+ releases <- release
44+ }
45+ }()
46+ }
47+
48+ close(start)
49+ wg.Wait()
50+ close(releases)
51+
52+ if got := wins.Load(); got != 1 {
53+ t.Fatalf("expected exactly one winner, got %d", got)
54+ }
55+ for release := range releases {
56+ release()
57+ }
58+ // After release, the lock directory is gone and can be re-acquired.
59+ release, acquired, err := acquireSpawnLock(dir)
60+ if err != nil || !acquired {
61+ t.Fatalf("re-acquire after release: acquired=%v err=%v", acquired, err)
62+ }
63+ release()
64+}
65+
66+func TestAcquireSpawnLockTakesOverStaleLock(t *testing.T) {
67+ dir := t.TempDir()
68+ lockPath := filepath.Join(dir, spawnLockName)
69+ if err := os.Mkdir(lockPath, 0o700); err != nil {
70+ t.Fatalf("seeding lock: %v", err)
71+ }
72+
73+ // A fresh lock must NOT be taken over.
74+ if _, acquired, err := acquireSpawnLock(dir); err != nil || acquired {
75+ t.Fatalf("fresh lock taken over: acquired=%v err=%v", acquired, err)
76+ }
77+
78+ // Backdate the lock beyond the stale threshold; now it is taken over.
79+ old := time.Now().Add(-2 * staleLockAge)
80+ if err := os.Chtimes(lockPath, old, old); err != nil {
81+ t.Fatalf("backdating lock: %v", err)
82+ }
83+ release, acquired, err := acquireSpawnLock(dir)
84+ if err != nil || !acquired {
85+ t.Fatalf("stale lock not taken over: acquired=%v err=%v", acquired, err)
86+ }
87+ release()
88+}
89+
90+func TestRetryDialSucceedsAfterTransientFailures(t *testing.T) {
91+ want := portsmocks.NewMockTransport(t)
92+ var calls atomic.Int32
93+ dial := func(string) (ports.Transport, error) {
94+ if calls.Add(1) <= 3 {
95+ return nil, errDialFailed
96+ }
97+ return want, nil
98+ }
99+
100+ got, err := retryDial(context.Background(), t.TempDir(), dial, fastBackoff)
101+ if err != nil {
102+ t.Fatalf("retryDial: %v", err)
103+ }
104+ if got != want {
105+ t.Fatalf("retryDial returned unexpected transport")
106+ }
107+ if calls.Load() != 4 {
108+ t.Fatalf("dial called %d times, want 4", calls.Load())
109+ }
110+}
111+
112+func TestRetryDialPermanentFailure(t *testing.T) {
113+ dial := func(string) (ports.Transport, error) { return nil, errDialFailed }
114+ _, err := retryDial(context.Background(), t.TempDir(), dial, fastBackoff)
115+ if !errors.Is(err, ErrDaemonUnreachable) {
116+ t.Fatalf("retryDial error = %v, want ErrDaemonUnreachable", err)
117+ }
118+}
119+
120+func TestRetryDialContextCancelled(t *testing.T) {
121+ ctx, cancel := context.WithCancel(context.Background())
122+ cancel()
123+ dial := func(string) (ports.Transport, error) { return nil, errDialFailed }
124+ _, err := retryDial(ctx, t.TempDir(), dial, fastBackoff)
125+ if !errors.Is(err, context.Canceled) {
126+ t.Fatalf("retryDial error = %v, want context.Canceled", err)
127+ }
128+}
129+
130+func TestEnsureDaemonSpawnsThenDials(t *testing.T) {
131+ dir := t.TempDir()
132+ want := portsmocks.NewMockTransport(t)
133+ var dialCalls, spawnCalls atomic.Int32
134+ dial := func(string) (ports.Transport, error) {
135+ // First (pre-spawn) dial fails; the post-spawn retry succeeds.
136+ if dialCalls.Add(1) == 1 {
137+ return nil, errDialFailed
138+ }
139+ return want, nil
140+ }
141+ spawn := func() error {
142+ spawnCalls.Add(1)
143+ return nil
144+ }
145+
146+ got, err := ensureDaemon(context.Background(), dir, dial, spawn, fastBackoff)
147+ if err != nil {
148+ t.Fatalf("ensureDaemon: %v", err)
149+ }
150+ if got != want {
151+ t.Fatalf("ensureDaemon returned unexpected transport")
152+ }
153+ if spawnCalls.Load() != 1 {
154+ t.Fatalf("spawn called %d times, want 1", spawnCalls.Load())
155+ }
156+ // The spawn lock must be released once ensureDaemon returns.
157+ if _, err := os.Stat(filepath.Join(dir, spawnLockName)); !os.IsNotExist(err) {
158+ t.Fatalf("spawn lock not released: stat err=%v", err)
159+ }
160+}
161+
162+func TestEnsureDaemonReturnsExistingDaemon(t *testing.T) {
163+ want := portsmocks.NewMockTransport(t)
164+ var spawnCalls atomic.Int32
165+ dial := func(string) (ports.Transport, error) { return want, nil }
166+ spawn := func() error { spawnCalls.Add(1); return nil }
167+
168+ got, err := ensureDaemon(context.Background(), t.TempDir(), dial, spawn, fastBackoff)
169+ if err != nil {
170+ t.Fatalf("ensureDaemon: %v", err)
171+ }
172+ if got != want {
173+ t.Fatalf("ensureDaemon returned unexpected transport")
174+ }
175+ if spawnCalls.Load() != 0 {
176+ t.Fatalf("spawn should not be called when daemon already up, got %d", spawnCalls.Load())
177+ }
178+}
179+
180+func TestEnsureDaemonSpawnFailure(t *testing.T) {
181+ spawnErr := errors.New("exec boom")
182+ dial := func(string) (ports.Transport, error) { return nil, errDialFailed }
183+ spawn := func() error { return spawnErr }
184+
185+ _, err := ensureDaemon(context.Background(), t.TempDir(), dial, spawn, fastBackoff)
186+ if !errors.Is(err, spawnErr) {
187+ t.Fatalf("ensureDaemon error = %v, want wrapped %v", err, spawnErr)
188+ }
189+}
A internal/usecase/client/client.go +331 -0
@@ -0,0 +1,331 @@
1+// Package client implements vev's thin attach client: it speaks the framed
2+// IPC protocol to a daemon, pumping terminal input, resize events, and PTY
3+// output between the controlling terminal and the session.
4+//
5+// The package depends only on ports (transport, terminal) plus the ipc
6+// message codecs, never on a concrete transport or terminal implementation;
7+// the app layer wires those in.
8+package client
9+
10+import (
11+ "context"
12+ "errors"
13+ "fmt"
14+ "io"
15+ "log/slog"
16+ "os"
17+
18+ "github.com/bnema/vev/internal/adapters/ipc"
19+ "github.com/bnema/vev/internal/domain"
20+ "github.com/bnema/vev/internal/ports"
21+)
22+
23+// ProtocolError is a session- or protocol-level failure reported by the
24+// daemon before attach could proceed (a Hello rejected with an ErrorMsg).
25+// The app layer prints it to the user.
26+type ProtocolError struct {
27+ Code uint16
28+ Text string
29+}
30+
31+func (e *ProtocolError) Error() string {
32+ if e.Text == "" {
33+ return fmt.Sprintf("vev: daemon rejected attach (code %d)", e.Code)
34+ }
35+ return fmt.Sprintf("vev: %s", e.Text)
36+}
37+
38+// DetachedError reports that the daemon detached this client for a reason
39+// other than an ordinary user-initiated detach.
40+type DetachedError struct {
41+ Reason uint8
42+ Text string
43+}
44+
45+func (e *DetachedError) Error() string { return "vev: " + e.Text }
46+
47+// milestones tracks lifecycle progress so a failure can report which stages
48+// were never reached — the key diagnostic when an attach dies early.
49+type milestones struct {
50+ dialed bool
51+ helloSent bool
52+ welcomed bool
53+ rawEntered bool
54+ firstOutput bool
55+ detached bool
56+}
57+
58+func (m milestones) missing() []string {
59+ var out []string
60+ if !m.dialed {
61+ out = append(out, "dialed")
62+ }
63+ if !m.helloSent {
64+ out = append(out, "hello_sent")
65+ }
66+ if !m.welcomed {
67+ out = append(out, "welcomed")
68+ }
69+ if !m.rawEntered {
70+ out = append(out, "raw_entered")
71+ }
72+ if !m.firstOutput {
73+ out = append(out, "first_output")
74+ }
75+ if !m.detached {
76+ out = append(out, "detached")
77+ }
78+ return out
79+}
80+
81+// stdinBufSize is the read buffer for the terminal input pump.
82+const stdinBufSize = 4096
83+
84+// sendQueueDepth bounds the buffered frames awaiting the single sender
85+// goroutine, decoupling the input/resize pumps from transport back-pressure.
86+const sendQueueDepth = 64
87+
88+// Attach connects an already-dialed transport to the controlling terminal
89+// and runs the attach loop until the session detaches, the daemon
90+// disappears, or the context is cancelled.
91+//
92+// It owns transport for its lifetime and closes it before returning. The
93+// terminal is put into raw mode after a successful Welcome and restored on
94+// every exit path — normal return, error, or panic.
95+func Attach(ctx context.Context, transport ports.Transport, term ports.Terminal, intent uint8, name string) (retErr error) {
96+ log := slog.Default().With("component", "client")
97+ ms := milestones{dialed: true} // transport arrives already connected
98+
99+ // Runs last: a single structured diagnostic on failure, to the log file
100+ // (never the terminal). Console output while raw is forbidden.
101+ defer func() {
102+ if retErr != nil {
103+ log.Error("attach ended with error", "err", retErr, "missing_milestones", ms.missing())
104+ }
105+ }()
106+
107+ // Attach owns the transport: close it on every exit path.
108+ defer func() { _ = transport.Close() }()
109+
110+ // 1. Handshake: send Hello with our size and TERM.
111+ size, err := term.Size()
112+ if err != nil {
113+ return fmt.Errorf("vev: reading terminal size: %w", err)
114+ }
115+ hello := ipc.Hello{
116+ Version: ports.ProtocolVersion,
117+ Intent: intent,
118+ Name: name,
119+ Size: size,
120+ TermEnv: os.Getenv("TERM"),
121+ }
122+ if err := transport.Send(ports.Frame{Type: ports.MsgHello, Payload: ipc.MarshalHello(hello)}); err != nil {
123+ return fmt.Errorf("vev: sending hello: %w", err)
124+ }
125+ ms.helloSent = true
126+
127+ // 2. Await Welcome or a typed rejection.
128+ reply, err := transport.Recv()
129+ if err != nil {
130+ return fmt.Errorf("vev: awaiting welcome: %w", err)
131+ }
132+ switch reply.Type {
133+ case ports.MsgWelcome:
134+ ms.welcomed = true
135+ case ports.MsgError:
136+ em, derr := ipc.UnmarshalErrorMsg(reply.Payload)
137+ if derr != nil {
138+ return fmt.Errorf("vev: decoding error reply: %w", derr)
139+ }
140+ return &ProtocolError{Code: em.Code, Text: em.Text}
141+ default:
142+ return fmt.Errorf("vev: unexpected reply type %d before welcome", reply.Type)
143+ }
144+
145+ // 3. Enter raw mode; restore on every subsequent exit path.
146+ restore, err := term.EnterRaw()
147+ if err != nil {
148+ return fmt.Errorf("vev: entering raw mode: %w", err)
149+ }
150+ ms.rawEntered = true
151+ defer func() {
152+ if rerr := restore(); rerr != nil && retErr == nil {
153+ retErr = fmt.Errorf("vev: restoring terminal: %w", rerr)
154+ }
155+ }()
156+
157+ // 4. Derive a cancellable context so the pumps always unwind when the
158+ // loop returns. cancel runs first (LIFO): it signals the pumps; the
159+ // deferred Close above then unblocks any pump parked in a syscall.
160+ loopCtx, cancel := context.WithCancel(ctx)
161+ defer cancel()
162+
163+ sendCh := make(chan ports.Frame, sendQueueDepth)
164+ sendErrCh := make(chan error, 1)
165+
166+ go runSender(loopCtx, cancel, transport, sendCh, sendErrCh)
167+ go runStdin(loopCtx, cancel, term.In(), sendCh)
168+ go runResize(loopCtx, term.ResizeEvents(), sendCh)
169+
170+ // 5. Output/main loop: the only goroutine that touches the terminal.
171+ recvCh := make(chan recvResult, 1)
172+ go runRecv(loopCtx, transport, recvCh)
173+
174+ for {
175+ select {
176+ case <-loopCtx.Done():
177+ // A pump initiated shutdown (stdin EOF/detach) or the parent
178+ // context was cancelled. A queued sender error takes priority.
179+ select {
180+ case serr := <-sendErrCh:
181+ return fmt.Errorf("vev: sending to daemon: %w", serr)
182+ default:
183+ return nil
184+ }
185+ case r := <-recvCh:
186+ if r.err != nil {
187+ if errors.Is(r.err, io.EOF) {
188+ return fmt.Errorf("vev: daemon vanished (missing: %v): %w", ms.missing(), r.err)
189+ }
190+ return fmt.Errorf("vev: receiving from daemon: %w", r.err)
191+ }
192+ switch r.frame.Type {
193+ case ports.MsgOutput:
194+ o, derr := ipc.UnmarshalOutput(r.frame.Payload)
195+ if derr != nil {
196+ return fmt.Errorf("vev: decoding output: %w", derr)
197+ }
198+ if _, werr := term.Out().Write(o.Data); werr != nil {
199+ return fmt.Errorf("vev: writing terminal output: %w", werr)
200+ }
201+ if ferr := term.Flush(); ferr != nil {
202+ return fmt.Errorf("vev: flushing terminal: %w", ferr)
203+ }
204+ ms.firstOutput = true
205+ case ports.MsgDetached:
206+ d, derr := ipc.UnmarshalDetached(r.frame.Payload)
207+ if derr != nil {
208+ return fmt.Errorf("vev: decoding detached: %w", derr)
209+ }
210+ ms.detached = true
211+ return detachedResult(d.Reason)
212+ case ports.MsgPong:
213+ // Liveness reply; nothing to do.
214+ default:
215+ // Unknown/out-of-band message types are ignored so a newer
216+ // daemon can add server->client messages without breaking us.
217+ }
218+ }
219+ }
220+}
221+
222+// recvResult carries one framed message (or a read error) from the receive
223+// pump to the main loop.
224+type recvResult struct {
225+ frame ports.Frame
226+ err error
227+}
228+
229+// runRecv reads frames until an error, forwarding each to the main loop.
230+// It exits on the first error or when the loop context is cancelled.
231+func runRecv(ctx context.Context, transport ports.Transport, out chan<- recvResult) {
232+ for {
233+ f, err := transport.Recv()
234+ select {
235+ case out <- recvResult{frame: f, err: err}:
236+ case <-ctx.Done():
237+ return
238+ }
239+ if err != nil {
240+ return
241+ }
242+ }
243+}
244+
245+// runSender is the sole caller of transport.Send: serialising the input and
246+// resize pumps through one goroutine keeps the transport's single-writer
247+// contract intact. A send failure cancels the loop and is surfaced once.
248+func runSender(ctx context.Context, cancel context.CancelFunc, transport ports.Transport, in <-chan ports.Frame, errCh chan<- error) {
249+ for {
250+ select {
251+ case f := <-in:
252+ if err := transport.Send(f); err != nil {
253+ select {
254+ case errCh <- err:
255+ default:
256+ }
257+ cancel()
258+ return
259+ }
260+ case <-ctx.Done():
261+ return
262+ }
263+ }
264+}
265+
266+// runStdin pumps terminal input to the daemon as Input frames. A read error
267+// or EOF is treated as a detach: it best-effort sends a Detach frame and
268+// cancels the loop.
269+func runStdin(ctx context.Context, cancel context.CancelFunc, in io.Reader, out chan<- ports.Frame) {
270+ buf := make([]byte, stdinBufSize)
271+ for {
272+ n, rerr := in.Read(buf)
273+ if n > 0 {
274+ data := make([]byte, n)
275+ copy(data, buf[:n])
276+ frame := ports.Frame{Type: ports.MsgInput, Payload: ipc.MarshalInput(ipc.Input{Data: data})}
277+ select {
278+ case out <- frame:
279+ case <-ctx.Done():
280+ return
281+ }
282+ }
283+ if rerr != nil {
284+ // Best-effort detach notification, then unwind.
285+ select {
286+ case out <- ports.Frame{Type: ports.MsgDetach, Payload: ipc.MarshalDetach(ipc.Detach{})}:
287+ case <-ctx.Done():
288+ }
289+ cancel()
290+ return
291+ }
292+ }
293+}
294+
295+// runResize forwards coalesced terminal resize events to the daemon. It
296+// tolerates an already-closed resize channel (which the terminal adapter
297+// hands back when restore ran before ResizeEvents was first called).
298+func runResize(ctx context.Context, events <-chan domain.Size, out chan<- ports.Frame) {
299+ for {
300+ select {
301+ case sz, ok := <-events:
302+ if !ok {
303+ return
304+ }
305+ frame := ports.Frame{Type: ports.MsgResize, Payload: ipc.MarshalResize(ipc.Resize{Size: sz})}
306+ select {
307+ case out <- frame:
308+ case <-ctx.Done():
309+ return
310+ }
311+ case <-ctx.Done():
312+ return
313+ }
314+ }
315+}
316+
317+// detachedResult maps a Detached reason to a return value: an ordinary
318+// user-initiated detach is a clean (nil) exit; anything else is surfaced so
319+// the user learns why the session went away.
320+func detachedResult(reason uint8) error {
321+ switch reason {
322+ case ipc.ReasonDetach:
323+ return nil
324+ case ipc.ReasonSessionKilled:
325+ return &DetachedError{Reason: reason, Text: "session was killed"}
326+ case ipc.ReasonServerShutdown:
327+ return &DetachedError{Reason: reason, Text: "daemon shut down"}
328+ default:
329+ return &DetachedError{Reason: reason, Text: "detached by daemon"}
330+ }
331+}
@@ -0,0 +1,331 @@
1+// Package client implements vev's thin attach client: it speaks the framed
2+// IPC protocol to a daemon, pumping terminal input, resize events, and PTY
3+// output between the controlling terminal and the session.
4+//
5+// The package depends only on ports (transport, terminal) plus the ipc
6+// message codecs, never on a concrete transport or terminal implementation;
7+// the app layer wires those in.
8+package client
9+
10+import (
11+ "context"
12+ "errors"
13+ "fmt"
14+ "io"
15+ "log/slog"
16+ "os"
17+
18+ "github.com/bnema/vev/internal/adapters/ipc"
19+ "github.com/bnema/vev/internal/domain"
20+ "github.com/bnema/vev/internal/ports"
21+)
22+
23+// ProtocolError is a session- or protocol-level failure reported by the
24+// daemon before attach could proceed (a Hello rejected with an ErrorMsg).
25+// The app layer prints it to the user.
26+type ProtocolError struct {
27+ Code uint16
28+ Text string
29+}
30+
31+func (e *ProtocolError) Error() string {
32+ if e.Text == "" {
33+ return fmt.Sprintf("vev: daemon rejected attach (code %d)", e.Code)
34+ }
35+ return fmt.Sprintf("vev: %s", e.Text)
36+}
37+
38+// DetachedError reports that the daemon detached this client for a reason
39+// other than an ordinary user-initiated detach.
40+type DetachedError struct {
41+ Reason uint8
42+ Text string
43+}
44+
45+func (e *DetachedError) Error() string { return "vev: " + e.Text }
46+
47+// milestones tracks lifecycle progress so a failure can report which stages
48+// were never reached — the key diagnostic when an attach dies early.
49+type milestones struct {
50+ dialed bool
51+ helloSent bool
52+ welcomed bool
53+ rawEntered bool
54+ firstOutput bool
55+ detached bool
56+}
57+
58+func (m milestones) missing() []string {
59+ var out []string
60+ if !m.dialed {
61+ out = append(out, "dialed")
62+ }
63+ if !m.helloSent {
64+ out = append(out, "hello_sent")
65+ }
66+ if !m.welcomed {
67+ out = append(out, "welcomed")
68+ }
69+ if !m.rawEntered {
70+ out = append(out, "raw_entered")
71+ }
72+ if !m.firstOutput {
73+ out = append(out, "first_output")
74+ }
75+ if !m.detached {
76+ out = append(out, "detached")
77+ }
78+ return out
79+}
80+
81+// stdinBufSize is the read buffer for the terminal input pump.
82+const stdinBufSize = 4096
83+
84+// sendQueueDepth bounds the buffered frames awaiting the single sender
85+// goroutine, decoupling the input/resize pumps from transport back-pressure.
86+const sendQueueDepth = 64
87+
88+// Attach connects an already-dialed transport to the controlling terminal
89+// and runs the attach loop until the session detaches, the daemon
90+// disappears, or the context is cancelled.
91+//
92+// It owns transport for its lifetime and closes it before returning. The
93+// terminal is put into raw mode after a successful Welcome and restored on
94+// every exit path — normal return, error, or panic.
95+func Attach(ctx context.Context, transport ports.Transport, term ports.Terminal, intent uint8, name string) (retErr error) {
96+ log := slog.Default().With("component", "client")
97+ ms := milestones{dialed: true} // transport arrives already connected
98+
99+ // Runs last: a single structured diagnostic on failure, to the log file
100+ // (never the terminal). Console output while raw is forbidden.
101+ defer func() {
102+ if retErr != nil {
103+ log.Error("attach ended with error", "err", retErr, "missing_milestones", ms.missing())
104+ }
105+ }()
106+
107+ // Attach owns the transport: close it on every exit path.
108+ defer func() { _ = transport.Close() }()
109+
110+ // 1. Handshake: send Hello with our size and TERM.
111+ size, err := term.Size()
112+ if err != nil {
113+ return fmt.Errorf("vev: reading terminal size: %w", err)
114+ }
115+ hello := ipc.Hello{
116+ Version: ports.ProtocolVersion,
117+ Intent: intent,
118+ Name: name,
119+ Size: size,
120+ TermEnv: os.Getenv("TERM"),
121+ }
122+ if err := transport.Send(ports.Frame{Type: ports.MsgHello, Payload: ipc.MarshalHello(hello)}); err != nil {
123+ return fmt.Errorf("vev: sending hello: %w", err)
124+ }
125+ ms.helloSent = true
126+
127+ // 2. Await Welcome or a typed rejection.
128+ reply, err := transport.Recv()
129+ if err != nil {
130+ return fmt.Errorf("vev: awaiting welcome: %w", err)
131+ }
132+ switch reply.Type {
133+ case ports.MsgWelcome:
134+ ms.welcomed = true
135+ case ports.MsgError:
136+ em, derr := ipc.UnmarshalErrorMsg(reply.Payload)
137+ if derr != nil {
138+ return fmt.Errorf("vev: decoding error reply: %w", derr)
139+ }
140+ return &ProtocolError{Code: em.Code, Text: em.Text}
141+ default:
142+ return fmt.Errorf("vev: unexpected reply type %d before welcome", reply.Type)
143+ }
144+
145+ // 3. Enter raw mode; restore on every subsequent exit path.
146+ restore, err := term.EnterRaw()
147+ if err != nil {
148+ return fmt.Errorf("vev: entering raw mode: %w", err)
149+ }
150+ ms.rawEntered = true
151+ defer func() {
152+ if rerr := restore(); rerr != nil && retErr == nil {
153+ retErr = fmt.Errorf("vev: restoring terminal: %w", rerr)
154+ }
155+ }()
156+
157+ // 4. Derive a cancellable context so the pumps always unwind when the
158+ // loop returns. cancel runs first (LIFO): it signals the pumps; the
159+ // deferred Close above then unblocks any pump parked in a syscall.
160+ loopCtx, cancel := context.WithCancel(ctx)
161+ defer cancel()
162+
163+ sendCh := make(chan ports.Frame, sendQueueDepth)
164+ sendErrCh := make(chan error, 1)
165+
166+ go runSender(loopCtx, cancel, transport, sendCh, sendErrCh)
167+ go runStdin(loopCtx, cancel, term.In(), sendCh)
168+ go runResize(loopCtx, term.ResizeEvents(), sendCh)
169+
170+ // 5. Output/main loop: the only goroutine that touches the terminal.
171+ recvCh := make(chan recvResult, 1)
172+ go runRecv(loopCtx, transport, recvCh)
173+
174+ for {
175+ select {
176+ case <-loopCtx.Done():
177+ // A pump initiated shutdown (stdin EOF/detach) or the parent
178+ // context was cancelled. A queued sender error takes priority.
179+ select {
180+ case serr := <-sendErrCh:
181+ return fmt.Errorf("vev: sending to daemon: %w", serr)
182+ default:
183+ return nil
184+ }
185+ case r := <-recvCh:
186+ if r.err != nil {
187+ if errors.Is(r.err, io.EOF) {
188+ return fmt.Errorf("vev: daemon vanished (missing: %v): %w", ms.missing(), r.err)
189+ }
190+ return fmt.Errorf("vev: receiving from daemon: %w", r.err)
191+ }
192+ switch r.frame.Type {
193+ case ports.MsgOutput:
194+ o, derr := ipc.UnmarshalOutput(r.frame.Payload)
195+ if derr != nil {
196+ return fmt.Errorf("vev: decoding output: %w", derr)
197+ }
198+ if _, werr := term.Out().Write(o.Data); werr != nil {
199+ return fmt.Errorf("vev: writing terminal output: %w", werr)
200+ }
201+ if ferr := term.Flush(); ferr != nil {
202+ return fmt.Errorf("vev: flushing terminal: %w", ferr)
203+ }
204+ ms.firstOutput = true
205+ case ports.MsgDetached:
206+ d, derr := ipc.UnmarshalDetached(r.frame.Payload)
207+ if derr != nil {
208+ return fmt.Errorf("vev: decoding detached: %w", derr)
209+ }
210+ ms.detached = true
211+ return detachedResult(d.Reason)
212+ case ports.MsgPong:
213+ // Liveness reply; nothing to do.
214+ default:
215+ // Unknown/out-of-band message types are ignored so a newer
216+ // daemon can add server->client messages without breaking us.
217+ }
218+ }
219+ }
220+}
221+
222+// recvResult carries one framed message (or a read error) from the receive
223+// pump to the main loop.
224+type recvResult struct {
225+ frame ports.Frame
226+ err error
227+}
228+
229+// runRecv reads frames until an error, forwarding each to the main loop.
230+// It exits on the first error or when the loop context is cancelled.
231+func runRecv(ctx context.Context, transport ports.Transport, out chan<- recvResult) {
232+ for {
233+ f, err := transport.Recv()
234+ select {
235+ case out <- recvResult{frame: f, err: err}:
236+ case <-ctx.Done():
237+ return
238+ }
239+ if err != nil {
240+ return
241+ }
242+ }
243+}
244+
245+// runSender is the sole caller of transport.Send: serialising the input and
246+// resize pumps through one goroutine keeps the transport's single-writer
247+// contract intact. A send failure cancels the loop and is surfaced once.
248+func runSender(ctx context.Context, cancel context.CancelFunc, transport ports.Transport, in <-chan ports.Frame, errCh chan<- error) {
249+ for {
250+ select {
251+ case f := <-in:
252+ if err := transport.Send(f); err != nil {
253+ select {
254+ case errCh <- err:
255+ default:
256+ }
257+ cancel()
258+ return
259+ }
260+ case <-ctx.Done():
261+ return
262+ }
263+ }
264+}
265+
266+// runStdin pumps terminal input to the daemon as Input frames. A read error
267+// or EOF is treated as a detach: it best-effort sends a Detach frame and
268+// cancels the loop.
269+func runStdin(ctx context.Context, cancel context.CancelFunc, in io.Reader, out chan<- ports.Frame) {
270+ buf := make([]byte, stdinBufSize)
271+ for {
272+ n, rerr := in.Read(buf)
273+ if n > 0 {
274+ data := make([]byte, n)
275+ copy(data, buf[:n])
276+ frame := ports.Frame{Type: ports.MsgInput, Payload: ipc.MarshalInput(ipc.Input{Data: data})}
277+ select {
278+ case out <- frame:
279+ case <-ctx.Done():
280+ return
281+ }
282+ }
283+ if rerr != nil {
284+ // Best-effort detach notification, then unwind.
285+ select {
286+ case out <- ports.Frame{Type: ports.MsgDetach, Payload: ipc.MarshalDetach(ipc.Detach{})}:
287+ case <-ctx.Done():
288+ }
289+ cancel()
290+ return
291+ }
292+ }
293+}
294+
295+// runResize forwards coalesced terminal resize events to the daemon. It
296+// tolerates an already-closed resize channel (which the terminal adapter
297+// hands back when restore ran before ResizeEvents was first called).
298+func runResize(ctx context.Context, events <-chan domain.Size, out chan<- ports.Frame) {
299+ for {
300+ select {
301+ case sz, ok := <-events:
302+ if !ok {
303+ return
304+ }
305+ frame := ports.Frame{Type: ports.MsgResize, Payload: ipc.MarshalResize(ipc.Resize{Size: sz})}
306+ select {
307+ case out <- frame:
308+ case <-ctx.Done():
309+ return
310+ }
311+ case <-ctx.Done():
312+ return
313+ }
314+ }
315+}
316+
317+// detachedResult maps a Detached reason to a return value: an ordinary
318+// user-initiated detach is a clean (nil) exit; anything else is surfaced so
319+// the user learns why the session went away.
320+func detachedResult(reason uint8) error {
321+ switch reason {
322+ case ipc.ReasonDetach:
323+ return nil
324+ case ipc.ReasonSessionKilled:
325+ return &DetachedError{Reason: reason, Text: "session was killed"}
326+ case ipc.ReasonServerShutdown:
327+ return &DetachedError{Reason: reason, Text: "daemon shut down"}
328+ default:
329+ return &DetachedError{Reason: reason, Text: "detached by daemon"}
330+ }
331+}
A internal/usecase/client/client_test.go +234 -0
@@ -0,0 +1,234 @@
1+package client_test
2+
3+import (
4+ "bytes"
5+ "context"
6+ "errors"
7+ "io"
8+ "log/slog"
9+ "sync/atomic"
10+ "testing"
11+ "time"
12+
13+ "github.com/stretchr/testify/mock"
14+ "github.com/stretchr/testify/require"
15+
16+ "github.com/bnema/vev/internal/adapters/ipc"
17+ "github.com/bnema/vev/internal/domain"
18+ "github.com/bnema/vev/internal/ports"
19+ portsmocks "github.com/bnema/vev/internal/ports/mocks"
20+ "github.com/bnema/vev/internal/usecase/client"
21+)
22+
23+func init() {
24+ // Keep client diagnostics out of the test runner's stderr.
25+ slog.SetDefault(slog.New(slog.NewTextHandler(io.Discard, nil)))
26+}
27+
28+// recvItem is one scripted Recv result.
29+type recvItem struct {
30+ f ports.Frame
31+ err error
32+}
33+
34+// scriptRecv wires a MockTransport's Recv to yield the given items in order,
35+// then block forever (simulating a live but idle connection). It returns a
36+// cleanup that unblocks the parked reader.
37+func scriptRecv(tr *portsmocks.MockTransport, items ...recvItem) func() {
38+ ch := make(chan recvItem, len(items))
39+ for _, it := range items {
40+ ch <- it
41+ }
42+ done := make(chan struct{})
43+ tr.EXPECT().Recv().RunAndReturn(func() (ports.Frame, error) {
44+ select {
45+ case it := <-ch:
46+ return it.f, it.err
47+ case <-done:
48+ return ports.Frame{}, io.EOF
49+ }
50+ }).Maybe()
51+ return func() { close(done) }
52+}
53+
54+func frameOf(t ports.MsgType, payload []byte) ports.Frame {
55+ return ports.Frame{Type: t, Payload: payload}
56+}
57+
58+// blockingReader blocks on Read until closed, then returns EOF. Stands in
59+// for a terminal's stdin that produces nothing during the test.
60+type blockingReader struct{ ch chan struct{} }
61+
62+func newBlockingReader() *blockingReader { return &blockingReader{ch: make(chan struct{})} }
63+
64+func (b *blockingReader) Read(p []byte) (int, error) {
65+ <-b.ch
66+ return 0, io.EOF
67+}
68+func (b *blockingReader) unblock() { close(b.ch) }
69+
70+func newHappyTerminal(t *testing.T, out *bytes.Buffer, restoreCount *atomic.Int32, resizeCh chan domain.Size) (*portsmocks.MockTerminal, *blockingReader) {
71+ tm := portsmocks.NewMockTerminal(t)
72+ tm.EXPECT().Size().Return(domain.Size{Cols: 80, Rows: 24}, nil).Once()
73+ tm.EXPECT().EnterRaw().Return(func() error {
74+ restoreCount.Add(1)
75+ return nil
76+ }, nil).Once()
77+ in := newBlockingReader()
78+ tm.EXPECT().In().Return(in).Maybe()
79+ tm.EXPECT().Out().Return(out).Maybe()
80+ tm.EXPECT().Flush().Return(nil).Maybe()
81+ tm.EXPECT().ResizeEvents().Return(resizeCh).Maybe()
82+ return tm, in
83+}
84+
85+func isType(typ ports.MsgType) any {
86+ return mock.MatchedBy(func(f ports.Frame) bool { return f.Type == typ })
87+}
88+
89+func TestAttachHappyPath(t *testing.T) {
90+ var out bytes.Buffer
91+ var restoreCount atomic.Int32
92+ resizeCh := make(chan domain.Size)
93+ tm, in := newHappyTerminal(t, &out, &restoreCount, resizeCh)
94+ defer in.unblock()
95+
96+ tr := portsmocks.NewMockTransport(t)
97+ tr.EXPECT().Send(isType(ports.MsgHello)).Return(nil).Once()
98+ unblock := scriptRecv(tr,
99+ recvItem{f: frameOf(ports.MsgWelcome, ipc.MarshalWelcome(ipc.Welcome{SessionID: "s1", SessionName: "main"}))},
100+ recvItem{f: frameOf(ports.MsgOutput, ipc.MarshalOutput(ipc.Output{Data: []byte("hello world")}))},
101+ recvItem{f: frameOf(ports.MsgDetached, ipc.MarshalDetached(ipc.Detached{Reason: ipc.ReasonDetach}))},
102+ )
103+ defer unblock()
104+ tr.EXPECT().Close().Return(nil).Once()
105+
106+ err := client.Attach(context.Background(), tr, tm, ipc.IntentEphemeral, "")
107+ require.NoError(t, err)
108+ require.Equal(t, "hello world", out.String())
109+ require.Equal(t, int32(1), restoreCount.Load(), "restore must run exactly once")
110+}
111+
112+func TestAttachVersionMismatch(t *testing.T) {
113+ tm := portsmocks.NewMockTerminal(t)
114+ tm.EXPECT().Size().Return(domain.Size{Cols: 80, Rows: 24}, nil).Once()
115+ // EnterRaw must NOT be called on the error-before-welcome path.
116+
117+ tr := portsmocks.NewMockTransport(t)
118+ tr.EXPECT().Send(isType(ports.MsgHello)).Return(nil).Once()
119+ tr.EXPECT().Recv().Return(
120+ frameOf(ports.MsgError, ipc.MarshalErrorMsg(ipc.ErrorMsg{Code: ipc.ErrVersionMismatch, Text: "version mismatch"})),
121+ nil,
122+ ).Once()
123+ tr.EXPECT().Close().Return(nil).Once()
124+
125+ err := client.Attach(context.Background(), tr, tm, ipc.IntentAttach, "main")
126+ require.Error(t, err)
127+ var pe *client.ProtocolError
128+ require.True(t, errors.As(err, &pe), "want *client.ProtocolError, got %T", err)
129+ require.Equal(t, ipc.ErrVersionMismatch, pe.Code)
130+}
131+
132+func TestAttachRestoredOnRecvErrorMidStream(t *testing.T) {
133+ var out bytes.Buffer
134+ var restoreCount atomic.Int32
135+ resizeCh := make(chan domain.Size)
136+ tm, in := newHappyTerminal(t, &out, &restoreCount, resizeCh)
137+ defer in.unblock()
138+
139+ tr := portsmocks.NewMockTransport(t)
140+ tr.EXPECT().Send(isType(ports.MsgHello)).Return(nil).Once()
141+ boom := errors.New("connection reset")
142+ unblock := scriptRecv(tr,
143+ recvItem{f: frameOf(ports.MsgWelcome, ipc.MarshalWelcome(ipc.Welcome{SessionID: "s1"}))},
144+ recvItem{err: boom},
145+ )
146+ defer unblock()
147+ tr.EXPECT().Close().Return(nil).Once()
148+
149+ err := client.Attach(context.Background(), tr, tm, ipc.IntentEphemeral, "")
150+ require.Error(t, err)
151+ require.Equal(t, int32(1), restoreCount.Load(), "restore must run after mid-stream Recv error")
152+}
153+
154+func TestAttachDaemonVanishedOnEOF(t *testing.T) {
155+ var out bytes.Buffer
156+ var restoreCount atomic.Int32
157+ resizeCh := make(chan domain.Size)
158+ tm, in := newHappyTerminal(t, &out, &restoreCount, resizeCh)
159+ defer in.unblock()
160+
161+ tr := portsmocks.NewMockTransport(t)
162+ tr.EXPECT().Send(isType(ports.MsgHello)).Return(nil).Once()
163+ unblock := scriptRecv(tr,
164+ recvItem{f: frameOf(ports.MsgWelcome, ipc.MarshalWelcome(ipc.Welcome{SessionID: "s1"}))},
165+ recvItem{err: io.EOF},
166+ )
167+ defer unblock()
168+ tr.EXPECT().Close().Return(nil).Once()
169+
170+ err := client.Attach(context.Background(), tr, tm, ipc.IntentEphemeral, "")
171+ require.Error(t, err)
172+ require.ErrorIs(t, err, io.EOF)
173+ require.Equal(t, int32(1), restoreCount.Load())
174+}
175+
176+func TestAttachForwardsResize(t *testing.T) {
177+ var out bytes.Buffer
178+ var restoreCount atomic.Int32
179+ resizeCh := make(chan domain.Size)
180+ tm, in := newHappyTerminal(t, &out, &restoreCount, resizeCh)
181+ defer in.unblock()
182+
183+ detachAfterResize := make(chan struct{})
184+
185+ tr := portsmocks.NewMockTransport(t)
186+ tr.EXPECT().Send(isType(ports.MsgHello)).Return(nil).Once()
187+ // The resize frame is forwarded via the sender goroutine.
188+ gotResize := make(chan ipc.Resize, 1)
189+ tr.EXPECT().Send(isType(ports.MsgResize)).RunAndReturn(func(f ports.Frame) error {
190+ r, _ := ipc.UnmarshalResize(f.Payload)
191+ gotResize <- r
192+ close(detachAfterResize)
193+ return nil
194+ }).Once()
195+
196+ welcome := frameOf(ports.MsgWelcome, ipc.MarshalWelcome(ipc.Welcome{SessionID: "s1"}))
197+ detached := frameOf(ports.MsgDetached, ipc.MarshalDetached(ipc.Detached{Reason: ipc.ReasonDetach}))
198+ recvCh := make(chan recvItem, 2)
199+ recvCh <- recvItem{f: welcome}
200+ done := make(chan struct{})
201+ tr.EXPECT().Recv().RunAndReturn(func() (ports.Frame, error) {
202+ select {
203+ case it := <-recvCh:
204+ return it.f, it.err
205+ case <-detachAfterResize:
206+ // Deliver the detach once the resize has been observed.
207+ select {
208+ case <-done:
209+ return ports.Frame{}, io.EOF
210+ default:
211+ close(done)
212+ return detached, nil
213+ }
214+ case <-done:
215+ return ports.Frame{}, io.EOF
216+ }
217+ }).Maybe()
218+ tr.EXPECT().Close().Return(nil).Once()
219+
220+ // Push a resize event shortly after attach begins.
221+ go func() {
222+ time.Sleep(20 * time.Millisecond)
223+ resizeCh <- domain.Size{Cols: 120, Rows: 40}
224+ }()
225+
226+ err := client.Attach(context.Background(), tr, tm, ipc.IntentEphemeral, "")
227+ require.NoError(t, err)
228+ select {
229+ case r := <-gotResize:
230+ require.Equal(t, domain.Size{Cols: 120, Rows: 40}, r.Size)
231+ default:
232+ t.Fatal("resize was not forwarded")
233+ }
234+}
@@ -0,0 +1,234 @@
1+package client_test
2+
3+import (
4+ "bytes"
5+ "context"
6+ "errors"
7+ "io"
8+ "log/slog"
9+ "sync/atomic"
10+ "testing"
11+ "time"
12+
13+ "github.com/stretchr/testify/mock"
14+ "github.com/stretchr/testify/require"
15+
16+ "github.com/bnema/vev/internal/adapters/ipc"
17+ "github.com/bnema/vev/internal/domain"
18+ "github.com/bnema/vev/internal/ports"
19+ portsmocks "github.com/bnema/vev/internal/ports/mocks"
20+ "github.com/bnema/vev/internal/usecase/client"
21+)
22+
23+func init() {
24+ // Keep client diagnostics out of the test runner's stderr.
25+ slog.SetDefault(slog.New(slog.NewTextHandler(io.Discard, nil)))
26+}
27+
28+// recvItem is one scripted Recv result.
29+type recvItem struct {
30+ f ports.Frame
31+ err error
32+}
33+
34+// scriptRecv wires a MockTransport's Recv to yield the given items in order,
35+// then block forever (simulating a live but idle connection). It returns a
36+// cleanup that unblocks the parked reader.
37+func scriptRecv(tr *portsmocks.MockTransport, items ...recvItem) func() {
38+ ch := make(chan recvItem, len(items))
39+ for _, it := range items {
40+ ch <- it
41+ }
42+ done := make(chan struct{})
43+ tr.EXPECT().Recv().RunAndReturn(func() (ports.Frame, error) {
44+ select {
45+ case it := <-ch:
46+ return it.f, it.err
47+ case <-done:
48+ return ports.Frame{}, io.EOF
49+ }
50+ }).Maybe()
51+ return func() { close(done) }
52+}
53+
54+func frameOf(t ports.MsgType, payload []byte) ports.Frame {
55+ return ports.Frame{Type: t, Payload: payload}
56+}
57+
58+// blockingReader blocks on Read until closed, then returns EOF. Stands in
59+// for a terminal's stdin that produces nothing during the test.
60+type blockingReader struct{ ch chan struct{} }
61+
62+func newBlockingReader() *blockingReader { return &blockingReader{ch: make(chan struct{})} }
63+
64+func (b *blockingReader) Read(p []byte) (int, error) {
65+ <-b.ch
66+ return 0, io.EOF
67+}
68+func (b *blockingReader) unblock() { close(b.ch) }
69+
70+func newHappyTerminal(t *testing.T, out *bytes.Buffer, restoreCount *atomic.Int32, resizeCh chan domain.Size) (*portsmocks.MockTerminal, *blockingReader) {
71+ tm := portsmocks.NewMockTerminal(t)
72+ tm.EXPECT().Size().Return(domain.Size{Cols: 80, Rows: 24}, nil).Once()
73+ tm.EXPECT().EnterRaw().Return(func() error {
74+ restoreCount.Add(1)
75+ return nil
76+ }, nil).Once()
77+ in := newBlockingReader()
78+ tm.EXPECT().In().Return(in).Maybe()
79+ tm.EXPECT().Out().Return(out).Maybe()
80+ tm.EXPECT().Flush().Return(nil).Maybe()
81+ tm.EXPECT().ResizeEvents().Return(resizeCh).Maybe()
82+ return tm, in
83+}
84+
85+func isType(typ ports.MsgType) any {
86+ return mock.MatchedBy(func(f ports.Frame) bool { return f.Type == typ })
87+}
88+
89+func TestAttachHappyPath(t *testing.T) {
90+ var out bytes.Buffer
91+ var restoreCount atomic.Int32
92+ resizeCh := make(chan domain.Size)
93+ tm, in := newHappyTerminal(t, &out, &restoreCount, resizeCh)
94+ defer in.unblock()
95+
96+ tr := portsmocks.NewMockTransport(t)
97+ tr.EXPECT().Send(isType(ports.MsgHello)).Return(nil).Once()
98+ unblock := scriptRecv(tr,
99+ recvItem{f: frameOf(ports.MsgWelcome, ipc.MarshalWelcome(ipc.Welcome{SessionID: "s1", SessionName: "main"}))},
100+ recvItem{f: frameOf(ports.MsgOutput, ipc.MarshalOutput(ipc.Output{Data: []byte("hello world")}))},
101+ recvItem{f: frameOf(ports.MsgDetached, ipc.MarshalDetached(ipc.Detached{Reason: ipc.ReasonDetach}))},
102+ )
103+ defer unblock()
104+ tr.EXPECT().Close().Return(nil).Once()
105+
106+ err := client.Attach(context.Background(), tr, tm, ipc.IntentEphemeral, "")
107+ require.NoError(t, err)
108+ require.Equal(t, "hello world", out.String())
109+ require.Equal(t, int32(1), restoreCount.Load(), "restore must run exactly once")
110+}
111+
112+func TestAttachVersionMismatch(t *testing.T) {
113+ tm := portsmocks.NewMockTerminal(t)
114+ tm.EXPECT().Size().Return(domain.Size{Cols: 80, Rows: 24}, nil).Once()
115+ // EnterRaw must NOT be called on the error-before-welcome path.
116+
117+ tr := portsmocks.NewMockTransport(t)
118+ tr.EXPECT().Send(isType(ports.MsgHello)).Return(nil).Once()
119+ tr.EXPECT().Recv().Return(
120+ frameOf(ports.MsgError, ipc.MarshalErrorMsg(ipc.ErrorMsg{Code: ipc.ErrVersionMismatch, Text: "version mismatch"})),
121+ nil,
122+ ).Once()
123+ tr.EXPECT().Close().Return(nil).Once()
124+
125+ err := client.Attach(context.Background(), tr, tm, ipc.IntentAttach, "main")
126+ require.Error(t, err)
127+ var pe *client.ProtocolError
128+ require.True(t, errors.As(err, &pe), "want *client.ProtocolError, got %T", err)
129+ require.Equal(t, ipc.ErrVersionMismatch, pe.Code)
130+}
131+
132+func TestAttachRestoredOnRecvErrorMidStream(t *testing.T) {
133+ var out bytes.Buffer
134+ var restoreCount atomic.Int32
135+ resizeCh := make(chan domain.Size)
136+ tm, in := newHappyTerminal(t, &out, &restoreCount, resizeCh)
137+ defer in.unblock()
138+
139+ tr := portsmocks.NewMockTransport(t)
140+ tr.EXPECT().Send(isType(ports.MsgHello)).Return(nil).Once()
141+ boom := errors.New("connection reset")
142+ unblock := scriptRecv(tr,
143+ recvItem{f: frameOf(ports.MsgWelcome, ipc.MarshalWelcome(ipc.Welcome{SessionID: "s1"}))},
144+ recvItem{err: boom},
145+ )
146+ defer unblock()
147+ tr.EXPECT().Close().Return(nil).Once()
148+
149+ err := client.Attach(context.Background(), tr, tm, ipc.IntentEphemeral, "")
150+ require.Error(t, err)
151+ require.Equal(t, int32(1), restoreCount.Load(), "restore must run after mid-stream Recv error")
152+}
153+
154+func TestAttachDaemonVanishedOnEOF(t *testing.T) {
155+ var out bytes.Buffer
156+ var restoreCount atomic.Int32
157+ resizeCh := make(chan domain.Size)
158+ tm, in := newHappyTerminal(t, &out, &restoreCount, resizeCh)
159+ defer in.unblock()
160+
161+ tr := portsmocks.NewMockTransport(t)
162+ tr.EXPECT().Send(isType(ports.MsgHello)).Return(nil).Once()
163+ unblock := scriptRecv(tr,
164+ recvItem{f: frameOf(ports.MsgWelcome, ipc.MarshalWelcome(ipc.Welcome{SessionID: "s1"}))},
165+ recvItem{err: io.EOF},
166+ )
167+ defer unblock()
168+ tr.EXPECT().Close().Return(nil).Once()
169+
170+ err := client.Attach(context.Background(), tr, tm, ipc.IntentEphemeral, "")
171+ require.Error(t, err)
172+ require.ErrorIs(t, err, io.EOF)
173+ require.Equal(t, int32(1), restoreCount.Load())
174+}
175+
176+func TestAttachForwardsResize(t *testing.T) {
177+ var out bytes.Buffer
178+ var restoreCount atomic.Int32
179+ resizeCh := make(chan domain.Size)
180+ tm, in := newHappyTerminal(t, &out, &restoreCount, resizeCh)
181+ defer in.unblock()
182+
183+ detachAfterResize := make(chan struct{})
184+
185+ tr := portsmocks.NewMockTransport(t)
186+ tr.EXPECT().Send(isType(ports.MsgHello)).Return(nil).Once()
187+ // The resize frame is forwarded via the sender goroutine.
188+ gotResize := make(chan ipc.Resize, 1)
189+ tr.EXPECT().Send(isType(ports.MsgResize)).RunAndReturn(func(f ports.Frame) error {
190+ r, _ := ipc.UnmarshalResize(f.Payload)
191+ gotResize <- r
192+ close(detachAfterResize)
193+ return nil
194+ }).Once()
195+
196+ welcome := frameOf(ports.MsgWelcome, ipc.MarshalWelcome(ipc.Welcome{SessionID: "s1"}))
197+ detached := frameOf(ports.MsgDetached, ipc.MarshalDetached(ipc.Detached{Reason: ipc.ReasonDetach}))
198+ recvCh := make(chan recvItem, 2)
199+ recvCh <- recvItem{f: welcome}
200+ done := make(chan struct{})
201+ tr.EXPECT().Recv().RunAndReturn(func() (ports.Frame, error) {
202+ select {
203+ case it := <-recvCh:
204+ return it.f, it.err
205+ case <-detachAfterResize:
206+ // Deliver the detach once the resize has been observed.
207+ select {
208+ case <-done:
209+ return ports.Frame{}, io.EOF
210+ default:
211+ close(done)
212+ return detached, nil
213+ }
214+ case <-done:
215+ return ports.Frame{}, io.EOF
216+ }
217+ }).Maybe()
218+ tr.EXPECT().Close().Return(nil).Once()
219+
220+ // Push a resize event shortly after attach begins.
221+ go func() {
222+ time.Sleep(20 * time.Millisecond)
223+ resizeCh <- domain.Size{Cols: 120, Rows: 40}
224+ }()
225+
226+ err := client.Attach(context.Background(), tr, tm, ipc.IntentEphemeral, "")
227+ require.NoError(t, err)
228+ select {
229+ case r := <-gotResize:
230+ require.Equal(t, domain.Size{Cols: 120, Rows: 40}, r.Size)
231+ default:
232+ t.Fatal("resize was not forwarded")
233+ }
234+}
A internal/usecase/daemon/daemon.go +29 -0
@@ -0,0 +1,29 @@
1+// Package daemon holds vev's server-side session multiplexer use case.
2+//
3+// The daemon core — accepting client connections, spawning PTY-backed
4+// sessions, and fanning I/O between them — is implemented in a later
5+// milestone (Task 11). This file provides only the Serve entry point that
6+// the app layer wires the hidden --daemon path to, so the client, CLI, and
7+// auto-spawn machinery can be built and tested against a stable seam now.
8+package daemon
9+
10+import (
11+ "errors"
12+
13+ "github.com/bnema/vev/internal/ports"
14+)
15+
16+// ErrNotImplemented is returned by Serve until the daemon core lands in
17+// Task 11. It is deliberately a distinct sentinel so the app layer (and
18+// tests) can recognise the not-yet-implemented state.
19+var ErrNotImplemented = errors.New("vev: daemon core not implemented yet (Task 11)")
20+
21+// Serve runs the daemon's accept loop over ln, owning ln for its lifetime.
22+//
23+// TODO(Task 11): implement the real accept/session/dispatch loop. For now
24+// it accepts the listener and immediately reports ErrNotImplemented without
25+// closing it (the caller owns ln and closes it).
26+func Serve(ln ports.Listener) error {
27+ _ = ln
28+ return ErrNotImplemented
29+}
@@ -0,0 +1,29 @@
1+// Package daemon holds vev's server-side session multiplexer use case.
2+//
3+// The daemon core — accepting client connections, spawning PTY-backed
4+// sessions, and fanning I/O between them — is implemented in a later
5+// milestone (Task 11). This file provides only the Serve entry point that
6+// the app layer wires the hidden --daemon path to, so the client, CLI, and
7+// auto-spawn machinery can be built and tested against a stable seam now.
8+package daemon
9+
10+import (
11+ "errors"
12+
13+ "github.com/bnema/vev/internal/ports"
14+)
15+
16+// ErrNotImplemented is returned by Serve until the daemon core lands in
17+// Task 11. It is deliberately a distinct sentinel so the app layer (and
18+// tests) can recognise the not-yet-implemented state.
19+var ErrNotImplemented = errors.New("vev: daemon core not implemented yet (Task 11)")
20+
21+// Serve runs the daemon's accept loop over ln, owning ln for its lifetime.
22+//
23+// TODO(Task 11): implement the real accept/session/dispatch loop. For now
24+// it accepts the listener and immediately reports ErrNotImplemented without
25+// closing it (the caller owns ln and closes it).
26+func Serve(ln ports.Listener) error {
27+ _ = ln
28+ return ErrNotImplemented
29+}

Keyboard shortcuts

Single-key shortcuts are on. ctrl k and esc work whatever this says.

ctrl k :
go to a section, branch, tag, file or repository
?
this list
j k
move the row cursor down / up
enter
open the row under the cursor; on Files, a directory opens or closes in place
↑ ↓ ← → home end
in the file tree (focus on a row): previous / next row, close / open a directory or go to its parent / first entry, first / last row. They never open a file
l h
on Files: open a directory or a file / close a directory or go up to its parent, then to the parent page. Elsewhere: into the side panel (on a merge request, its files) / back to the content; there j k move between its controls
g g G
first / last row
/
filter the list; on Files, find a file anywhere in the repository (esc clears, enter opens the best match)
b
switch branch or tag
g s g f g c g b g m g i g r g a
tabs: summary, files, commits, branches & tags, merge requests, issues, releases, actions (when available)
g h
home
w
wide mode: use the whole width of the screen, or not (remembered)
y
copy a permalink pinned to the commit
d u x
diff: split / unified / collapse or expand all
e E
diff: the whole file in view / every file, with the lines the hunks hide
n
merge request: next note in the diff
esc
close a dialog, leave a field

Branches 12

Tags 11

↑↓ moveenter openesc close