1package daemon
2
3// Explicit CommandResult daemon contract for the one-shot control path: a
4// dispatched command has exactly one closed terminal outcome; a request whose
5// deadline expires while its command is still running answers with exactly one
6// correlated CommandOutcomeUnknown and never a second result for the late
7// completion; and every definite refusal or success preserves the wire request
8// correlation and a bounded error. All cases are deterministic and bounded
9// (fake clock / PTY gates / awaitFrame timeouts), never sleeping.
10
11import (
12 "context"
13 "io"
14 "sync"
15 "testing"
16 "time"
17
18 "github.com/stretchr/testify/require"
19
20 "github.com/bnema/vev/internal/adapters/sessionwire"
21 "github.com/bnema/vev/internal/domain"
22 "github.com/bnema/vev/internal/ports"
23 "github.com/bnema/vev/internal/protocol"
24 "github.com/bnema/vev/internal/protocol/wire"
25)
26
27// gatedControlFactory blocks its first pane Open until release, so a test holds
28// a dispatched command inside its PTY spawn while the request deadline expires.
29// Later Opens (a background floating prewarm) return immediately.
30type gatedControlFactory struct {
31 entered chan struct{}
32 release chan struct{}
33 // gateOnce signals first-Open entry exactly once; releaseOnce closes the
34 // release channel exactly once. They are separate because Open blocks inside
35 // its own Once body waiting for the release that releaseOnce performs.
36 gateOnce sync.Once
37 releaseOnce sync.Once
38
39 mu sync.Mutex
40 ptys []*gatedControlPTY
41 first *gatedControlPTY
42 firstOpened bool
43 shut bool
44}
45
46func newGatedControlFactory() *gatedControlFactory {
47 f := &gatedControlFactory{entered: make(chan struct{}), release: make(chan struct{})}
48 // Pre-allocate the gated pane so a test can await its reader even while the
49 // first Open is still parked inside the gate.
50 f.first = &gatedControlPTY{readStarted: make(chan struct{}), stop: make(chan struct{})}
51 return f
52}
53
54func (f *gatedControlFactory) Open(_ context.Context, _ string, _ []string, _ []string, _ string, _ domain.Geometry) (ports.PTY, error) {
55 f.mu.Lock()
56 first := f.first
57 gate := !f.firstOpened
58 if gate {
59 f.firstOpened = true
60 }
61 f.mu.Unlock()
62 if gate {
63 f.gateOnce.Do(func() { close(f.entered) })
64 <-f.release
65 return first, nil
66 }
67 p := &gatedControlPTY{readStarted: make(chan struct{}), stop: make(chan struct{})}
68 f.mu.Lock()
69 f.ptys = append(f.ptys, p)
70 shut := f.shut
71 f.mu.Unlock()
72 if shut {
73 _ = p.Close()
74 }
75 return p, nil
76}
77
78func (f *gatedControlFactory) releaseOpen() {
79 f.releaseOnce.Do(func() { close(f.release) })
80}
81
82func (f *gatedControlFactory) firstReadStarted() <-chan struct{} {
83 f.mu.Lock()
84 defer f.mu.Unlock()
85 return f.first.readStarted
86}
87
88func (f *gatedControlFactory) close() {
89 f.releaseOpen()
90 f.mu.Lock()
91 defer f.mu.Unlock()
92 f.shut = true
93 _ = f.first.Close()
94 for _, p := range f.ptys {
95 _ = p.Close()
96 }
97}
98
99// gatedControlPTY signals when its reader starts and parks until closed, so a
100// test observes a committed tab's launched reader without sleeping.
101type gatedControlPTY struct {
102 readStarted chan struct{}
103 stop chan struct{}
104 startOnce sync.Once
105 closeOnce sync.Once
106}
107
108func (p *gatedControlPTY) Read([]byte) (int, error) {
109 p.startOnce.Do(func() { close(p.readStarted) })
110 <-p.stop
111 return 0, io.EOF
112}
113
114func (*gatedControlPTY) Write(b []byte) (int, error) { return len(b), nil }
115func (*gatedControlPTY) Resize(domain.Geometry) error { return nil }
116func (*gatedControlPTY) Pid() int { return 0 }
117func (*gatedControlPTY) ForegroundPgid() (int, error) { return 0, nil }
118
119func (p *gatedControlPTY) Close() error {
120 p.closeOnce.Do(func() { close(p.stop) })
121 return nil
122}
123
124func controlSessionTabCount(t *testing.T, d *Daemon, name string) int {
125 t.Helper()
126 d.mu.Lock()
127 sess := d.findByNameLocked(name)
128 d.mu.Unlock()
129 require.NotNil(t, sess, "session %q must exist", name)
130 sess.mu.Lock()
131 defer sess.mu.Unlock()
132 return len(sess.tabs)
133}
134
135// requireNoFurtherCommandResults proves the handler produced no second result.
136func requireNoFurtherCommandResults(t *testing.T, sends chan wire.Envelope) {
137 t.Helper()
138 for {
139 select {
140 case frame := <-sends:
141 if envelopeMessageName(t, frame.Payload) != "CommandResult" {
142 continue
143 }
144 message, err := sessionwire.DecodeServerEnvelope(frame.Payload)
145 require.NoError(t, err)
146 t.Fatalf("command answered a second result: %+v", message)
147 default:
148 return
149 }
150 }
151}
152
153// commandResultCaptureTransport records every server envelope and whether the
154// one-shot control connection was retired, so a test can assert both without a
155// mock expectation race.
156type commandResultCaptureTransport struct {
157 mu sync.Mutex
158 sends []wire.Envelope
159 closed bool
160}
161
162func (c *commandResultCaptureTransport) Send(envelope wire.Envelope) error {
163 c.mu.Lock()
164 defer c.mu.Unlock()
165 c.sends = append(c.sends, envelope)
166 return nil
167}
168
169func (*commandResultCaptureTransport) Recv() (wire.Envelope, error) {
170 return wire.Envelope{}, io.EOF
171}
172
173func (c *commandResultCaptureTransport) Close() error {
174 c.mu.Lock()
175 defer c.mu.Unlock()
176 c.closed = true
177 return nil
178}
179
180func (c *commandResultCaptureTransport) commandResults(t *testing.T) []protocol.CommandResult {
181 t.Helper()
182 c.mu.Lock()
183 frames := append([]wire.Envelope(nil), c.sends...)
184 c.mu.Unlock()
185 results := make([]protocol.CommandResult, 0, len(frames))
186 for _, frame := range frames {
187 message, err := sessionwire.DecodeServerEnvelope(frame.Payload)
188 require.NoError(t, err)
189 if result, ok := message.(protocol.CommandResult); ok {
190 results = append(results, result)
191 }
192 }
193 return results
194}
195
196func (c *commandResultCaptureTransport) isClosed() bool {
197 c.mu.Lock()
198 defer c.mu.Unlock()
199 return c.closed
200}
201
202// TestHandleCommandTimeoutSendsOneCorrelatedUnknownAndDropsLateCompletion is
203// the core timeout contract: the command is dispatched and still running when
204// the deadline expires, so the client receives exactly one CommandOutcomeUnknown
205// correlated to its RequestID. The command then commits late; that late
206// completion produces no second result and cannot change the observed outcome.
207func TestHandleCommandTimeoutSendsOneCorrelatedUnknownAndDropsLateCompletion(t *testing.T) {
208 factory := newGatedControlFactory()
209 clock := &signalClock{timers: make(chan *signalTimer, 16)}
210 d := newTestDaemon(t, factory, clock)
211 addControlSession(d, "work", "t_work", "p_work")
212 t.Cleanup(func() {
213 factory.close()
214 d.sessWg.Wait()
215 })
216
217 frame := commandFrame(t, protocol.CommandRequest{RequestID: 17, Slug: "new-tab", TargetSession: "work"})
218 tr, sends, releaseConn := newConn(t, frame)
219 defer releaseConn()
220 done := make(chan error, 1)
221 go func() { done <- d.handleCommandFrame(tr, frame) }()
222
223 // The command is admitted and parked inside its pane spawn.
224 awaitTestCompletion(t, factory.entered, "command was not dispatched")
225 timer := <-clock.timers
226 require.Equal(t, CommandRequestTimeout, timer.duration)
227 timer.ch <- time.Time{}
228
229 result := awaitCommandResult(t, sends)
230 require.Equal(t, uint64(17), result.RequestID, "the unknown outcome must correlate to its request")
231 require.Equal(t, protocol.CommandOutcomeUnknown, result.Outcome)
232 require.True(t, result.Valid(), "%+v must obey the outcome contract", result)
233 require.Zero(t, result.Code, "an unknown outcome carries no definite code")
234 require.Empty(t, result.Output, "an unknown outcome carries no output")
235 require.Equal(t, ErrCommandRequestTimeout.Error(), result.Text)
236 require.NoError(t, <-done, "the one-shot handler must answer and return")
237
238 // Release the parked command: it now commits after the client already saw
239 // the unknown outcome.
240 factory.releaseOpen()
241 awaitTestCompletion(t, factory.firstReadStarted(), "late completion never committed its tab")
242 require.Equal(t, 2, controlSessionTabCount(t, d, "work"), "the timed-out command still committed late")
243
244 requireNoFurtherCommandResults(t, sends)
245}
246
247// TestHandleCommandTimeoutRetiresConnectionAfterOneResult proves the timeout
248// path attempts exactly one correlated unknown result and then retires the
249// one-shot control connection, so a late completion has no channel to answer on.
250func TestHandleCommandTimeoutRetiresConnectionAfterOneResult(t *testing.T) {
251 factory := newGatedControlFactory()
252 clock := &signalClock{timers: make(chan *signalTimer, 16)}
253 d := newTestDaemon(t, factory, clock)
254 addControlSession(d, "work", "t_work", "p_work")
255 t.Cleanup(func() {
256 factory.close()
257 d.sessWg.Wait()
258 })
259
260 capture := &commandResultCaptureTransport{}
261 frame := commandFrame(t, protocol.CommandRequest{RequestID: 23, Slug: "new-tab", TargetSession: "work"})
262 done := make(chan error, 1)
263 go func() { done <- d.handleCommandFrame(&rawServerConnection{raw: capture}, frame) }()
264
265 awaitTestCompletion(t, factory.entered, "command was not dispatched")
266 timer := <-clock.timers
267 require.Equal(t, CommandRequestTimeout, timer.duration)
268 timer.ch <- time.Time{}
269 require.NoError(t, <-done)
270
271 results := capture.commandResults(t)
272 require.Len(t, results, 1, "a timed-out command answers exactly one result")
273 require.Equal(t, uint64(23), results[0].RequestID)
274 require.Equal(t, protocol.CommandOutcomeUnknown, results[0].Outcome)
275 require.True(t, capture.isClosed(), "a one-shot control connection is retired after its result")
276}
277
278// TestHandleCommandDefiniteOutcomesPreserveCorrelationAndOutcome proves each
279// definite refusal and each success answers once with the closed outcome, the
280// wire request correlation, and (for failures) a bounded error and no output.
281func TestHandleCommandDefiniteOutcomesPreserveCorrelationAndOutcome(t *testing.T) {
282 tests := []struct {
283 name string
284 request protocol.CommandRequest
285 wantOutcome protocol.CommandOutcome
286 wantCode uint16
287 wantOutput bool
288 }{
289 {
290 name: "success carries a bounded output",
291 request: protocol.CommandRequest{Slug: "list-sessions"},
292 wantOutcome: protocol.CommandSucceeded,
293 },
294 {
295 name: "unknown slug is a definite failure",
296 request: protocol.CommandRequest{Slug: "no-such"},
297 wantOutcome: protocol.CommandFailed,
298 wantCode: protocol.ErrUnknownCommand,
299 },
300 {
301 name: "non-scriptable command is a definite failure",
302 request: protocol.CommandRequest{Slug: "session-picker"},
303 wantOutcome: protocol.CommandFailed,
304 wantCode: protocol.ErrNotScriptable,
305 },
306 {
307 name: "version mismatch is a definite failure",
308 request: protocol.CommandRequest{Version: protocol.Version + 1, Slug: "list-sessions"},
309 wantOutcome: protocol.CommandFailed,
310 wantCode: protocol.ErrVersionMismatch,
311 },
312 {
313 name: "attached relay refusal is a definite failure",
314 request: protocol.CommandRequest{Attached: true, Slug: "list-sessions"},
315 wantOutcome: protocol.CommandFailed,
316 wantCode: protocol.ErrNotScriptable,
317 },
318 }
319 for i, tt := range tests {
320 t.Run(tt.name, func(t *testing.T) {
321 d := newTestDaemon(t, nil, stubClock{})
322 request := tt.request
323 request.RequestID = uint64(31 + i)
324
325 result := sendCommand(t, d, request)
326
327 require.Equal(t, request.RequestID, result.RequestID, "the outcome must correlate to its request")
328 require.Equal(t, tt.wantOutcome, result.Outcome)
329 require.True(t, result.Valid(), "%+v must obey the outcome contract", result)
330 if tt.wantOutcome == protocol.CommandFailed {
331 require.Equal(t, tt.wantCode, result.Code)
332 require.NotEmpty(t, result.Text, "a definite failure must carry bounded error text")
333 require.LessOrEqual(t, len(result.Text), wire.ControlEnvelopeLimit)
334 require.Empty(t, result.Output, "a failure must not carry success output")
335 } else {
336 require.Zero(t, result.Code)
337 require.Empty(t, result.Text)
338 require.NotEmpty(t, result.Output)
339 }
340 })
341 }
342}
343
344// TestAttachedCommandRefusalSendsDefiniteFailedOutcome proves an attached
345// command refusal answers with a decodable, definite failure correlated to the
346// frame's RequestID rather than an unspecified outcome the wire boundary would
347// refuse.
348func TestAttachedCommandRefusalSendsDefiniteFailedOutcome(t *testing.T) {
349 d, sess, ac, _ := newManualSessionWithPTYs(t, nil)
350 transport := &closeTrackingTransport{}
351 ac.replaceTransport(transport)
352 rc := d.attachCoordinator(sess, nil, ac, true)
353 token := sess.captureAttachmentCapability(ac, transport)
354 token.lease = rc.attachmentLease(ac)
355 ac.installTestAttachmentCapability(token)
356
357 require.False(t, d.handleAttachmentClientFrame(token, mustClientEnvelope(protocol.CommandRequest{
358 Version: protocol.Version, RequestID: 41, Slug: "no-such", Attached: true,
359 })))
360
361 var result protocol.CommandResult
362 found := false
363 for _, frame := range transport.Sends() {
364 message, err := sessionwire.DecodeServerEnvelope(frame.Payload)
365 require.NoError(t, err)
366 if decoded, ok := message.(protocol.CommandResult); ok {
367 result, found = decoded, true
368 }
369 }
370 require.True(t, found, "an attached refusal must answer with a CommandResult")
371 require.Equal(t, uint64(41), result.RequestID)
372 require.Equal(t, protocol.CommandFailed, result.Outcome)
373 require.Equal(t, protocol.ErrUnknownCommand, result.Code)
374 require.True(t, result.Valid(), "%+v must obey the outcome contract", result)
375}