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.RecentRouteSnapshot:
192 replayIdentity := effect.ac.setRouteSnapshot(message)
193 d.pokeAttentionTicker()
194 d.invalidateRender(effect.sess, effect.ac, false, "client_frame_routing.go:route-snapshot")
195 if !replayIdentity {
196 break
197 }
198 if err := d.sendCommittedRouteIdentityForAttachment(effect); err != nil {
199 d.detachOnAttachmentSendError(effect.capability(), effect.transport.transport)
200 }
201 case protocol.RouteAttentionSubscription:
202 effect.ac.setRouteAttentionSubscription(message, effect.transport, d.clock.Now())
203 d.reconcileRouteHistory(effect.ac, effect)
204 d.invalidateRender(effect.sess, effect.ac, false, "client_frame_routing.go:route-attention")
205 case protocol.SessionCreationFailure:
206 notice := "couldn't create session"
207 switch message.Code {
208 case protocol.RouteFailureStaleSelection, protocol.RouteFailureNoSuchRoute:
209 notice = "that destination is no longer available"
210 case protocol.RouteFailureTargetChanged:
211 notice = "the destination returned an unexpected session"
212 case protocol.RouteFailureOriginUnavailable:
213 notice = "couldn't restore the original session after creation failed"
214 case protocol.RouteFailureUnavailable:
215 notice = "the selected destination is unavailable"
216 }
217 d.notify(effect.sess, domain.NoticeWarn, domain.NoticeSessionUnavailable, notice, nil)
218 case protocol.RouteNavigationFailure:
219 notice := "route navigation failed"
220 switch message.Code {
221 case protocol.RouteFailureStaleSelection:
222 notice = "that recent route is no longer available"
223 case protocol.RouteFailureTargetChanged:
224 notice = "that recent route changed before attach"
225 case protocol.RouteFailureOriginUnavailable:
226 notice = "the original route is no longer available"
227 case protocol.RouteFailureUnavailable:
228 notice = "the selected route is unavailable"
229 case protocol.RouteFailureNoSuchRoute:
230 notice = "that recent route no longer exists"
231 }
232 d.notify(effect.sess, domain.NoticeWarn, domain.NoticeSessionUnavailable, notice, nil)
233 case protocol.CommandRequest:
234 result := d.executeAttachedCommand(effect, message)
235 resultEffect, ok := d.commandResultEffect(effect)
236 if !ok {
237 // A remote handoff or explicit detach deliberately retired the old
238 // attachment; it must not be treated as a stale-result send error.
239 return true
240 }
241 if err := resultEffect.sendControl(serverCommandResult(result)); err != nil {
242 resultEffect.End()
243 d.detachOnAttachmentSendError(resultEffect.capability(), resultEffect.transport.transport)
244 return true
245 }
246 resultEffect.End()
247 case protocol.NavigationInventoryPublication:
248 overlays := effect.ac.overlays
249 if overlays == nil {
250 break
251 }
252 overlays.paletteMu.Lock()
253 admitted := admitPaletteInventoryPublication(overlays, message)
254 overlays.paletteMu.Unlock()
255 if admitted {
256 d.refreshPalette(effect.ac)
257 d.invalidateRender(effect.sess, effect.ac, true, "client_frame_routing.go:inventory-publication")
258 }
259 case protocol.PickerClose:
260 if effect.ac == nil || effect.ac.overlays == nil {
261 break
262 }
263 if d.closePickerForAttachment(effect.ac, effect, message.InteractionID) {
264 // The client retired the interaction: force the authoritative
265 // full paint its release path waits for before normal input and
266 // action completion resume.
267 if sess := effect.ac.currentAttachmentSession(); sess != nil {
268 d.invalidateRender(sess, effect.ac, true, "client_frame_routing.go:picker-close")
269 }
270 }
271 case protocol.PickerSelection:
272 effect.uiActionID = message.CauseActionID
273 d.resolvePickerSelection(effect, message)
274 case protocol.NavigationInventoryFailure:
275 overlays := effect.ac.overlays
276 if overlays == nil {
277 break
278 }
279 overlays.paletteMu.Lock()
280 // Failures correlate to the pending selection, which survives the
281 // selection close; the next open clears it. Unmatched failures
282 // from older interactions drop. An open palette for the same
283 // interaction also shows contextual feedback; the durable notice
284 // always records so a closed palette still surfaces the cause.
285 matched := overlays.paletteInventorySelected != nil &&
286 overlays.paletteInventorySelected.InteractionGeneration == message.InteractionGeneration &&
287 overlays.paletteInventorySelected.SourceKey == message.SourceKey &&
288 overlays.paletteInventorySelected.EntryKey == message.EntryKey
289 if overlays.palette != nil && overlays.paletteInventoryOpen &&
290 overlays.paletteInventoryInteraction == message.InteractionGeneration {
291 overlays.paletteFeedback = inventoryFailureNotice(message.Code)
292 }
293 overlays.paletteMu.Unlock()
294 if matched {
295 d.notify(effect.sess, domain.NoticeError, domain.NoticeNavigationInventory, inventoryFailureNotice(message.Code), nil)
296 }
297 d.invalidateRender(effect.sess, effect.ac, true, "client_frame_routing.go:inventory-failure")
298 case protocol.Ping:
299 if err := effect.sendControl(serverPong()); err != nil {
300 effect.End()
301 d.detachOnAttachmentSendError(effect.capability(), effect.transport.transport)
302 return true
303 }
304 }
305 return false
306}
307
308// commandResultEffect renews admission after a local navigation transition
309// ends the command's initiating effect. A retired Transport or Session means
310// the command deliberately handed off or detached and has no result carriage.
311func (d *Daemon) commandResultEffect(effect *attachmentEffect) (*attachmentEffect, bool) {
312 if effect == nil || effect.ac == nil || !effect.ac.currentTransportIs(effect.transport.transport) {
313 return nil, false
314 }
315 if effect.current() && effect.attachmentCapability.current() {
316 return effect, true
317 }
318 sess := effect.ac.currentAttachmentSession()
319 if sess == nil {
320 return nil, false
321 }
322 capability := captureAttachmentCapability(sess, effect.ac, effect.transport.transport)
323 if capability.ac == nil {
324 return nil, false
325 }
326 return effect.ac.beginAttachmentEffect(capability)
327}
328
329func (d *Daemon) executeAttachedCommand(effect *attachmentEffect, request protocol.CommandRequest) protocol.CommandResult {
330 // Every validation arm below answers with a definite failure; the
331 // success arm replaces the outcome explicitly. A zero (unspecified)
332 // outcome would be refused by the wire decoder on the client side.
333 result := protocol.CommandResult{RequestID: request.RequestID, Outcome: protocol.CommandFailed}
334 if request.Version != protocol.Version {
335 result.Code = protocol.ErrInvalidCommandArgs
336 result.Text = "unsupported command protocol version"
337 return result
338 }
339 if !request.Attached {
340 result.Code = protocol.ErrInvalidCommandArgs
341 result.Text = "attached command flag is required"
342 return result
343 }
344 if request.RequestID == 0 {
345 result.Code = protocol.ErrInvalidCommandArgs
346 result.Text = "command request id is required"
347 return result
348 }
349 if request.Self {
350 result.Code = protocol.ErrInvalidCommandArgs
351 result.Text = "attached commands cannot target themselves"
352 return result
353 }
354 if request.TargetSession != "" || request.TargetTab != "" || request.TargetPane != "" {
355 result.Code = protocol.ErrInvalidCommandArgs
356 result.Text = "attached commands cannot override their active session target"
357 return result
358 }
359 sess := effect.sess
360 if sess == nil || !effect.current() {
361 result.Code = protocol.ErrNoSuchTarget
362 result.Text = "attached session is no longer active"
363 return result
364 }
365 cmd, ok := command.BySlug(request.Slug)
366 if !ok {
367 result.Code = protocol.ErrUnknownCommand
368 result.Text = "unknown command: " + request.Slug
369 return result
370 }
371 if !cmd.PaletteVisible {
372 result.Code = protocol.ErrNotScriptable
373 result.Text = request.Slug + " is not available in the palette"
374 return result
375 }
376 if cmd.Run == nil {
377 result.Code = protocol.ErrNotScriptable
378 result.Text = request.Slug + " is not scriptable"
379 return result
380 }
381 // The frame's attachment token is the target capability. No registry/name lookup is
382 // performed, so the command cannot escape to another remote session.
383 err := sess.runMutation(func() error {
384 return cmd.Run(paletteExec{d: d, sess: sess, ac: effect.ac, effect: effect}, request.Args)
385 })
386 if err == nil {
387 result.Outcome = protocol.CommandSucceeded
388 return result
389 }
390 if errors.Is(err, command.ErrInvalidArguments) {
391 result.Code = protocol.ErrInvalidCommandArgs
392 result.Text = "usage: " + cmd.Usage
393 return result
394 }
395 result.Code = protocol.ErrInternal
396 result.Text = err.Error()
397 return result
398}
399
400// resetOutput rebases only the exact attachment and transport admitted
401// by the frame router. sendMu serializes the rebase with every output send;
402// transport revalidation under that lock rejects a link replaced while waiting.
403func (d *Daemon) resetOutput(effect *attachmentEffect) bool {
404 if !effect.current() || effect.ac == nil {
405 return false
406 }
407 ac := effect.ac
408 ac.sendMu.Lock()
409 if !effect.current() || ac.output == nil {
410 ac.sendMu.Unlock()
411 return false
412 }
413 ac.restartOutputLocked(false)
414 ac.sendMu.Unlock()
415 go d.paint(effect.sess, ac, true, effect.lease)
416 return true
417}
418
419func (d *Daemon) ackOutput(effect *attachmentEffect, epoch, state uint64) bool {
420 if !effect.current() || effect.ac == nil {
421 return false
422 }
423 acknowledged := effect.ac.ackOutputState(epoch, state)
424 if acknowledged {
425 if rc := effect.sess.renderCoordinator(); rc != nil {
426 rc.notifyAckForLease(effect.lease)
427 }
428 }
429 return acknowledged
430}