1package daemon
2
3import (
4 "errors"
5 "io"
6 "net"
7 "runtime"
8
9 "github.com/bnema/vev/internal/domain"
10 "github.com/bnema/vev/internal/ports"
11 "github.com/bnema/vev/internal/protocol"
12 "github.com/bnema/vev/internal/usecase/command"
13)
14
15const connectionSnapshotAttempts = 8
16
17// currentAttachmentConnection retries the lock-free session-to-role snapshot when a
18// handoff lands between those reads. The attachment gate still performs the final
19// mutation admission after this function returns.
20func (d *Daemon) currentAttachmentConnection(ac *attachedClient, tr ports.ServerConnection) (*session, attachmentCapability, bool) {
21 for range connectionSnapshotAttempts {
22 if !ac.currentTransportIs(tr) {
23 break
24 }
25 sess := ac.currentAttachmentSession()
26 if sess == nil {
27 return nil, attachmentCapability{}, false
28 }
29 if d.afterConnectionSessionSnapshot != nil {
30 d.afterConnectionSessionSnapshot(sess)
31 }
32 token := captureAttachmentCapability(sess, ac, tr)
33 if token.ac == nil || ac.currentAttachmentSession() != sess {
34 runtime.Gosched()
35 continue
36 }
37 return sess, token, true
38 }
39 return nil, attachmentCapability{}, false
40}
41
42// runConnLoop is the per-connection input router: it pumps client messages
43// until detach, EOF, or a transport error. The attachment generation is
44// resolved after every Recv so a stale transport cannot execute a new frame.
45func (d *Daemon) runConnLoop(ac *attachedClient) {
46 tr := ac.transport()
47 if tr == nil {
48 return
49 }
50 for {
51 if !ac.currentTransportIs(tr) || ac.currentAttachmentSession() == nil {
52 return
53 }
54 message, err := tr.ReceiveClient()
55 if err != nil {
56 var failure *protocol.DecodeFailure
57 if errors.As(err, &failure) {
58 d.log.Warn("rejected client message",
59 "category", failure.Category,
60 "kind", failure.Kind,
61 "type", failure.Type,
62 "version", failure.Version,
63 "request", failure.RequestID,
64 "has_request", failure.HasRequestID,
65 )
66 continue
67 }
68 // A transport the daemon already swapped out or closed itself is
69 // ordinary teardown, not a lost connection.
70 if !errors.Is(err, io.EOF) && !errors.Is(err, net.ErrClosed) && ac.currentTransportIs(tr) {
71 if sess := ac.currentAttachmentSession(); sess != nil {
72 d.log.Warn("client connection lost", "session", sess.nameSnapshot(), "err", err)
73 }
74 }
75 for range connectionSnapshotAttempts {
76 sess, _, ok := d.currentAttachmentConnection(ac, tr)
77 if !ok {
78 return
79 }
80 d.clientGone(sess, ac, tr, false)
81 current := ac.currentAttachmentSession()
82 if current == sess || !ac.currentTransportIs(tr) || current == nil {
83 return
84 }
85 }
86 return
87 }
88 _, token, ok := d.currentAttachmentConnection(ac, tr)
89 if !ok {
90 return
91 }
92 if d.handleAttachmentClientMessage(token, message) {
93 return
94 }
95 }
96}
97
98// handleAttachmentClientMessage owns the attached-client protocol.
99func (d *Daemon) handleAttachmentClientMessage(capability attachmentCapability, message protocol.ClientMessage) bool {
100 if !capability.current() {
101 return false
102 }
103 if d.afterAttachmentFrameDispatch != nil {
104 d.afterAttachmentFrameDispatch(capability)
105 }
106 var transitionErr error
107 switch request := message.(type) {
108 case protocol.SuspendAttachment:
109 transitionErr = d.suspendAttachment(capability, request)
110 case protocol.ActivateAttachment:
111 transitionErr = d.activateAttachment(capability.ac, capability.transport, request)
112 default:
113 if capability.ac.attachmentActivity() != attachmentActive {
114 if _, ping := message.(protocol.Ping); ping {
115 if d.boundedSendErr(capability.ac, protocol.Pong{}) != nil {
116 d.clientGoneWithoutNotice(capability.sess, capability.ac, capability.transport.transport, true)
117 return true
118 }
119 }
120 if detach, ok := message.(protocol.Detach); ok {
121 d.clientGoneWithoutNotice(capability.sess, capability.ac, capability.transport.transport, true)
122 if detach.Closed {
123 // Safe even if this call did not detach: the reap re-checks
124 // emptiness under the architecture locks.
125 d.reapAbandonedEphemeral(capability.sess)
126 }
127 return true
128 }
129 return false
130 }
131 return d.handleActiveAttachmentClientMessage(capability, message)
132 }
133 if transitionErr != nil {
134 d.clientGoneWithoutNotice(capability.sess, capability.ac, capability.transport.transport, true)
135 return true
136 }
137 return false
138}
139
140func (d *Daemon) handleActiveAttachmentClientMessage(capability attachmentCapability, message protocol.ClientMessage) bool {
141 effect, admitted := capability.ac.beginAttachmentEffect(capability)
142 if !admitted {
143 return false
144 }
145 defer effect.End()
146 if d.afterAttachmentEffectAdmitted != nil {
147 d.afterAttachmentEffectAdmitted(effect.capability())
148 }
149 switch message := message.(type) {
150 case protocol.Input:
151 effect.uiActionID = message.ActionID
152 d.handleSequencedInputForAttachment(effect, message.InputSeq, message.Data)
153 case protocol.UIFence:
154 if !d.registerAttachmentUIFence(effect, message.ActionID) {
155 // Invalid/overflowing fences retire this exact attachment, not just
156 // its carriage. Cleanup is asynchronous so even refusal cannot pin
157 // dispatch behind a blocked render send.
158 launchCleanup := d.reserveAttachmentSendErrorCleanup(effect.capability(), effect.transport.transport)
159 effect.End()
160 launchCleanup()
161 return true
162 }
163 case protocol.Resize:
164 if effect.current() {
165 d.resizeAttachmentGeometryForLease(effect, message.Geometry())
166 }
167 case protocol.Theme:
168 d.applyThemeForAttachment(effect, message)
169 case protocol.ImagePush:
170 d.handleSequencedImagePushForAttachment(effect, message.InputSeq, message)
171 case protocol.ClientNotice:
172 d.handleClientNoticeForAttachment(effect, message)
173 case protocol.Detach:
174 if effect.current() {
175 sess := effect.capability().sess
176 if d.clientGoneForAttachment(effect, true) && message.Closed {
177 d.reapAbandonedEphemeral(sess)
178 }
179 return true
180 }
181 case protocol.Ack:
182 d.ackOutput(effect, message.Epoch, message.State)
183 case protocol.OutputResetRequest:
184 d.resetOutput(effect)
185 case protocol.SamePeerSwitchRequest:
186 d.switchSamePeerForAttachment(effect, message)
187 case protocol.SelectTab:
188 d.selectTabForAttachment(effect, message)
189 case protocol.TerminalFocus:
190 d.applyTerminalFocusForAttachment(effect, message)
191 case protocol.ModifierState:
192 d.applyModifierStateForAttachment(effect, message)
193 case protocol.RecentRouteSnapshot:
194 replayIdentity := effect.ac.setRouteSnapshot(message)
195 d.pokeAttentionTicker()
196 d.invalidateRender(effect.sess, effect.ac, false, "client_frame_routing.go:route-snapshot")
197 if !replayIdentity {
198 break
199 }
200 if err := d.sendCommittedRouteIdentityForAttachment(effect); err != nil {
201 d.detachOnAttachmentSendError(effect.capability(), effect.transport.transport)
202 }
203 case protocol.RouteAttentionSubscription:
204 effect.ac.setRouteAttentionSubscription(message, effect.transport, d.clock.Now())
205 d.reconcileRouteHistory(effect.ac, effect)
206 d.invalidateRender(effect.sess, effect.ac, false, "client_frame_routing.go:route-attention")
207 case protocol.SessionCreationFailure:
208 notice := "couldn't create session"
209 switch message.Code {
210 case protocol.RouteFailureStaleSelection, protocol.RouteFailureNoSuchRoute:
211 notice = "that destination is no longer available"
212 case protocol.RouteFailureTargetChanged:
213 notice = "the destination returned an unexpected session"
214 case protocol.RouteFailureOriginUnavailable:
215 notice = "couldn't restore the original session after creation failed"
216 case protocol.RouteFailureUnavailable:
217 notice = "the selected destination is unavailable"
218 }
219 d.notify(effect.sess, domain.NoticeWarn, domain.NoticeSessionUnavailable, notice, nil)
220 case protocol.RouteNavigationFailure:
221 notice := "route navigation failed"
222 switch message.Code {
223 case protocol.RouteFailureStaleSelection:
224 notice = "that recent route is no longer available"
225 case protocol.RouteFailureTargetChanged:
226 notice = "that recent route changed before attach"
227 case protocol.RouteFailureOriginUnavailable:
228 notice = "the original route is no longer available"
229 case protocol.RouteFailureUnavailable:
230 notice = "the selected route is unavailable"
231 case protocol.RouteFailureNoSuchRoute:
232 notice = "that recent route no longer exists"
233 }
234 d.notify(effect.sess, domain.NoticeWarn, domain.NoticeSessionUnavailable, notice, nil)
235 case protocol.CommandRequest:
236 result := d.executeAttachedCommand(effect, message)
237 resultEffect, ok := d.commandResultEffect(effect)
238 if !ok {
239 // A remote handoff or explicit detach deliberately retired the old
240 // attachment; it must not be treated as a stale-result send error.
241 return true
242 }
243 if err := resultEffect.sendControl(serverCommandResult(result)); err != nil {
244 resultEffect.End()
245 d.detachOnAttachmentSendError(resultEffect.capability(), resultEffect.transport.transport)
246 return true
247 }
248 resultEffect.End()
249 case protocol.NavigationInventoryPublication:
250 overlays := effect.ac.overlays
251 if overlays == nil {
252 break
253 }
254 overlays.paletteMu.Lock()
255 admitted := admitPaletteInventoryPublication(overlays, message)
256 overlays.paletteMu.Unlock()
257 if admitted {
258 d.refreshPalette(effect.ac)
259 d.invalidateRender(effect.sess, effect.ac, true, "client_frame_routing.go:inventory-publication")
260 }
261 case protocol.PickerClose:
262 if effect.ac == nil || effect.ac.overlays == nil {
263 break
264 }
265 if d.closePickerForAttachment(effect.ac, effect, message.InteractionID) {
266 // The client retired the interaction: force the authoritative
267 // full paint its release path waits for before normal input and
268 // action completion resume.
269 if sess := effect.ac.currentAttachmentSession(); sess != nil {
270 d.invalidateRender(sess, effect.ac, true, "client_frame_routing.go:picker-close")
271 }
272 }
273 case protocol.PickerSelection:
274 effect.uiActionID = message.CauseActionID
275 d.resolvePickerSelection(effect, message)
276 case protocol.NavigationInventoryFailure:
277 overlays := effect.ac.overlays
278 if overlays == nil {
279 break
280 }
281 overlays.paletteMu.Lock()
282 // Failures correlate to the pending selection, which survives the
283 // selection close; the next open clears it. Unmatched failures
284 // from older interactions drop. An open palette for the same
285 // interaction also shows contextual feedback; the durable notice
286 // always records so a closed palette still surfaces the cause.
287 matched := overlays.paletteInventorySelected != nil &&
288 overlays.paletteInventorySelected.InteractionGeneration == message.InteractionGeneration &&
289 overlays.paletteInventorySelected.SourceKey == message.SourceKey &&
290 overlays.paletteInventorySelected.EntryKey == message.EntryKey
291 if overlays.palette != nil && overlays.paletteInventoryOpen &&
292 overlays.paletteInventoryInteraction == message.InteractionGeneration {
293 overlays.paletteFeedback = inventoryFailureNotice(message.Code)
294 }
295 overlays.paletteMu.Unlock()
296 if matched {
297 d.notify(effect.sess, domain.NoticeError, domain.NoticeNavigationInventory, inventoryFailureNotice(message.Code), nil)
298 }
299 d.invalidateRender(effect.sess, effect.ac, true, "client_frame_routing.go:inventory-failure")
300 case protocol.Ping:
301 if err := effect.sendControl(serverPong()); err != nil {
302 effect.End()
303 d.detachOnAttachmentSendError(effect.capability(), effect.transport.transport)
304 return true
305 }
306 }
307 return false
308}
309
310// commandResultEffect renews admission after a local navigation transition
311// ends the command's initiating effect. A retired Transport or Session means
312// the command deliberately handed off or detached and has no result carriage.
313func (d *Daemon) commandResultEffect(effect *attachmentEffect) (*attachmentEffect, bool) {
314 if effect == nil || effect.ac == nil || !effect.ac.currentTransportIs(effect.transport.transport) {
315 return nil, false
316 }
317 if effect.current() && effect.attachmentCapability.current() {
318 return effect, true
319 }
320 sess := effect.ac.currentAttachmentSession()
321 if sess == nil {
322 return nil, false
323 }
324 capability := captureAttachmentCapability(sess, effect.ac, effect.transport.transport)
325 if capability.ac == nil {
326 return nil, false
327 }
328 return effect.ac.beginAttachmentEffect(capability)
329}
330
331func (d *Daemon) executeAttachedCommand(effect *attachmentEffect, request protocol.CommandRequest) protocol.CommandResult {
332 // Every validation arm below answers with a definite failure; the
333 // success arm replaces the outcome explicitly. A zero (unspecified)
334 // outcome would be refused by the wire decoder on the client side.
335 result := protocol.CommandResult{RequestID: request.RequestID, Outcome: protocol.CommandFailed}
336 if request.Version != protocol.Version {
337 result.Code = protocol.ErrInvalidCommandArgs
338 result.Text = "unsupported command protocol version"
339 return result
340 }
341 if !request.Attached {
342 result.Code = protocol.ErrInvalidCommandArgs
343 result.Text = "attached command flag is required"
344 return result
345 }
346 if request.RequestID == 0 {
347 result.Code = protocol.ErrInvalidCommandArgs
348 result.Text = "command request id is required"
349 return result
350 }
351 if request.Self {
352 result.Code = protocol.ErrInvalidCommandArgs
353 result.Text = "attached commands cannot target themselves"
354 return result
355 }
356 if request.TargetSession != "" || request.TargetTab != "" || request.TargetPane != "" {
357 result.Code = protocol.ErrInvalidCommandArgs
358 result.Text = "attached commands cannot override their active session target"
359 return result
360 }
361 sess := effect.sess
362 if sess == nil || !effect.current() {
363 result.Code = protocol.ErrNoSuchTarget
364 result.Text = "attached session is no longer active"
365 return result
366 }
367 cmd, ok := command.BySlug(request.Slug)
368 if !ok {
369 result.Code = protocol.ErrUnknownCommand
370 result.Text = "unknown command: " + request.Slug
371 return result
372 }
373 if !cmd.PaletteVisible {
374 result.Code = protocol.ErrNotScriptable
375 result.Text = request.Slug + " is not available in the palette"
376 return result
377 }
378 if cmd.Run == nil {
379 result.Code = protocol.ErrNotScriptable
380 result.Text = request.Slug + " is not scriptable"
381 return result
382 }
383 // The frame's attachment token is the target capability. No registry/name lookup is
384 // performed, so the command cannot escape to another remote session.
385 err := sess.runMutation(func() error {
386 return cmd.Run(paletteExec{d: d, sess: sess, ac: effect.ac, effect: effect}, request.Args)
387 })
388 if err == nil {
389 result.Outcome = protocol.CommandSucceeded
390 return result
391 }
392 if errors.Is(err, command.ErrInvalidArguments) {
393 result.Code = protocol.ErrInvalidCommandArgs
394 result.Text = "usage: " + cmd.Usage
395 return result
396 }
397 result.Code = protocol.ErrInternal
398 result.Text = err.Error()
399 return result
400}
401
402// resetOutput rebases only the exact attachment and transport admitted
403// by the frame router. sendMu serializes the rebase with every output send;
404// transport revalidation under that lock rejects a link replaced while waiting.
405func (d *Daemon) resetOutput(effect *attachmentEffect) bool {
406 if !effect.current() || effect.ac == nil {
407 return false
408 }
409 ac := effect.ac
410 ac.sendMu.Lock()
411 if !effect.current() || ac.output == nil {
412 ac.sendMu.Unlock()
413 return false
414 }
415 ac.restartOutputLocked(false)
416 ac.sendMu.Unlock()
417 go d.paint(effect.sess, ac, true, effect.lease)
418 return true
419}
420
421func (d *Daemon) ackOutput(effect *attachmentEffect, epoch, state uint64) bool {
422 if !effect.current() || effect.ac == nil {
423 return false
424 }
425 acknowledged := effect.ac.ackOutputState(epoch, state)
426 if acknowledged {
427 if rc := effect.sess.renderCoordinator(); rc != nil {
428 rc.notifyAckForLease(effect.lease)
429 }
430 }
431 return acknowledged
432}