1package client
2
3import (
4 "context"
5 "time"
6
7 "github.com/bnema/vev/internal/ports"
8 "github.com/bnema/vev/internal/protocol"
9)
10
11// bindForeground runs at the existing foreground ownership boundary.
12func (u *UI) bindForeground(ctx context.Context, input *terminalInputPump, consumer uint64) uint64 {
13 u.mu.Lock()
14 defer u.mu.Unlock()
15 u.generation++
16 u.input = input
17 u.consumer = consumer
18 u.foreground = ctx
19 u.boundary = ports.UIActionResult{}
20 u.suppressed = ports.UIContext{}
21 if u.handoff != nil && u.handoff.destinationGeneration == 0 {
22 u.handoff.destinationGeneration = u.generation
23 } else if u.pending != 0 {
24 u.finishLocked(u.pending, ports.UIActionOutcomeUnknown, ports.UIActionResult{})
25 u.pending = 0
26 }
27 u.signalLocked()
28 return u.generation
29}
30
31// follow retains the accepted input action across an exact daemon navigation.
32// It waits for the destination's first committed publication, not merely a
33// successful send on the source attachment.
34func (u *UI) follow(generation, actionID uint64) bool {
35 u.mu.Lock()
36 defer u.mu.Unlock()
37 record, ok := u.records[actionID]
38 if !ok || actionID == 0 || generation != u.generation || record.Context.Generation != generation ||
39 (record.Status != ports.UIActionPending && record.Status != ports.UIActionProcessed) {
40 return false
41 }
42 if u.pending != 0 && u.pending != actionID {
43 u.finishLocked(u.pending, ports.UIActionOutcomeUnknown, ports.UIActionResult{})
44 }
45 record.Status = ports.UIActionPending
46 u.records[actionID] = record
47 u.pending = actionID
48 u.handoff = &uiActionHandoff{actionID: actionID, sourceGeneration: generation}
49 u.signalLocked()
50 return true
51}
52
53// arriveInPlace marks the destination of a followed same-peer switch as
54// committed on the unchanged foreground generation. The switch keeps the
55// attachment, so no new foreground binding names the destination: the worker
56// calls this just before publishing the destination's first output, and that
57// publication settles the action.
58func (u *UI) arriveInPlace(generation uint64) {
59 u.mu.Lock()
60 defer u.mu.Unlock()
61 if u.handoff == nil || generation != u.generation || u.handoff.sourceGeneration != generation || u.handoff.destinationGeneration != 0 {
62 return
63 }
64 u.handoff.destinationGeneration = generation
65}
66
67// failInPlace fails a followed same-peer switch the daemon refused: the source
68// attachment is unchanged and the destination never commits.
69func (u *UI) failInPlace(generation uint64) {
70 u.mu.Lock()
71 defer u.mu.Unlock()
72 if u.handoff == nil || generation != u.generation || u.handoff.sourceGeneration != generation || u.handoff.destinationGeneration != 0 {
73 return
74 }
75 u.finishLocked(u.handoff.actionID, ports.UIActionNavigationFailed, ports.UIActionResult{})
76}
77
78// releaseForeground retires exactly the binding installed for generation.
79// The supervisor owns the presentation fence after the foreground drains.
80func (u *UI) releaseForeground(generation uint64) {
81 u.mu.Lock()
82 if generation != u.generation {
83 u.mu.Unlock()
84 return
85 }
86 u.input = nil
87 u.consumer = 0
88 u.foreground = nil
89 u.boundary = ports.UIActionResult{}
90 if u.pending != 0 && (u.handoff == nil || u.handoff.actionID != u.pending) {
91 u.finishLocked(u.pending, ports.UIActionOutcomeUnknown, ports.UIActionResult{})
92 }
93 u.signalLocked()
94 u.mu.Unlock()
95}
96
97func (u *UI) accept(id, generation uint64) bool {
98 u.mu.Lock()
99 defer u.mu.Unlock()
100 if u.pending != id || u.generation != generation {
101 return false
102 }
103 if len(u.order) == uiActionHistory {
104 evicted := u.order[0]
105 delete(u.records, evicted)
106 delete(u.dispatched, evicted)
107 if done, ok := u.completion[evicted]; ok {
108 close(done)
109 delete(u.completion, evicted)
110 }
111 u.order = u.order[1:]
112 }
113 u.order = append(u.order, id)
114 u.completion[id] = make(chan struct{})
115 u.records[id] = ports.UIActionResult{ActionID: id, Accepted: true, Status: ports.UIActionPending, Context: u.reservedContext}
116 u.signalLocked()
117 return true
118}
119
120func (u *UI) finishLocked(id uint64, status ports.UIActionStatus, boundary ports.UIActionResult) {
121 record, ok := u.records[id]
122 if !ok {
123 return
124 }
125 record.Status = status
126 if status == ports.UIActionProcessed {
127 u.dispatched[id] = true
128 record.Revision = boundary.Revision
129 record.Context = boundary.Context
130 }
131 u.records[id] = record
132 if done, ok := u.completion[id]; ok {
133 close(done)
134 delete(u.completion, id)
135 }
136 if u.handoff != nil && u.handoff.actionID == id {
137 u.handoff = nil
138 }
139 if u.pending == id {
140 u.pending = 0
141 }
142 if u.overlayAction == id {
143 u.overlayAction = 0
144 }
145 u.signalLocked()
146}
147
148// published remembers only a committed output boundary, not receipt receive
149// time or an unrelated status-only revision.
150func (u *UI) published(generation uint64) {
151 snapshot, err := u.state.Snapshot()
152 if err != nil {
153 return
154 }
155 u.mu.Lock()
156 defer u.mu.Unlock()
157 if generation != u.generation || snapshot.Context.Generation != generation {
158 return
159 }
160 u.boundary = ports.UIActionResult{Revision: snapshot.Revision, Context: snapshot.Context}
161 u.suppressed = ports.UIContext{}
162 if u.handoff != nil && u.handoff.destinationGeneration == generation && snapshot.Context.Status == ports.UIStatusAttached {
163 u.handoff.boundary = u.boundary
164 if u.dispatched[u.handoff.actionID] {
165 u.finishLocked(u.handoff.actionID, ports.UIActionProcessed, u.boundary)
166 }
167 }
168 u.signalLocked()
169}
170
171func (u *UI) receipt(generation uint64, receipt protocol.UIReceipt) {
172 u.mu.Lock()
173 defer u.mu.Unlock()
174 if u.handoff != nil && u.handoff.actionID == receipt.ActionID {
175 return
176 }
177 record, ok := u.records[receipt.ActionID]
178 if !ok || record.Status != ports.UIActionPending || record.Context.Generation != generation || generation != u.generation || receipt.Validate() != nil {
179 return
180 }
181 if receipt.Outcome == protocol.UIReceiptUnavailable {
182 u.finishLocked(receipt.ActionID, ports.UIActionUnavailable, ports.UIActionResult{})
183 return
184 }
185 boundary := u.boundary
186 if boundary.Revision == 0 {
187 return
188 }
189 if receiptMatches(boundary.Context, receipt) {
190 u.finishLocked(receipt.ActionID, ports.UIActionProcessed, boundary)
191 return
192 }
193 // The client picker overlay suppressed the output that reached this
194 // boundary: it was applied, not written. Its committed context is the
195 // boundary, and the last written revision keeps later waits eligible.
196 if u.suppressed.Generation == generation && receiptMatches(u.suppressed, receipt) {
197 u.finishLocked(receipt.ActionID, ports.UIActionProcessed, ports.UIActionResult{Revision: boundary.Revision, Context: u.suppressed})
198 }
199}
200
201func receiptMatches(context ports.UIContext, receipt protocol.UIReceipt) bool {
202 return context.OutputEpoch == receipt.Epoch && context.OutputState == receipt.State && context.ViewPublication == receipt.ViewPublication
203}
204
205// suppressedOutput records the committed context of attachment output the
206// client picker overlay applied without writing, so a daemon receipt fenced
207// on it still settles its action.
208func (u *UI) suppressedOutput(generation uint64, context ports.UIContext) {
209 u.mu.Lock()
210 defer u.mu.Unlock()
211 if generation == 0 || generation != u.generation || context.Generation != generation || context.Status != ports.UIStatusAttached {
212 return
213 }
214 u.suppressed = context
215}
216
217// OverlayContext is the attached context a composition publishes while the
218// client picker is composed over the live attachment. The attachment is still
219// the actionable generation, so the picker frame keeps its session identity and
220// generation instead of an unattached Picker presentation. It reports false when
221// no attached boundary of the current foreground exists.
222func (u *UI) OverlayContext() (ports.UIContext, bool) {
223 if u == nil {
224 return ports.UIContext{}, false
225 }
226 u.mu.Lock()
227 defer u.mu.Unlock()
228 context := u.boundary.Context
229 if u.suppressed.Generation != 0 {
230 context = u.suppressed
231 }
232 if u.input == nil || context.Generation == 0 || context.Generation != u.generation || context.Status != ports.UIStatusAttached {
233 return ports.UIContext{}, false
234 }
235 return context, true
236}
237
238// markOverlayInput records that the admitted action id is about to be consumed
239// by the client picker overlay. It runs before the picker sees the bytes, so a
240// commit the read decides can always find the action to follow.
241func (u *UI) markOverlayInput(generation, id uint64) {
242 u.mu.Lock()
243 defer u.mu.Unlock()
244 if id == 0 || generation != u.generation || u.pending != id {
245 return
246 }
247 u.overlayAction = id
248}
249
250// followOverlay retains the overlay-consumed action across the attachment swap
251// a picker commit started, so it settles on the destination's first committed
252// publication.
253func (u *UI) followOverlay(generation uint64) {
254 u.mu.Lock()
255 id := u.overlayAction
256 u.mu.Unlock()
257 if id != 0 {
258 u.follow(generation, id)
259 }
260}
261
262// settleOverlay completes one overlay-consumed action at the supervisor's
263// event-loop boundary: the picker applied its input and repainted. A followed
264// action is left to the destination publication.
265func (u *UI) settleOverlay(generation, id uint64) {
266 snapshot, err := u.state.Snapshot()
267 u.mu.Lock()
268 defer u.mu.Unlock()
269 if id == 0 || u.overlayAction != id || u.pending != id || generation != u.generation {
270 return
271 }
272 if u.handoff != nil && u.handoff.actionID == id {
273 return
274 }
275 boundary := u.boundary
276 if err == nil && snapshot.Context.Generation == generation && snapshot.Context.Status == ports.UIStatusAttached {
277 boundary = ports.UIActionResult{Revision: snapshot.Revision, Context: snapshot.Context}
278 }
279 u.finishLocked(id, ports.UIActionProcessed, boundary)
280}
281
282func (u *UI) Action(ctx context.Context, request ports.UIActionRequest) (ports.UIActionResult, error) {
283 timeout, err := uiTimeout(request.Timeout)
284 if err != nil {
285 return ports.UIActionResult{}, err
286 }
287 if request.Attachment != u.handle {
288 return ports.UIActionResult{}, &ports.UIError{Code: ports.UIErrStaleAttachment}
289 }
290 u.mu.Lock()
291 available := u.input != nil
292 u.mu.Unlock()
293 if !available {
294 return ports.UIActionResult{}, &ports.UIError{Code: ports.UIErrUnavailable}
295 }
296 snapshot, err := u.state.Snapshot()
297 if err != nil {
298 return ports.UIActionResult{}, err
299 }
300 if request.Generation == 0 || snapshot.Context.Generation != request.Generation {
301 return ports.UIActionResult{}, &ports.UIError{Code: ports.UIErrStaleAttachment}
302 }
303 var data []byte
304 switch {
305 case len(request.Keys) > 0 && request.Text == "":
306 data, err = encodeUIKeys(request.Keys, snapshot.ApplicationCursor)
307 case len(request.Keys) == 0 && request.Text != "":
308 data, err = encodeUIText(request.Text)
309 default:
310 err = errUIInvalidInput
311 }
312 if err != nil {
313 return ports.UIActionResult{}, &ports.UIError{Code: ports.UIErrInvalidRequest}
314 }
315 u.mu.Lock()
316 if u.input == nil {
317 u.mu.Unlock()
318 return ports.UIActionResult{}, &ports.UIError{Code: ports.UIErrUnavailable}
319 }
320 if request.Generation != u.generation {
321 u.mu.Unlock()
322 return ports.UIActionResult{}, &ports.UIError{Code: ports.UIErrStaleAttachment}
323 }
324 if u.pending != 0 {
325 u.mu.Unlock()
326 return ports.UIActionResult{}, &ports.UIError{Code: ports.UIErrBusy}
327 }
328 if u.foreground == nil || u.foreground.Err() != nil || snapshot.Context.Status != ports.UIStatusAttached {
329 u.mu.Unlock()
330 return ports.UIActionResult{}, &ports.UIError{Code: ports.UIErrUnavailable}
331 }
332 u.nextAction++
333 id := u.nextAction
334 u.pending = id
335 u.reservedContext = snapshot.Context
336 input, consumer, foreground := u.input, u.consumer, u.foreground
337 u.mu.Unlock()
338 timer := u.clock.NewTimer(timeout)
339 defer timer.Stop()
340 admissionCtx, cancelAdmission := context.WithCancel(ctx)
341 defer cancelAdmission()
342 batch := terminalAutomationRequest{ctx: admissionCtx, owner: u, consumer: consumer, record: terminalReadResult{data: data, source: terminalInputAutomation, generation: request.Generation, actionID: id, endBatch: true, keys: request.Keys, text: request.Text}, admitted: make(chan bool, 1), dispatched: make(chan bool, 1)}
343 reject := func(code ports.UIErrorCode) (ports.UIActionResult, error) {
344 u.mu.Lock()
345 delete(u.records, id)
346 if u.pending == id {
347 u.pending = 0
348 }
349 for i, entry := range u.order {
350 if entry == id {
351 u.order = append(u.order[:i], u.order[i+1:]...)
352 break
353 }
354 }
355 u.signalLocked()
356 u.mu.Unlock()
357 return ports.UIActionResult{}, &ports.UIError{Code: code}
358 }
359 select {
360 case input.automation <- batch:
361 case <-ctx.Done():
362 return reject(ports.UIErrTimeout)
363 case <-foreground.Done():
364 return reject(ports.UIErrUnavailable)
365 case <-timer.C():
366 return reject(ports.UIErrTimeout)
367 }
368 // Once the owner receives the request it returns admission without I/O. Learn
369 // that decision even if cancellation races it; never claim rollback of input.
370 admitted := <-batch.admitted
371 if !admitted {
372 return reject(ports.UIErrInputBusy)
373 }
374 go u.trackDispatch(foreground, id, batch.dispatched)
375 for {
376 u.mu.Lock()
377 record, exists := u.records[id]
378 changed := u.changed
379 u.mu.Unlock()
380 if !exists {
381 return ports.UIActionResult{}, &ports.UIError{Code: ports.UIErrActionExpired, Accepted: true, ActionID: id}
382 }
383 if record.Status == ports.UIActionProcessed {
384 return record, nil
385 }
386 if record.Status != ports.UIActionPending {
387 return record, &ports.UIError{Code: ports.UIErrorCode(record.Status), Accepted: true, ActionID: id}
388 }
389 select {
390 case <-changed:
391 case <-ctx.Done():
392 return record, &ports.UIError{Code: ports.UIErrTimeout, Accepted: true, ActionID: id}
393 case <-timer.C():
394 return record, &ports.UIError{Code: ports.UIErrTimeout, Accepted: true, ActionID: id}
395 }
396 }
397}
398
399func (u *UI) trackDispatch(ctx context.Context, id uint64, dispatched <-chan bool) {
400 timer := u.clock.NewTimer(30 * time.Second)
401 defer timer.Stop()
402 foregroundDone := ctx.Done()
403 for {
404 u.mu.Lock()
405 record, exists := u.records[id]
406 changed := u.changed
407 u.mu.Unlock()
408 if !exists || record.Status != ports.UIActionPending {
409 return
410 }
411 var status ports.UIActionStatus
412 select {
413 case ok := <-dispatched:
414 if !ok {
415 status = ports.UIActionOutcomeUnknown
416 } else {
417 u.mu.Lock()
418 if _, exists := u.records[id]; exists {
419 u.dispatched[id] = true
420 if u.handoff != nil && u.handoff.boundary.Revision != 0 {
421 u.finishLocked(u.handoff.actionID, ports.UIActionProcessed, u.handoff.boundary)
422 }
423 }
424 u.mu.Unlock()
425 }
426 dispatched = nil
427 case <-foregroundDone:
428 u.mu.Lock()
429 following := u.handoff != nil && u.handoff.actionID == id
430 u.mu.Unlock()
431 if following {
432 foregroundDone = nil
433 } else {
434 status = ports.UIActionOutcomeUnknown
435 }
436 case <-timer.C():
437 status = ports.UIActionUnavailable
438 case <-changed:
439 }
440 if status != "" {
441 u.mu.Lock()
442 if current := u.records[id]; current.Status == ports.UIActionPending {
443 u.finishLocked(id, status, ports.UIActionResult{})
444 }
445 u.mu.Unlock()
446 return
447 }
448 }
449}
450
451var _ ports.UIService = (*UI)(nil)