feat(client): bind UI actions in the supervisor
- commit
- 2bcc916cdaa7031e99b7442fd1776b261290d2dd
- author
- brice · · 2026-09-19 19:22 UTC
- parent
- 6dee53f
- signature
- GPG Verified
CD88F9E564E0102646FB6077A1A07C7C7C3E349D· arch-gpg Verified for the committer email
7 files · +668 -61
M internal/app/broker_offline_client.go +9 -11
| @@ -81,7 +81,7 @@ func runBrokerClientCommand(ctx context.Context, options brokerClientOptions) er | |||
| 81 | 81 | log := slog.New(slog.NewTextHandler(io.Discard, nil)) | |
| 82 | 82 | switch options.harness { | |
| 83 | 83 | case offlineClientTerminal: | |
| 84 | - | return runOfflineClient(ctx, brokeripc.SocketPath(layout.Runtime), term.New(), log, nil, nil) | |
| 84 | + | return runOfflineClient(ctx, brokeripc.SocketPath(layout.Runtime), term.New(), nil, log, nil, nil) | |
| 85 | 85 | case offlineClientUIDriver: | |
| 86 | 86 | terminal, err := uiterm.New(ctx, domain.Geometry{Size: domain.Size{Cols: uiDriverDefaultColumns, Rows: uiDriverDefaultRows}}, "") | |
| 87 | 87 | if err != nil { | |
| @@ -99,7 +99,7 @@ func runBrokerClientCommand(ctx context.Context, options brokerClientOptions) er | |||
| 99 | 99 | // physical terminal, headless UI driver, and browser terminal harnesses. The | |
| 100 | 100 | // connector is the real broker IPC adapter and InitialNavigation opens a real | |
| 101 | 101 | // broker logical stream; no direct daemon dialer is present. | |
| 102 | - | func runOfflineClient(ctx context.Context, socket string, terminal ports.Terminal, log *slog.Logger, onState func(client.State), onFailure func(error)) error { | |
| 102 | + | func runOfflineClient(ctx context.Context, socket string, terminal ports.Terminal, ui *client.UI, log *slog.Logger, onState func(client.State), onFailure func(error)) error { | |
| 103 | 103 | if terminal == nil { | |
| 104 | 104 | return errors.New("vev: offline client requires a terminal") | |
| 105 | 105 | } | |
| @@ -140,6 +140,7 @@ func runOfflineClient(ctx context.Context, socket string, terminal ports.Termina | |||
| 140 | 140 | Connector: brokeripc.NewConnector(socket, brokeripc.Config{}), | |
| 141 | 141 | Terminal: terminal, | |
| 142 | 142 | Clock: clk, | |
| 143 | + | UI: ui, | |
| 143 | 144 | Picker: picker, | |
| 144 | 145 | InitialNavigation: &navigation, | |
| 145 | 146 | AttachmentEnvironment: client.AttachmentEnvironment{ | |
| @@ -175,7 +176,7 @@ func runOfflineUIDriver(ctx context.Context, socket string, terminal *uiterm.Ter | |||
| 175 | 176 | runnerDone := make(chan error, 1) | |
| 176 | 177 | states := make(chan client.State, 1) | |
| 177 | 178 | go func() { | |
| 178 | - | runnerDone <- runOfflineClient(runCtx, socket, terminal, log, func(state client.State) { | |
| 179 | + | runnerDone <- runOfflineClient(runCtx, socket, terminal, ui, log, func(state client.State) { | |
| 179 | 180 | select { | |
| 180 | 181 | case states <- state: | |
| 181 | 182 | default: | |
| @@ -183,20 +184,17 @@ func runOfflineUIDriver(ctx context.Context, socket string, terminal *uiterm.Ter | |||
| 183 | 184 | }, nil) | |
| 184 | 185 | }() | |
| 185 | 186 | ||
| 186 | - | var state client.State | |
| 187 | 187 | select { | |
| 188 | - | case state = <-states: | |
| 188 | + | case <-states: | |
| 189 | 189 | case err := <-runnerDone: | |
| 190 | 190 | return err | |
| 191 | 191 | case <-runCtx.Done(): | |
| 192 | 192 | return runCtx.Err() | |
| 193 | 193 | } | |
| 194 | - | // The autonomous supervisor publishes terminal transactions but does not own | |
| 195 | - | // the ordinary Runner's UI action-binding loop. Reconnecting is therefore | |
| 196 | - | // the honest discovery status, and Control is false: advertising control | |
| 197 | - | // would invite actions that no foreground binding can accept. Generation is | |
| 198 | - | // the real supervisor attempt generation rather than a fabricated constant. | |
| 199 | - | ready := uidriver.Ready{Attachment: ui.Handle(), Generation: state.Generation, Control: false, Status: ports.UIStatusReconnecting} | |
| 194 | + | // Ready advertises the currently actionable UI generation. Before the first | |
| 195 | + | // attachment there is no actionable generation; attached snapshots publish | |
| 196 | + | // the non-zero generation clients must fence actions against. | |
| 197 | + | ready := uidriver.Ready{Attachment: ui.Handle(), Control: true, Status: ports.UIStatusReconnecting} | |
| 200 | 198 | server := uidriver.New(ui, clock.New()) | |
| 201 | 199 | serveDone := make(chan error, 1) | |
| 202 | 200 | go func() { serveDone <- server.Serve(runCtx, stream, ready) }() | |
| @@ -81,7 +81,7 @@ func runBrokerClientCommand(ctx context.Context, options brokerClientOptions) er | |||||
| 81 | log := slog.New(slog.NewTextHandler(io.Discard, nil)) | 81 | log := slog.New(slog.NewTextHandler(io.Discard, nil)) | ||
| 82 | switch options.harness { | 82 | switch options.harness { | ||
| 83 | case offlineClientTerminal: | 83 | case offlineClientTerminal: | ||
| 84 | - | return runOfflineClient(ctx, brokeripc.SocketPath(layout.Runtime), term.New(), log, nil, nil) | 84 | + | return runOfflineClient(ctx, brokeripc.SocketPath(layout.Runtime), term.New(), nil, log, nil, nil) |
| 85 | case offlineClientUIDriver: | 85 | case offlineClientUIDriver: | ||
| 86 | terminal, err := uiterm.New(ctx, domain.Geometry{Size: domain.Size{Cols: uiDriverDefaultColumns, Rows: uiDriverDefaultRows}}, "") | 86 | terminal, err := uiterm.New(ctx, domain.Geometry{Size: domain.Size{Cols: uiDriverDefaultColumns, Rows: uiDriverDefaultRows}}, "") | ||
| 87 | if err != nil { | 87 | if err != nil { | ||
| @@ -99,7 +99,7 @@ func runBrokerClientCommand(ctx context.Context, options brokerClientOptions) er | |||||
| 99 | // physical terminal, headless UI driver, and browser terminal harnesses. The | 99 | // physical terminal, headless UI driver, and browser terminal harnesses. The | ||
| 100 | // connector is the real broker IPC adapter and InitialNavigation opens a real | 100 | // connector is the real broker IPC adapter and InitialNavigation opens a real | ||
| 101 | // broker logical stream; no direct daemon dialer is present. | 101 | // broker logical stream; no direct daemon dialer is present. | ||
| 102 | - | func runOfflineClient(ctx context.Context, socket string, terminal ports.Terminal, log *slog.Logger, onState func(client.State), onFailure func(error)) error { | 102 | + | func runOfflineClient(ctx context.Context, socket string, terminal ports.Terminal, ui *client.UI, log *slog.Logger, onState func(client.State), onFailure func(error)) error { |
| 103 | if terminal == nil { | 103 | if terminal == nil { | ||
| 104 | return errors.New("vev: offline client requires a terminal") | 104 | return errors.New("vev: offline client requires a terminal") | ||
| 105 | } | 105 | } | ||
| @@ -140,6 +140,7 @@ func runOfflineClient(ctx context.Context, socket string, terminal ports.Termina | |||||
| 140 | Connector: brokeripc.NewConnector(socket, brokeripc.Config{}), | 140 | Connector: brokeripc.NewConnector(socket, brokeripc.Config{}), | ||
| 141 | Terminal: terminal, | 141 | Terminal: terminal, | ||
| 142 | Clock: clk, | 142 | Clock: clk, | ||
| 143 | + | UI: ui, | |||
| 143 | Picker: picker, | 144 | Picker: picker, | ||
| 144 | InitialNavigation: &navigation, | 145 | InitialNavigation: &navigation, | ||
| 145 | AttachmentEnvironment: client.AttachmentEnvironment{ | 146 | AttachmentEnvironment: client.AttachmentEnvironment{ | ||
| @@ -175,7 +176,7 @@ func runOfflineUIDriver(ctx context.Context, socket string, terminal *uiterm.Ter | |||||
| 175 | runnerDone := make(chan error, 1) | 176 | runnerDone := make(chan error, 1) | ||
| 176 | states := make(chan client.State, 1) | 177 | states := make(chan client.State, 1) | ||
| 177 | go func() { | 178 | go func() { | ||
| 178 | - | runnerDone <- runOfflineClient(runCtx, socket, terminal, log, func(state client.State) { | 179 | + | runnerDone <- runOfflineClient(runCtx, socket, terminal, ui, log, func(state client.State) { |
| 179 | select { | 180 | select { | ||
| 180 | case states <- state: | 181 | case states <- state: | ||
| 181 | default: | 182 | default: | ||
| @@ -183,20 +184,17 @@ func runOfflineUIDriver(ctx context.Context, socket string, terminal *uiterm.Ter | |||||
| 183 | }, nil) | 184 | }, nil) | ||
| 184 | }() | 185 | }() | ||
| 185 | 186 | ||||
| 186 | - | var state client.State | |||
| 187 | select { | 187 | select { | ||
| 188 | - | case state = <-states: | 188 | + | case <-states: |
| 189 | case err := <-runnerDone: | 189 | case err := <-runnerDone: | ||
| 190 | return err | 190 | return err | ||
| 191 | case <-runCtx.Done(): | 191 | case <-runCtx.Done(): | ||
| 192 | return runCtx.Err() | 192 | return runCtx.Err() | ||
| 193 | } | 193 | } | ||
| 194 | - | // The autonomous supervisor publishes terminal transactions but does not own | 194 | + | // Ready advertises the currently actionable UI generation. Before the first |
| 195 | - | // the ordinary Runner's UI action-binding loop. Reconnecting is therefore | 195 | + | // attachment there is no actionable generation; attached snapshots publish |
| 196 | - | // the honest discovery status, and Control is false: advertising control | 196 | + | // the non-zero generation clients must fence actions against. |
| 197 | - | // would invite actions that no foreground binding can accept. Generation is | 197 | + | ready := uidriver.Ready{Attachment: ui.Handle(), Control: true, Status: ports.UIStatusReconnecting} |
| 198 | - | // the real supervisor attempt generation rather than a fabricated constant. | |||
| 199 | - | ready := uidriver.Ready{Attachment: ui.Handle(), Generation: state.Generation, Control: false, Status: ports.UIStatusReconnecting} | |||
| 200 | server := uidriver.New(ui, clock.New()) | 198 | server := uidriver.New(ui, clock.New()) | ||
| 201 | serveDone := make(chan error, 1) | 199 | serveDone := make(chan error, 1) | ||
| 202 | go func() { serveDone <- server.Serve(runCtx, stream, ready) }() | 200 | go func() { serveDone <- server.Serve(runCtx, stream, ready) }() | ||
M internal/app/broker_offline_client_test.go +5 -5
| @@ -242,7 +242,7 @@ func TestOfflineClientTerminalReachesSessionThroughBrokerIPC(t *testing.T) { | |||
| 242 | 242 | done := make(chan error, 1) | |
| 243 | 243 | failures := make(chan error, 1) | |
| 244 | 244 | go func() { | |
| 245 | - | done <- runOfflineClient(ctx, fixture.socket, terminal, discardLog(), nil, func(err error) { | |
| 245 | + | done <- runOfflineClient(ctx, fixture.socket, terminal, nil, discardLog(), nil, func(err error) { | |
| 246 | 246 | select { | |
| 247 | 247 | case failures <- err: | |
| 248 | 248 | default: | |
| @@ -294,9 +294,9 @@ func TestOfflineClientUIDriverReachesSessionThroughBrokerIPC(t *testing.T) { | |||
| 294 | 294 | case ready := <-readyResult: | |
| 295 | 295 | result, ok := ready["result"].(map[string]any) | |
| 296 | 296 | require.True(t, ok) | |
| 297 | - | require.Equal(t, false, result["control"], "autonomous supervisor has no UI action binding") | |
| 298 | - | require.Equal(t, string(ports.UIStatusReconnecting), result["status"], "reconnecting is the honest discovery status until autonomous UI binding exists") | |
| 299 | - | require.NotZero(t, result["generation"], "ready carries the real supervisor generation") | |
| 297 | + | require.Equal(t, true, result["control"], "autonomous supervisor binds UI actions") | |
| 298 | + | require.Equal(t, string(ports.UIStatusReconnecting), result["status"], "ready precedes the attachment publication") | |
| 299 | + | require.Equal(t, float64(0), result["generation"], "ready has no actionable generation before attachment publication") | |
| 300 | 300 | case err := <-readyError: | |
| 301 | 301 | t.Fatalf("decode UI-driver ready: %v", err) | |
| 302 | 302 | case <-time.After(brokerTestWait): | |
| @@ -512,7 +512,7 @@ func TestOfflineClientBrowserReachesSessionThroughBrokerIPC(t *testing.T) { | |||
| 512 | 512 | defer cancel() | |
| 513 | 513 | token := strings.Repeat("b", 43) | |
| 514 | 514 | handler, err := webterm.NewServer(ctx, webterm.Settings{}, token, func(runCtx context.Context, terminal *webterm.Terminal) error { | |
| 515 | - | return runOfflineClient(runCtx, fixture.socket, terminal, discardLog(), nil, nil) | |
| 515 | + | return runOfflineClient(runCtx, fixture.socket, terminal, nil, discardLog(), nil, nil) | |
| 516 | 516 | }) | |
| 517 | 517 | require.NoError(t, err) | |
| 518 | 518 | server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { r.Host = webterm.Address; handler.ServeHTTP(w, r) })) | |
| @@ -242,7 +242,7 @@ func TestOfflineClientTerminalReachesSessionThroughBrokerIPC(t *testing.T) { | |||||
| 242 | done := make(chan error, 1) | 242 | done := make(chan error, 1) | ||
| 243 | failures := make(chan error, 1) | 243 | failures := make(chan error, 1) | ||
| 244 | go func() { | 244 | go func() { | ||
| 245 | - | done <- runOfflineClient(ctx, fixture.socket, terminal, discardLog(), nil, func(err error) { | 245 | + | done <- runOfflineClient(ctx, fixture.socket, terminal, nil, discardLog(), nil, func(err error) { |
| 246 | select { | 246 | select { | ||
| 247 | case failures <- err: | 247 | case failures <- err: | ||
| 248 | default: | 248 | default: | ||
| @@ -294,9 +294,9 @@ func TestOfflineClientUIDriverReachesSessionThroughBrokerIPC(t *testing.T) { | |||||
| 294 | case ready := <-readyResult: | 294 | case ready := <-readyResult: | ||
| 295 | result, ok := ready["result"].(map[string]any) | 295 | result, ok := ready["result"].(map[string]any) | ||
| 296 | require.True(t, ok) | 296 | require.True(t, ok) | ||
| 297 | - | require.Equal(t, false, result["control"], "autonomous supervisor has no UI action binding") | 297 | + | require.Equal(t, true, result["control"], "autonomous supervisor binds UI actions") |
| 298 | - | require.Equal(t, string(ports.UIStatusReconnecting), result["status"], "reconnecting is the honest discovery status until autonomous UI binding exists") | 298 | + | require.Equal(t, string(ports.UIStatusReconnecting), result["status"], "ready precedes the attachment publication") |
| 299 | - | require.NotZero(t, result["generation"], "ready carries the real supervisor generation") | 299 | + | require.Equal(t, float64(0), result["generation"], "ready has no actionable generation before attachment publication") |
| 300 | case err := <-readyError: | 300 | case err := <-readyError: | ||
| 301 | t.Fatalf("decode UI-driver ready: %v", err) | 301 | t.Fatalf("decode UI-driver ready: %v", err) | ||
| 302 | case <-time.After(brokerTestWait): | 302 | case <-time.After(brokerTestWait): | ||
| @@ -512,7 +512,7 @@ func TestOfflineClientBrowserReachesSessionThroughBrokerIPC(t *testing.T) { | |||||
| 512 | defer cancel() | 512 | defer cancel() | ||
| 513 | token := strings.Repeat("b", 43) | 513 | token := strings.Repeat("b", 43) | ||
| 514 | handler, err := webterm.NewServer(ctx, webterm.Settings{}, token, func(runCtx context.Context, terminal *webterm.Terminal) error { | 514 | handler, err := webterm.NewServer(ctx, webterm.Settings{}, token, func(runCtx context.Context, terminal *webterm.Terminal) error { | ||
| 515 | - | return runOfflineClient(runCtx, fixture.socket, terminal, discardLog(), nil, nil) | 515 | + | return runOfflineClient(runCtx, fixture.socket, terminal, nil, discardLog(), nil, nil) |
| 516 | }) | 516 | }) | ||
| 517 | require.NoError(t, err) | 517 | require.NoError(t, err) | ||
| 518 | server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { r.Host = webterm.Address; handler.ServeHTTP(w, r) })) | 518 | server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { r.Host = webterm.Address; handler.ServeHTTP(w, r) })) | ||
M internal/usecase/client/attachment_session.go +45 -3
| @@ -178,7 +178,7 @@ func (w *sessionAttachmentWorker) awaitInitialPublication(ctx context.Context, f | |||
| 178 | 178 | return state, &event | |
| 179 | 179 | } | |
| 180 | 180 | state = next | |
| 181 | - | if err := fg.Output(state.uiContext(ports.UIContext{Generation: token.Generation}, ports.UIStatusTransitioning), typed.Data); err != nil { | |
| 181 | + | if err := fg.Output(state.uiContext(ports.UIContext{Generation: attachmentActionableGeneration(fg, token)}, ports.UIStatusTransitioning), typed.Data); err != nil { | |
| 182 | 182 | event := AttachmentEvent{Token: token, Kind: AttachmentEventFailed, Err: fmt.Errorf("vev: publishing initial output: %w", err)} | |
| 183 | 183 | return state, &event | |
| 184 | 184 | } | |
| @@ -233,6 +233,7 @@ func (w *sessionAttachmentWorker) pumpAttached(ctx context.Context, fg Attachmen | |||
| 233 | 233 | go pumpAttachmentInput(pumpCtx, fg, outgoing) | |
| 234 | 234 | go pumpAttachmentGeometry(pumpCtx, fg, outgoing) | |
| 235 | 235 | go pumpAttachmentLifecycle(pumpCtx, fg, outgoing) | |
| 236 | + | outputResetRequested := false | |
| 236 | 237 | for { | |
| 237 | 238 | select { | |
| 238 | 239 | case result := <-incoming: | |
| @@ -246,9 +247,27 @@ func (w *sessionAttachmentWorker) pumpAttached(ctx context.Context, fg Attachmen | |||
| 246 | 247 | continue | |
| 247 | 248 | } | |
| 248 | 249 | state = next | |
| 249 | - | if err := fg.Output(state.uiContext(ports.UIContext{Generation: token.Generation}, ports.UIStatusAttached), typed.Data); err != nil { | |
| 250 | + | outputResetRequested = false | |
| 251 | + | if err := fg.Output(state.uiContext(ports.UIContext{Generation: attachmentActionableGeneration(fg, token)}, ports.UIStatusAttached), typed.Data); err != nil { | |
| 250 | 252 | return AttachmentEvent{Token: token, Kind: AttachmentEventFailed, Err: fmt.Errorf("vev: publishing output: %w", err)} | |
| 251 | 253 | } | |
| 254 | + | case protocol.UIViewUpdate: | |
| 255 | + | next, accepted, needsReset := state.nextView(typed) | |
| 256 | + | if needsReset && !outputResetRequested { | |
| 257 | + | if err := w.send(ctx, fg, stream, protocol.OutputResetRequest{}); err != nil { | |
| 258 | + | return w.settle(ctx, fg, stream, token, err) | |
| 259 | + | } | |
| 260 | + | outputResetRequested = true | |
| 261 | + | } | |
| 262 | + | if !accepted { | |
| 263 | + | continue | |
| 264 | + | } | |
| 265 | + | state = next | |
| 266 | + | if err := fg.Output(state.uiContext(ports.UIContext{Generation: attachmentActionableGeneration(fg, token)}, ports.UIStatusAttached), nil); err != nil { | |
| 267 | + | return AttachmentEvent{Token: token, Kind: AttachmentEventFailed, Err: fmt.Errorf("vev: publishing view update: %w", err)} | |
| 268 | + | } | |
| 269 | + | case protocol.UIReceipt: | |
| 270 | + | attachmentUIReceipt(fg, typed) | |
| 252 | 271 | case protocol.ErrorMsg: | |
| 253 | 272 | return AttachmentEvent{Token: token, Kind: AttachmentEventFailed, Err: &ProtocolError{Code: typed.Code, Text: typed.Text}} | |
| 254 | 273 | case protocol.Detached: | |
| @@ -267,6 +286,11 @@ func (w *sessionAttachmentWorker) pumpAttached(ctx context.Context, fg Attachmen | |||
| 267 | 286 | } | |
| 268 | 287 | if event.input != nil { | |
| 269 | 288 | fg.AckInput() | |
| 289 | + | if event.input.actionID != 0 { | |
| 290 | + | if err := w.send(ctx, fg, stream, protocol.UIFence{ActionID: event.input.actionID}); err != nil { | |
| 291 | + | return w.settle(ctx, fg, stream, token, err) | |
| 292 | + | } | |
| 293 | + | } | |
| 270 | 294 | } | |
| 271 | 295 | if event.action == AttachmentDetachToPicker { | |
| 272 | 296 | return AttachmentEvent{Token: token, Kind: AttachmentEventEnded, Err: errDetachToPicker} | |
| @@ -282,6 +306,24 @@ func (w *sessionAttachmentWorker) pumpAttached(ctx context.Context, fg Attachmen | |||
| 282 | 306 | } | |
| 283 | 307 | } | |
| 284 | 308 | ||
| 309 | + | type attachmentUIForeground interface { | |
| 310 | + | actionableGeneration() uint64 | |
| 311 | + | uiReceipt(protocol.UIReceipt) | |
| 312 | + | } | |
| 313 | + | ||
| 314 | + | func attachmentActionableGeneration(fg AttachmentForeground, token AttachmentToken) uint64 { | |
| 315 | + | if ui, ok := fg.(attachmentUIForeground); ok { | |
| 316 | + | return ui.actionableGeneration() | |
| 317 | + | } | |
| 318 | + | return token.Generation | |
| 319 | + | } | |
| 320 | + | ||
| 321 | + | func attachmentUIReceipt(fg AttachmentForeground, receipt protocol.UIReceipt) { | |
| 322 | + | if ui, ok := fg.(attachmentUIForeground); ok { | |
| 323 | + | ui.uiReceipt(receipt) | |
| 324 | + | } | |
| 325 | + | } | |
| 326 | + | ||
| 285 | 327 | func pumpAttachmentInput(ctx context.Context, fg AttachmentForeground, out chan<- attachmentClientEvent) { | |
| 286 | 328 | var sequence uint64 | |
| 287 | 329 | for { | |
| @@ -300,7 +342,7 @@ func pumpAttachmentInput(ctx context.Context, fg AttachmentForeground, out chan< | |||
| 300 | 342 | sequence++ | |
| 301 | 343 | copyEvent := event | |
| 302 | 344 | select { | |
| 303 | - | case out <- attachmentClientEvent{message: protocol.Input{InputSeq: sequence, Data: append([]byte(nil), event.Data...)}, input: ©Event}: | |
| 345 | + | case out <- attachmentClientEvent{message: protocol.Input{InputSeq: sequence, ActionID: event.actionID, Data: append([]byte(nil), event.Data...)}, input: ©Event}: | |
| 304 | 346 | case <-ctx.Done(): | |
| 305 | 347 | fg.PreserveInput(event.Data) | |
| 306 | 348 | return | |
| @@ -178,7 +178,7 @@ func (w *sessionAttachmentWorker) awaitInitialPublication(ctx context.Context, f | |||||
| 178 | return state, &event | 178 | return state, &event | ||
| 179 | } | 179 | } | ||
| 180 | state = next | 180 | state = next | ||
| 181 | - | if err := fg.Output(state.uiContext(ports.UIContext{Generation: token.Generation}, ports.UIStatusTransitioning), typed.Data); err != nil { | 181 | + | if err := fg.Output(state.uiContext(ports.UIContext{Generation: attachmentActionableGeneration(fg, token)}, ports.UIStatusTransitioning), typed.Data); err != nil { |
| 182 | event := AttachmentEvent{Token: token, Kind: AttachmentEventFailed, Err: fmt.Errorf("vev: publishing initial output: %w", err)} | 182 | event := AttachmentEvent{Token: token, Kind: AttachmentEventFailed, Err: fmt.Errorf("vev: publishing initial output: %w", err)} | ||
| 183 | return state, &event | 183 | return state, &event | ||
| 184 | } | 184 | } | ||
| @@ -233,6 +233,7 @@ func (w *sessionAttachmentWorker) pumpAttached(ctx context.Context, fg Attachmen | |||||
| 233 | go pumpAttachmentInput(pumpCtx, fg, outgoing) | 233 | go pumpAttachmentInput(pumpCtx, fg, outgoing) | ||
| 234 | go pumpAttachmentGeometry(pumpCtx, fg, outgoing) | 234 | go pumpAttachmentGeometry(pumpCtx, fg, outgoing) | ||
| 235 | go pumpAttachmentLifecycle(pumpCtx, fg, outgoing) | 235 | go pumpAttachmentLifecycle(pumpCtx, fg, outgoing) | ||
| 236 | + | outputResetRequested := false | |||
| 236 | for { | 237 | for { | ||
| 237 | select { | 238 | select { | ||
| 238 | case result := <-incoming: | 239 | case result := <-incoming: | ||
| @@ -246,9 +247,27 @@ func (w *sessionAttachmentWorker) pumpAttached(ctx context.Context, fg Attachmen | |||||
| 246 | continue | 247 | continue | ||
| 247 | } | 248 | } | ||
| 248 | state = next | 249 | state = next | ||
| 249 | - | if err := fg.Output(state.uiContext(ports.UIContext{Generation: token.Generation}, ports.UIStatusAttached), typed.Data); err != nil { | 250 | + | outputResetRequested = false |
| 251 | + | if err := fg.Output(state.uiContext(ports.UIContext{Generation: attachmentActionableGeneration(fg, token)}, ports.UIStatusAttached), typed.Data); err != nil { | |||
| 250 | return AttachmentEvent{Token: token, Kind: AttachmentEventFailed, Err: fmt.Errorf("vev: publishing output: %w", err)} | 252 | return AttachmentEvent{Token: token, Kind: AttachmentEventFailed, Err: fmt.Errorf("vev: publishing output: %w", err)} | ||
| 251 | } | 253 | } | ||
| 254 | + | case protocol.UIViewUpdate: | |||
| 255 | + | next, accepted, needsReset := state.nextView(typed) | |||
| 256 | + | if needsReset && !outputResetRequested { | |||
| 257 | + | if err := w.send(ctx, fg, stream, protocol.OutputResetRequest{}); err != nil { | |||
| 258 | + | return w.settle(ctx, fg, stream, token, err) | |||
| 259 | + | } | |||
| 260 | + | outputResetRequested = true | |||
| 261 | + | } | |||
| 262 | + | if !accepted { | |||
| 263 | + | continue | |||
| 264 | + | } | |||
| 265 | + | state = next | |||
| 266 | + | if err := fg.Output(state.uiContext(ports.UIContext{Generation: attachmentActionableGeneration(fg, token)}, ports.UIStatusAttached), nil); err != nil { | |||
| 267 | + | return AttachmentEvent{Token: token, Kind: AttachmentEventFailed, Err: fmt.Errorf("vev: publishing view update: %w", err)} | |||
| 268 | + | } | |||
| 269 | + | case protocol.UIReceipt: | |||
| 270 | + | attachmentUIReceipt(fg, typed) | |||
| 252 | case protocol.ErrorMsg: | 271 | case protocol.ErrorMsg: | ||
| 253 | return AttachmentEvent{Token: token, Kind: AttachmentEventFailed, Err: &ProtocolError{Code: typed.Code, Text: typed.Text}} | 272 | return AttachmentEvent{Token: token, Kind: AttachmentEventFailed, Err: &ProtocolError{Code: typed.Code, Text: typed.Text}} | ||
| 254 | case protocol.Detached: | 273 | case protocol.Detached: | ||
| @@ -267,6 +286,11 @@ func (w *sessionAttachmentWorker) pumpAttached(ctx context.Context, fg Attachmen | |||||
| 267 | } | 286 | } | ||
| 268 | if event.input != nil { | 287 | if event.input != nil { | ||
| 269 | fg.AckInput() | 288 | fg.AckInput() | ||
| 289 | + | if event.input.actionID != 0 { | |||
| 290 | + | if err := w.send(ctx, fg, stream, protocol.UIFence{ActionID: event.input.actionID}); err != nil { | |||
| 291 | + | return w.settle(ctx, fg, stream, token, err) | |||
| 292 | + | } | |||
| 293 | + | } | |||
| 270 | } | 294 | } | ||
| 271 | if event.action == AttachmentDetachToPicker { | 295 | if event.action == AttachmentDetachToPicker { | ||
| 272 | return AttachmentEvent{Token: token, Kind: AttachmentEventEnded, Err: errDetachToPicker} | 296 | return AttachmentEvent{Token: token, Kind: AttachmentEventEnded, Err: errDetachToPicker} | ||
| @@ -282,6 +306,24 @@ func (w *sessionAttachmentWorker) pumpAttached(ctx context.Context, fg Attachmen | |||||
| 282 | } | 306 | } | ||
| 283 | } | 307 | } | ||
| 284 | 308 | ||||
| 309 | + | type attachmentUIForeground interface { | |||
| 310 | + | actionableGeneration() uint64 | |||
| 311 | + | uiReceipt(protocol.UIReceipt) | |||
| 312 | + | } | |||
| 313 | + | ||||
| 314 | + | func attachmentActionableGeneration(fg AttachmentForeground, token AttachmentToken) uint64 { | |||
| 315 | + | if ui, ok := fg.(attachmentUIForeground); ok { | |||
| 316 | + | return ui.actionableGeneration() | |||
| 317 | + | } | |||
| 318 | + | return token.Generation | |||
| 319 | + | } | |||
| 320 | + | ||||
| 321 | + | func attachmentUIReceipt(fg AttachmentForeground, receipt protocol.UIReceipt) { | |||
| 322 | + | if ui, ok := fg.(attachmentUIForeground); ok { | |||
| 323 | + | ui.uiReceipt(receipt) | |||
| 324 | + | } | |||
| 325 | + | } | |||
| 326 | + | ||||
| 285 | func pumpAttachmentInput(ctx context.Context, fg AttachmentForeground, out chan<- attachmentClientEvent) { | 327 | func pumpAttachmentInput(ctx context.Context, fg AttachmentForeground, out chan<- attachmentClientEvent) { | ||
| 286 | var sequence uint64 | 328 | var sequence uint64 | ||
| 287 | for { | 329 | for { | ||
| @@ -300,7 +342,7 @@ func pumpAttachmentInput(ctx context.Context, fg AttachmentForeground, out chan< | |||||
| 300 | sequence++ | 342 | sequence++ | ||
| 301 | copyEvent := event | 343 | copyEvent := event | ||
| 302 | select { | 344 | select { | ||
| 303 | - | case out <- attachmentClientEvent{message: protocol.Input{InputSeq: sequence, Data: append([]byte(nil), event.Data...)}, input: ©Event}: | 345 | + | case out <- attachmentClientEvent{message: protocol.Input{InputSeq: sequence, ActionID: event.actionID, Data: append([]byte(nil), event.Data...)}, input: ©Event}: |
| 304 | case <-ctx.Done(): | 346 | case <-ctx.Done(): | ||
| 305 | fg.PreserveInput(event.Data) | 347 | fg.PreserveInput(event.Data) | ||
| 306 | return | 348 | return | ||
M internal/usecase/client/attachment_worker.go +81 -33
| @@ -11,6 +11,7 @@ import ( | |||
| 11 | 11 | ||
| 12 | 12 | "github.com/bnema/vev/internal/domain" | |
| 13 | 13 | "github.com/bnema/vev/internal/ports" | |
| 14 | + | "github.com/bnema/vev/internal/protocol" | |
| 14 | 15 | ) | |
| 15 | 16 | ||
| 16 | 17 | // Attachment worker mechanisms (Plan 001 P5.1b, offline and unactivated). | |
| @@ -44,14 +45,11 @@ import ( | |||
| 44 | 45 | // activation it must use this same pump for picker and attachment consumers; | |
| 45 | 46 | // never start both readers. Runner and supervisor wiring remain unchanged. | |
| 46 | 47 | // | |
| 47 | - | // UI actions are deliberately not part of this slice. AttachmentForeground has | |
| 48 | - | // no action admission seam and attachmentHostConfig has no Actions field: | |
| 49 | - | // binding UI.bindForeground and admitting actions requires the P5 supervisor's | |
| 50 | - | // automation loop, which does not exist yet, so exposing it would publish a | |
| 51 | - | // generation no component can drive. The transactional UI output path | |
| 52 | - | // (ports.UIOutputTransaction, reached through Output) stays independent and | |
| 53 | - | // usable. When the P5 supervisor owns input automation, it must bind and revoke | |
| 54 | - | // UI actions at the same foreground boundaries this host finalizes. | |
| 48 | + | // UI action admission is optional and reuses this host's single pump claim. | |
| 49 | + | // Binding occurs after grant claims the pump and before the worker starts; | |
| 50 | + | // release occurs when finalization starts, before the claim can return to the | |
| 51 | + | // picker. The UI-generated generation is authoritative and is stamped onto the | |
| 52 | + | // attachment token and every publication. | |
| 55 | 53 | ||
| 56 | 54 | // attachmentWorkerJoinTimeout bounds the supervisor's join of a cancelled | |
| 57 | 55 | // worker during the final input window. A worker that honors neither | |
| @@ -125,8 +123,9 @@ type AttachmentEvent struct { | |||
| 125 | 123 | // AttachmentInputEvent is one authorized terminal input delivery. Err carries a | |
| 126 | 124 | // terminal read failure (for example io.EOF) alongside any final bytes. | |
| 127 | 125 | type AttachmentInputEvent struct { | |
| 128 | - | Data []byte | |
| 129 | - | Err error | |
| 126 | + | Data []byte | |
| 127 | + | Err error | |
| 128 | + | actionID uint64 | |
| 130 | 129 | } | |
| 131 | 130 | ||
| 132 | 131 | // AttachmentForeground is the supervisor-granted handle through which a worker | |
| @@ -227,8 +226,7 @@ func (a *attachmentAuthority) currentToken() AttachmentToken { | |||
| 227 | 226 | ||
| 228 | 227 | // grant installs fg as the sole foreground in the active state. It fails while | |
| 229 | 228 | // another grant is live or while the shared input pump already has a consumer, | |
| 230 | - | // which is what keeps exactly one foreground owner. The host never allocates a | |
| 231 | - | // UI generation here: UI action binding is deferred to the P5 supervisor. | |
| 229 | + | // which is what keeps exactly one foreground owner. | |
| 232 | 230 | func (a *attachmentAuthority) grant(fg *attachmentForeground) bool { | |
| 233 | 231 | if a == nil || fg == nil { | |
| 234 | 232 | return false | |
| @@ -344,10 +342,8 @@ func (a *attachmentAuthority) retain(fg *attachmentForeground, fn func()) bool { | |||
| 344 | 342 | } | |
| 345 | 343 | ||
| 346 | 344 | // attachmentHostConfig supplies the supervisor-owned dependencies. Terminal, | |
| 347 | - | // clock, and input belong to the supervisor; UI and OnAttached are optional | |
| 348 | - | // seams. There is no Actions seam: UI action admission needs the P5 | |
| 349 | - | // supervisor's automation loop before it can be exposed (see the package | |
| 350 | - | // integration requirement above). | |
| 345 | + | // clock, and input belong to the supervisor; UI seams and OnAttached are | |
| 346 | + | // optional. | |
| 351 | 347 | type attachmentHostConfig struct { | |
| 352 | 348 | // Terminal is the controlling terminal the supervisor owns. Optional only | |
| 353 | 349 | // so a mechanism test can omit output entirely; production supplies it. | |
| @@ -363,6 +359,8 @@ type attachmentHostConfig struct { | |||
| 363 | 359 | // UI is the supervisor-owned UI publication transaction. When nil it is | |
| 364 | 360 | // taken from Terminal when that implements ports.UIOutputTransaction. | |
| 365 | 361 | UI ports.UIOutputTransaction | |
| 362 | + | // ActionUI optionally binds automation to the same claimed input consumer. | |
| 363 | + | ActionUI *UI | |
| 366 | 364 | // OnAttached observes a successful MarkAttached. Optional. | |
| 367 | 365 | OnAttached func(AttachmentToken) | |
| 368 | 366 | } | |
| @@ -377,6 +375,7 @@ type attachmentHost struct { | |||
| 377 | 375 | clock ports.Clock | |
| 378 | 376 | input *terminalInputPump | |
| 379 | 377 | ui ports.UIOutputTransaction | |
| 378 | + | actionUI *UI | |
| 380 | 379 | resizes <-chan domain.Geometry | |
| 381 | 380 | actions <-chan AttachmentLifecycleAction | |
| 382 | 381 | onAttach func(AttachmentToken) | |
| @@ -413,6 +412,7 @@ func newAttachmentHost(cfg attachmentHostConfig) *attachmentHost { | |||
| 413 | 412 | clock: clock, | |
| 414 | 413 | input: cfg.Input, | |
| 415 | 414 | ui: ui, | |
| 415 | + | actionUI: cfg.ActionUI, | |
| 416 | 416 | resizes: resizes, | |
| 417 | 417 | actions: cfg.Actions, | |
| 418 | 418 | onAttach: cfg.OnAttached, | |
| @@ -431,8 +431,10 @@ func newAttachmentHost(cfg attachmentHostConfig) *attachmentHost { | |||
| 431 | 431 | // - the shared input pump already has a consumer (a picker or rival | |
| 432 | 432 | // foreground owns input). | |
| 433 | 433 | // | |
| 434 | - | // The granted token, event token, and published UI context always use the | |
| 435 | - | // caller-supplied Generation. The caller must Wait or Cancel the returned run. | |
| 434 | + | // The granted token and event token always retain the caller-supplied | |
| 435 | + | // supervisor generation. When action UI is configured, its independent | |
| 436 | + | // generation is used only for UI publication and action fencing. The caller | |
| 437 | + | // must Wait or Cancel the returned run. | |
| 436 | 438 | func (h *attachmentHost) Begin(ctx context.Context, token AttachmentToken, worker AttachmentWorker, stream ports.BrokerLogicalConnection) (*attachmentRun, bool) { | |
| 437 | 439 | if h == nil || token.IsZero() || supervisorNil(worker) || supervisorNil(stream) { | |
| 438 | 440 | return nil, false | |
| @@ -446,6 +448,9 @@ func (h *attachmentHost) Begin(ctx context.Context, token AttachmentToken, worke | |||
| 446 | 448 | cancel() | |
| 447 | 449 | return nil, false | |
| 448 | 450 | } | |
| 451 | + | if h.actionUI != nil { | |
| 452 | + | fg.uiGeneration = h.actionUI.bindForeground(workerCtx, fg.input, fg.consumer) | |
| 453 | + | } | |
| 449 | 454 | run := &attachmentRun{ | |
| 450 | 455 | host: h, | |
| 451 | 456 | token: token, | |
| @@ -674,18 +679,19 @@ func (r *attachmentRun) join() { | |||
| 674 | 679 | // AttachmentForeground. It reuses the existing terminal input pump for input | |
| 675 | 680 | // and the existing foreground send lease for output serialization. | |
| 676 | 681 | type attachmentForeground struct { | |
| 677 | - | authority *attachmentAuthority | |
| 678 | - | token AttachmentToken | |
| 679 | - | stream ports.BrokerLogicalConnection | |
| 680 | - | input *terminalInputPump | |
| 681 | - | consumer uint64 | |
| 682 | - | lease *foregroundSendLease | |
| 683 | - | term ports.Terminal | |
| 684 | - | ui ports.UIOutputTransaction | |
| 685 | - | resizes <-chan domain.Geometry | |
| 686 | - | actions <-chan AttachmentLifecycleAction | |
| 687 | - | host *attachmentHost | |
| 688 | - | done chan struct{} | |
| 682 | + | authority *attachmentAuthority | |
| 683 | + | token AttachmentToken | |
| 684 | + | stream ports.BrokerLogicalConnection | |
| 685 | + | input *terminalInputPump | |
| 686 | + | consumer uint64 | |
| 687 | + | lease *foregroundSendLease | |
| 688 | + | term ports.Terminal | |
| 689 | + | ui ports.UIOutputTransaction | |
| 690 | + | uiGeneration uint64 | |
| 691 | + | resizes <-chan domain.Geometry | |
| 692 | + | actions <-chan AttachmentLifecycleAction | |
| 693 | + | host *attachmentHost | |
| 694 | + | done chan struct{} | |
| 689 | 695 | // finalizedOnce and finishedOnce split the single teardown into the | |
| 690 | 696 | // deterministic final input window: finalize closes Done and stops output | |
| 691 | 697 | // first, finish revokes the retained consumer only after the join. | |
| @@ -705,6 +711,26 @@ type attachmentForeground struct { | |||
| 705 | 711 | // Token is the generation/attempt identity of this grant. | |
| 706 | 712 | func (f *attachmentForeground) Token() AttachmentToken { return f.token } | |
| 707 | 713 | ||
| 714 | + | func (f *attachmentForeground) actionableGeneration() uint64 { | |
| 715 | + | if f.uiGeneration != 0 { | |
| 716 | + | return f.uiGeneration | |
| 717 | + | } | |
| 718 | + | return f.token.Generation | |
| 719 | + | } | |
| 720 | + | ||
| 721 | + | func (f *attachmentForeground) uiPublished() { | |
| 722 | + | if f.uiGeneration == 0 || f.host == nil || f.host.actionUI == nil { | |
| 723 | + | return | |
| 724 | + | } | |
| 725 | + | f.host.actionUI.published(f.uiGeneration) | |
| 726 | + | } | |
| 727 | + | ||
| 728 | + | func (f *attachmentForeground) uiReceipt(receipt protocol.UIReceipt) { | |
| 729 | + | if f.uiGeneration != 0 && f.host != nil && f.host.actionUI != nil { | |
| 730 | + | f.host.actionUI.receipt(f.uiGeneration, receipt) | |
| 731 | + | } | |
| 732 | + | } | |
| 733 | + | ||
| 708 | 734 | // Stream is the supervisor-admitted logical stream this worker owns. | |
| 709 | 735 | func (f *attachmentForeground) Stream() ports.BrokerLogicalConnection { return f.stream } | |
| 710 | 736 | ||
| @@ -730,6 +756,9 @@ func (f *attachmentForeground) finalize() { | |||
| 730 | 756 | if f.lease != nil { | |
| 731 | 757 | f.lease.stop() | |
| 732 | 758 | } | |
| 759 | + | if f.uiGeneration != 0 && f.host != nil && f.host.actionUI != nil { | |
| 760 | + | f.host.actionUI.releaseForeground(f.uiGeneration) | |
| 761 | + | } | |
| 733 | 762 | close(f.done) | |
| 734 | 763 | }) | |
| 735 | 764 | } | |
| @@ -788,7 +817,7 @@ func (f *attachmentForeground) Input(ctx context.Context) (AttachmentInputEvent, | |||
| 788 | 817 | result, ok := f.input.take(ctx, f.consumer) | |
| 789 | 818 | if ok { | |
| 790 | 819 | f.beginDelivery() | |
| 791 | - | return AttachmentInputEvent{Data: result.data, Err: result.err}, true | |
| 820 | + | return AttachmentInputEvent{Data: result.data, Err: result.err, actionID: result.actionID}, true | |
| 792 | 821 | } | |
| 793 | 822 | if ctx.Err() != nil { | |
| 794 | 823 | return AttachmentInputEvent{}, false | |
| @@ -798,6 +827,20 @@ func (f *attachmentForeground) Input(ctx context.Context) (AttachmentInputEvent, | |||
| 798 | 827 | return AttachmentInputEvent{}, false | |
| 799 | 828 | case <-f.done: | |
| 800 | 829 | return AttachmentInputEvent{}, false | |
| 830 | + | case request := <-f.input.automation: | |
| 831 | + | f.input.mu.Lock() | |
| 832 | + | clean := f.input.consumer == f.consumer && request.consumer == f.consumer && f.input.pending == nil && len(f.input.residual) == 0 && f.input.delivering == 0 | |
| 833 | + | f.input.mu.Unlock() | |
| 834 | + | clean = clean && request.ctx.Err() == nil && request.record.source == terminalInputAutomation && request.record.actionID != 0 && request.record.generation == f.uiGeneration && request.record.endBatch && len(request.record.data) > 0 && len(request.record.data) <= uiMaxInputBytes && validUIKeyBatch(request.record.data) | |
| 835 | + | if clean && request.owner != nil { | |
| 836 | + | clean = request.owner.accept(request.record.actionID, request.record.generation) | |
| 837 | + | } | |
| 838 | + | request.admitted <- clean | |
| 839 | + | if !clean { | |
| 840 | + | continue | |
| 841 | + | } | |
| 842 | + | request.dispatched <- true | |
| 843 | + | return AttachmentInputEvent{Data: append([]byte(nil), request.record.data...), actionID: request.record.actionID}, true | |
| 801 | 844 | case <-f.input.readyFor(f.consumer): | |
| 802 | 845 | } | |
| 803 | 846 | } | |
| @@ -927,10 +970,15 @@ func (f *attachmentForeground) Output(uiContext ports.UIContext, data []byte) er | |||
| 927 | 970 | outputErr = ports.ErrUIUnavailable | |
| 928 | 971 | return true | |
| 929 | 972 | } | |
| 930 | - | uiContext.Generation = f.token.Generation | |
| 973 | + | uiContext.Generation = f.actionableGeneration() | |
| 931 | 974 | f.ui.BeginOutput(uiContext) | |
| 932 | 975 | success := false | |
| 933 | - | defer func() { f.ui.EndOutput(success) }() | |
| 976 | + | defer func() { | |
| 977 | + | f.ui.EndOutput(success) | |
| 978 | + | if success { | |
| 979 | + | f.uiPublished() | |
| 980 | + | } | |
| 981 | + | }() | |
| 934 | 982 | // An unavailable capture means the optional UI observation channel is | |
| 935 | 983 | // disabled or closed, so the terminal frame still writes and flushes. | |
| 936 | 984 | // Any other publication error aborts before the frame is written. | |
| @@ -11,6 +11,7 @@ import ( | |||||
| 11 | 11 | ||||
| 12 | "github.com/bnema/vev/internal/domain" | 12 | "github.com/bnema/vev/internal/domain" | ||
| 13 | "github.com/bnema/vev/internal/ports" | 13 | "github.com/bnema/vev/internal/ports" | ||
| 14 | + | "github.com/bnema/vev/internal/protocol" | |||
| 14 | ) | 15 | ) | ||
| 15 | 16 | ||||
| 16 | // Attachment worker mechanisms (Plan 001 P5.1b, offline and unactivated). | 17 | // Attachment worker mechanisms (Plan 001 P5.1b, offline and unactivated). | ||
| @@ -44,14 +45,11 @@ import ( | |||||
| 44 | // activation it must use this same pump for picker and attachment consumers; | 45 | // activation it must use this same pump for picker and attachment consumers; | ||
| 45 | // never start both readers. Runner and supervisor wiring remain unchanged. | 46 | // never start both readers. Runner and supervisor wiring remain unchanged. | ||
| 46 | // | 47 | // | ||
| 47 | - | // UI actions are deliberately not part of this slice. AttachmentForeground has | 48 | + | // UI action admission is optional and reuses this host's single pump claim. |
| 48 | - | // no action admission seam and attachmentHostConfig has no Actions field: | 49 | + | // Binding occurs after grant claims the pump and before the worker starts; |
| 49 | - | // binding UI.bindForeground and admitting actions requires the P5 supervisor's | 50 | + | // release occurs when finalization starts, before the claim can return to the |
| 50 | - | // automation loop, which does not exist yet, so exposing it would publish a | 51 | + | // picker. The UI-generated generation is authoritative and is stamped onto the |
| 51 | - | // generation no component can drive. The transactional UI output path | 52 | + | // attachment token and every publication. |
| 52 | - | // (ports.UIOutputTransaction, reached through Output) stays independent and | |||
| 53 | - | // usable. When the P5 supervisor owns input automation, it must bind and revoke | |||
| 54 | - | // UI actions at the same foreground boundaries this host finalizes. | |||
| 55 | 53 | ||||
| 56 | // attachmentWorkerJoinTimeout bounds the supervisor's join of a cancelled | 54 | // attachmentWorkerJoinTimeout bounds the supervisor's join of a cancelled | ||
| 57 | // worker during the final input window. A worker that honors neither | 55 | // worker during the final input window. A worker that honors neither | ||
| @@ -125,8 +123,9 @@ type AttachmentEvent struct { | |||||
| 125 | // AttachmentInputEvent is one authorized terminal input delivery. Err carries a | 123 | // AttachmentInputEvent is one authorized terminal input delivery. Err carries a | ||
| 126 | // terminal read failure (for example io.EOF) alongside any final bytes. | 124 | // terminal read failure (for example io.EOF) alongside any final bytes. | ||
| 127 | type AttachmentInputEvent struct { | 125 | type AttachmentInputEvent struct { | ||
| 128 | - | Data []byte | 126 | + | Data []byte |
| 129 | - | Err error | 127 | + | Err error |
| 128 | + | actionID uint64 | |||
| 130 | } | 129 | } | ||
| 131 | 130 | ||||
| 132 | // AttachmentForeground is the supervisor-granted handle through which a worker | 131 | // AttachmentForeground is the supervisor-granted handle through which a worker | ||
| @@ -227,8 +226,7 @@ func (a *attachmentAuthority) currentToken() AttachmentToken { | |||||
| 227 | 226 | ||||
| 228 | // grant installs fg as the sole foreground in the active state. It fails while | 227 | // grant installs fg as the sole foreground in the active state. It fails while | ||
| 229 | // another grant is live or while the shared input pump already has a consumer, | 228 | // another grant is live or while the shared input pump already has a consumer, | ||
| 230 | - | // which is what keeps exactly one foreground owner. The host never allocates a | 229 | + | // which is what keeps exactly one foreground owner. |
| 231 | - | // UI generation here: UI action binding is deferred to the P5 supervisor. | |||
| 232 | func (a *attachmentAuthority) grant(fg *attachmentForeground) bool { | 230 | func (a *attachmentAuthority) grant(fg *attachmentForeground) bool { | ||
| 233 | if a == nil || fg == nil { | 231 | if a == nil || fg == nil { | ||
| 234 | return false | 232 | return false | ||
| @@ -344,10 +342,8 @@ func (a *attachmentAuthority) retain(fg *attachmentForeground, fn func()) bool { | |||||
| 344 | } | 342 | } | ||
| 345 | 343 | ||||
| 346 | // attachmentHostConfig supplies the supervisor-owned dependencies. Terminal, | 344 | // attachmentHostConfig supplies the supervisor-owned dependencies. Terminal, | ||
| 347 | - | // clock, and input belong to the supervisor; UI and OnAttached are optional | 345 | + | // clock, and input belong to the supervisor; UI seams and OnAttached are |
| 348 | - | // seams. There is no Actions seam: UI action admission needs the P5 | 346 | + | // optional. |
| 349 | - | // supervisor's automation loop before it can be exposed (see the package | |||
| 350 | - | // integration requirement above). | |||
| 351 | type attachmentHostConfig struct { | 347 | type attachmentHostConfig struct { | ||
| 352 | // Terminal is the controlling terminal the supervisor owns. Optional only | 348 | // Terminal is the controlling terminal the supervisor owns. Optional only | ||
| 353 | // so a mechanism test can omit output entirely; production supplies it. | 349 | // so a mechanism test can omit output entirely; production supplies it. | ||
| @@ -363,6 +359,8 @@ type attachmentHostConfig struct { | |||||
| 363 | // UI is the supervisor-owned UI publication transaction. When nil it is | 359 | // UI is the supervisor-owned UI publication transaction. When nil it is | ||
| 364 | // taken from Terminal when that implements ports.UIOutputTransaction. | 360 | // taken from Terminal when that implements ports.UIOutputTransaction. | ||
| 365 | UI ports.UIOutputTransaction | 361 | UI ports.UIOutputTransaction | ||
| 362 | + | // ActionUI optionally binds automation to the same claimed input consumer. | |||
| 363 | + | ActionUI *UI | |||
| 366 | // OnAttached observes a successful MarkAttached. Optional. | 364 | // OnAttached observes a successful MarkAttached. Optional. | ||
| 367 | OnAttached func(AttachmentToken) | 365 | OnAttached func(AttachmentToken) | ||
| 368 | } | 366 | } | ||
| @@ -377,6 +375,7 @@ type attachmentHost struct { | |||||
| 377 | clock ports.Clock | 375 | clock ports.Clock | ||
| 378 | input *terminalInputPump | 376 | input *terminalInputPump | ||
| 379 | ui ports.UIOutputTransaction | 377 | ui ports.UIOutputTransaction | ||
| 378 | + | actionUI *UI | |||
| 380 | resizes <-chan domain.Geometry | 379 | resizes <-chan domain.Geometry | ||
| 381 | actions <-chan AttachmentLifecycleAction | 380 | actions <-chan AttachmentLifecycleAction | ||
| 382 | onAttach func(AttachmentToken) | 381 | onAttach func(AttachmentToken) | ||
| @@ -413,6 +412,7 @@ func newAttachmentHost(cfg attachmentHostConfig) *attachmentHost { | |||||
| 413 | clock: clock, | 412 | clock: clock, | ||
| 414 | input: cfg.Input, | 413 | input: cfg.Input, | ||
| 415 | ui: ui, | 414 | ui: ui, | ||
| 415 | + | actionUI: cfg.ActionUI, | |||
| 416 | resizes: resizes, | 416 | resizes: resizes, | ||
| 417 | actions: cfg.Actions, | 417 | actions: cfg.Actions, | ||
| 418 | onAttach: cfg.OnAttached, | 418 | onAttach: cfg.OnAttached, | ||
| @@ -431,8 +431,10 @@ func newAttachmentHost(cfg attachmentHostConfig) *attachmentHost { | |||||
| 431 | // - the shared input pump already has a consumer (a picker or rival | 431 | // - the shared input pump already has a consumer (a picker or rival | ||
| 432 | // foreground owns input). | 432 | // foreground owns input). | ||
| 433 | // | 433 | // | ||
| 434 | - | // The granted token, event token, and published UI context always use the | 434 | + | // The granted token and event token always retain the caller-supplied |
| 435 | - | // caller-supplied Generation. The caller must Wait or Cancel the returned run. | 435 | + | // supervisor generation. When action UI is configured, its independent |
| 436 | + | // generation is used only for UI publication and action fencing. The caller | |||
| 437 | + | // must Wait or Cancel the returned run. | |||
| 436 | func (h *attachmentHost) Begin(ctx context.Context, token AttachmentToken, worker AttachmentWorker, stream ports.BrokerLogicalConnection) (*attachmentRun, bool) { | 438 | func (h *attachmentHost) Begin(ctx context.Context, token AttachmentToken, worker AttachmentWorker, stream ports.BrokerLogicalConnection) (*attachmentRun, bool) { | ||
| 437 | if h == nil || token.IsZero() || supervisorNil(worker) || supervisorNil(stream) { | 439 | if h == nil || token.IsZero() || supervisorNil(worker) || supervisorNil(stream) { | ||
| 438 | return nil, false | 440 | return nil, false | ||
| @@ -446,6 +448,9 @@ func (h *attachmentHost) Begin(ctx context.Context, token AttachmentToken, worke | |||||
| 446 | cancel() | 448 | cancel() | ||
| 447 | return nil, false | 449 | return nil, false | ||
| 448 | } | 450 | } | ||
| 451 | + | if h.actionUI != nil { | |||
| 452 | + | fg.uiGeneration = h.actionUI.bindForeground(workerCtx, fg.input, fg.consumer) | |||
| 453 | + | } | |||
| 449 | run := &attachmentRun{ | 454 | run := &attachmentRun{ | ||
| 450 | host: h, | 455 | host: h, | ||
| 451 | token: token, | 456 | token: token, | ||
| @@ -674,18 +679,19 @@ func (r *attachmentRun) join() { | |||||
| 674 | // AttachmentForeground. It reuses the existing terminal input pump for input | 679 | // AttachmentForeground. It reuses the existing terminal input pump for input | ||
| 675 | // and the existing foreground send lease for output serialization. | 680 | // and the existing foreground send lease for output serialization. | ||
| 676 | type attachmentForeground struct { | 681 | type attachmentForeground struct { | ||
| 677 | - | authority *attachmentAuthority | 682 | + | authority *attachmentAuthority |
| 678 | - | token AttachmentToken | 683 | + | token AttachmentToken |
| 679 | - | stream ports.BrokerLogicalConnection | 684 | + | stream ports.BrokerLogicalConnection |
| 680 | - | input *terminalInputPump | 685 | + | input *terminalInputPump |
| 681 | - | consumer uint64 | 686 | + | consumer uint64 |
| 682 | - | lease *foregroundSendLease | 687 | + | lease *foregroundSendLease |
| 683 | - | term ports.Terminal | 688 | + | term ports.Terminal |
| 684 | - | ui ports.UIOutputTransaction | 689 | + | ui ports.UIOutputTransaction |
| 685 | - | resizes <-chan domain.Geometry | 690 | + | uiGeneration uint64 |
| 686 | - | actions <-chan AttachmentLifecycleAction | 691 | + | resizes <-chan domain.Geometry |
| 687 | - | host *attachmentHost | 692 | + | actions <-chan AttachmentLifecycleAction |
| 688 | - | done chan struct{} | 693 | + | host *attachmentHost |
| 694 | + | done chan struct{} | |||
| 689 | // finalizedOnce and finishedOnce split the single teardown into the | 695 | // finalizedOnce and finishedOnce split the single teardown into the | ||
| 690 | // deterministic final input window: finalize closes Done and stops output | 696 | // deterministic final input window: finalize closes Done and stops output | ||
| 691 | // first, finish revokes the retained consumer only after the join. | 697 | // first, finish revokes the retained consumer only after the join. | ||
| @@ -705,6 +711,26 @@ type attachmentForeground struct { | |||||
| 705 | // Token is the generation/attempt identity of this grant. | 711 | // Token is the generation/attempt identity of this grant. | ||
| 706 | func (f *attachmentForeground) Token() AttachmentToken { return f.token } | 712 | func (f *attachmentForeground) Token() AttachmentToken { return f.token } | ||
| 707 | 713 | ||||
| 714 | + | func (f *attachmentForeground) actionableGeneration() uint64 { | |||
| 715 | + | if f.uiGeneration != 0 { | |||
| 716 | + | return f.uiGeneration | |||
| 717 | + | } | |||
| 718 | + | return f.token.Generation | |||
| 719 | + | } | |||
| 720 | + | ||||
| 721 | + | func (f *attachmentForeground) uiPublished() { | |||
| 722 | + | if f.uiGeneration == 0 || f.host == nil || f.host.actionUI == nil { | |||
| 723 | + | return | |||
| 724 | + | } | |||
| 725 | + | f.host.actionUI.published(f.uiGeneration) | |||
| 726 | + | } | |||
| 727 | + | ||||
| 728 | + | func (f *attachmentForeground) uiReceipt(receipt protocol.UIReceipt) { | |||
| 729 | + | if f.uiGeneration != 0 && f.host != nil && f.host.actionUI != nil { | |||
| 730 | + | f.host.actionUI.receipt(f.uiGeneration, receipt) | |||
| 731 | + | } | |||
| 732 | + | } | |||
| 733 | + | ||||
| 708 | // Stream is the supervisor-admitted logical stream this worker owns. | 734 | // Stream is the supervisor-admitted logical stream this worker owns. | ||
| 709 | func (f *attachmentForeground) Stream() ports.BrokerLogicalConnection { return f.stream } | 735 | func (f *attachmentForeground) Stream() ports.BrokerLogicalConnection { return f.stream } | ||
| 710 | 736 | ||||
| @@ -730,6 +756,9 @@ func (f *attachmentForeground) finalize() { | |||||
| 730 | if f.lease != nil { | 756 | if f.lease != nil { | ||
| 731 | f.lease.stop() | 757 | f.lease.stop() | ||
| 732 | } | 758 | } | ||
| 759 | + | if f.uiGeneration != 0 && f.host != nil && f.host.actionUI != nil { | |||
| 760 | + | f.host.actionUI.releaseForeground(f.uiGeneration) | |||
| 761 | + | } | |||
| 733 | close(f.done) | 762 | close(f.done) | ||
| 734 | }) | 763 | }) | ||
| 735 | } | 764 | } | ||
| @@ -788,7 +817,7 @@ func (f *attachmentForeground) Input(ctx context.Context) (AttachmentInputEvent, | |||||
| 788 | result, ok := f.input.take(ctx, f.consumer) | 817 | result, ok := f.input.take(ctx, f.consumer) | ||
| 789 | if ok { | 818 | if ok { | ||
| 790 | f.beginDelivery() | 819 | f.beginDelivery() | ||
| 791 | - | return AttachmentInputEvent{Data: result.data, Err: result.err}, true | 820 | + | return AttachmentInputEvent{Data: result.data, Err: result.err, actionID: result.actionID}, true |
| 792 | } | 821 | } | ||
| 793 | if ctx.Err() != nil { | 822 | if ctx.Err() != nil { | ||
| 794 | return AttachmentInputEvent{}, false | 823 | return AttachmentInputEvent{}, false | ||
| @@ -798,6 +827,20 @@ func (f *attachmentForeground) Input(ctx context.Context) (AttachmentInputEvent, | |||||
| 798 | return AttachmentInputEvent{}, false | 827 | return AttachmentInputEvent{}, false | ||
| 799 | case <-f.done: | 828 | case <-f.done: | ||
| 800 | return AttachmentInputEvent{}, false | 829 | return AttachmentInputEvent{}, false | ||
| 830 | + | case request := <-f.input.automation: | |||
| 831 | + | f.input.mu.Lock() | |||
| 832 | + | clean := f.input.consumer == f.consumer && request.consumer == f.consumer && f.input.pending == nil && len(f.input.residual) == 0 && f.input.delivering == 0 | |||
| 833 | + | f.input.mu.Unlock() | |||
| 834 | + | clean = clean && request.ctx.Err() == nil && request.record.source == terminalInputAutomation && request.record.actionID != 0 && request.record.generation == f.uiGeneration && request.record.endBatch && len(request.record.data) > 0 && len(request.record.data) <= uiMaxInputBytes && validUIKeyBatch(request.record.data) | |||
| 835 | + | if clean && request.owner != nil { | |||
| 836 | + | clean = request.owner.accept(request.record.actionID, request.record.generation) | |||
| 837 | + | } | |||
| 838 | + | request.admitted <- clean | |||
| 839 | + | if !clean { | |||
| 840 | + | continue | |||
| 841 | + | } | |||
| 842 | + | request.dispatched <- true | |||
| 843 | + | return AttachmentInputEvent{Data: append([]byte(nil), request.record.data...), actionID: request.record.actionID}, true | |||
| 801 | case <-f.input.readyFor(f.consumer): | 844 | case <-f.input.readyFor(f.consumer): | ||
| 802 | } | 845 | } | ||
| 803 | } | 846 | } | ||
| @@ -927,10 +970,15 @@ func (f *attachmentForeground) Output(uiContext ports.UIContext, data []byte) er | |||||
| 927 | outputErr = ports.ErrUIUnavailable | 970 | outputErr = ports.ErrUIUnavailable | ||
| 928 | return true | 971 | return true | ||
| 929 | } | 972 | } | ||
| 930 | - | uiContext.Generation = f.token.Generation | 973 | + | uiContext.Generation = f.actionableGeneration() |
| 931 | f.ui.BeginOutput(uiContext) | 974 | f.ui.BeginOutput(uiContext) | ||
| 932 | success := false | 975 | success := false | ||
| 933 | - | defer func() { f.ui.EndOutput(success) }() | 976 | + | defer func() { |
| 977 | + | f.ui.EndOutput(success) | |||
| 978 | + | if success { | |||
| 979 | + | f.uiPublished() | |||
| 980 | + | } | |||
| 981 | + | }() | |||
| 934 | // An unavailable capture means the optional UI observation channel is | 982 | // An unavailable capture means the optional UI observation channel is | ||
| 935 | // disabled or closed, so the terminal frame still writes and flushes. | 983 | // disabled or closed, so the terminal frame still writes and flushes. | ||
| 936 | // Any other publication error aborts before the frame is written. | 984 | // Any other publication error aborts before the frame is written. | ||
M internal/usecase/client/supervisor.go +7 -2
| @@ -279,6 +279,10 @@ type SupervisorConfig struct { | |||
| 279 | 279 | // randomness seam so tests can pin the retry cadence exactly. Optional; the | |
| 280 | 280 | // default is a process-local random source. | |
| 281 | 281 | Jitter func() float64 | |
| 282 | + | // UI optionally enables supervisor-owned action admission. Each attachment | |
| 283 | + | // is bound only after it owns the pump claim; no UI preserves the prior | |
| 284 | + | // publication-only behavior with no admitted automation. | |
| 285 | + | UI *UI | |
| 282 | 286 | // Picker is the client-owned picker (Plan 001 P5.2b). When set, the | |
| 283 | 287 | // supervisor folds every broker publication into it and hands it every | |
| 284 | 288 | // terminal read from the same single input lifetime used for EOF detection, | |
| @@ -385,12 +389,13 @@ func NewSupervisor(cfg SupervisorConfig) (*Supervisor, error) { | |||
| 385 | 389 | // The host is the supervisor's existing foreground grant. It owns no raw | |
| 386 | 390 | // mode and starts no reader: the supervisor keeps its one terminal input | |
| 387 | 391 | // lifetime, so the attachment path adds neither a second reader nor a | |
| 388 | - | // second writer. Terminal input pumping into a session and UI actions are | |
| 389 | - | // deliberately left to P5.4 (see the P5.1b integration note). | |
| 392 | + | // second writer. The optional UI binds to this host's existing pump claim; | |
| 393 | + | // it never starts another terminal reader. | |
| 390 | 394 | supervisor.attachments = newAttachmentHost(attachmentHostConfig{ | |
| 391 | 395 | Terminal: cfg.Terminal, | |
| 392 | 396 | Clock: cfg.Clock, | |
| 393 | 397 | Actions: cfg.LifecycleActions, | |
| 398 | + | ActionUI: cfg.UI, | |
| 394 | 399 | OnAttached: func(AttachmentToken) { supervisor.transition(supervisorEvent{kind: supervisorAttached}) }, | |
| 395 | 400 | }) | |
| 396 | 401 | return supervisor, nil | |
| @@ -279,6 +279,10 @@ type SupervisorConfig struct { | |||||
| 279 | // randomness seam so tests can pin the retry cadence exactly. Optional; the | 279 | // randomness seam so tests can pin the retry cadence exactly. Optional; the | ||
| 280 | // default is a process-local random source. | 280 | // default is a process-local random source. | ||
| 281 | Jitter func() float64 | 281 | Jitter func() float64 | ||
| 282 | + | // UI optionally enables supervisor-owned action admission. Each attachment | |||
| 283 | + | // is bound only after it owns the pump claim; no UI preserves the prior | |||
| 284 | + | // publication-only behavior with no admitted automation. | |||
| 285 | + | UI *UI | |||
| 282 | // Picker is the client-owned picker (Plan 001 P5.2b). When set, the | 286 | // Picker is the client-owned picker (Plan 001 P5.2b). When set, the | ||
| 283 | // supervisor folds every broker publication into it and hands it every | 287 | // supervisor folds every broker publication into it and hands it every | ||
| 284 | // terminal read from the same single input lifetime used for EOF detection, | 288 | // terminal read from the same single input lifetime used for EOF detection, | ||
| @@ -385,12 +389,13 @@ func NewSupervisor(cfg SupervisorConfig) (*Supervisor, error) { | |||||
| 385 | // The host is the supervisor's existing foreground grant. It owns no raw | 389 | // The host is the supervisor's existing foreground grant. It owns no raw | ||
| 386 | // mode and starts no reader: the supervisor keeps its one terminal input | 390 | // mode and starts no reader: the supervisor keeps its one terminal input | ||
| 387 | // lifetime, so the attachment path adds neither a second reader nor a | 391 | // lifetime, so the attachment path adds neither a second reader nor a | ||
| 388 | - | // second writer. Terminal input pumping into a session and UI actions are | 392 | + | // second writer. The optional UI binds to this host's existing pump claim; |
| 389 | - | // deliberately left to P5.4 (see the P5.1b integration note). | 393 | + | // it never starts another terminal reader. |
| 390 | supervisor.attachments = newAttachmentHost(attachmentHostConfig{ | 394 | supervisor.attachments = newAttachmentHost(attachmentHostConfig{ | ||
| 391 | Terminal: cfg.Terminal, | 395 | Terminal: cfg.Terminal, | ||
| 392 | Clock: cfg.Clock, | 396 | Clock: cfg.Clock, | ||
| 393 | Actions: cfg.LifecycleActions, | 397 | Actions: cfg.LifecycleActions, | ||
| 398 | + | ActionUI: cfg.UI, | |||
| 394 | OnAttached: func(AttachmentToken) { supervisor.transition(supervisorEvent{kind: supervisorAttached}) }, | 399 | OnAttached: func(AttachmentToken) { supervisor.transition(supervisorEvent{kind: supervisorAttached}) }, | ||
| 395 | }) | 400 | }) | ||
| 396 | return supervisor, nil | 401 | return supervisor, nil | ||
M internal/usecase/client/supervisor_attach_test.go +488 -7
| @@ -31,9 +31,33 @@ type attachTestTerminal struct { | |||
| 31 | 31 | flushes int | |
| 32 | 32 | restores int | |
| 33 | 33 | publications []ports.UIContext | |
| 34 | + | pending ports.UIContext | |
| 35 | + | revision uint64 | |
| 36 | + | txDepth int | |
| 37 | + | txFailed bool | |
| 38 | + | available bool | |
| 34 | 39 | successes []bool | |
| 35 | 40 | publishEntered chan struct{} | |
| 36 | 41 | publishRelease <-chan struct{} | |
| 42 | + | // publishMatch narrows the publication block to one matching transaction. | |
| 43 | + | // A nil predicate blocks every publication, which is what the existing | |
| 44 | + | // initial-publication tests rely on. | |
| 45 | + | publishMatch func(ports.UIContext) bool | |
| 46 | + | // detachedSignals receives one signal per committed Detached publication, so | |
| 47 | + | // a test observes attachment release without polling. | |
| 48 | + | detachedSignals chan struct{} | |
| 49 | + | // events records the transaction order: each begin, each publication, and | |
| 50 | + | // each successful drain. A test proves release ordering against it instead | |
| 51 | + | // of inferring it from separate slices. | |
| 52 | + | events []attachTestEvent | |
| 53 | + | changes chan struct{} | |
| 54 | + | } | |
| 55 | + | ||
| 56 | + | // attachTestEvent is one ordered terminal transaction event. | |
| 57 | + | type attachTestEvent struct { | |
| 58 | + | kind string // begin | publish | drain | |
| 59 | + | status ports.UIPresentationStatus | |
| 60 | + | state uint64 | |
| 37 | 61 | } | |
| 38 | 62 | ||
| 39 | 63 | func (t *attachTestTerminal) EnterRaw() (func() error, error) { | |
| @@ -62,28 +86,123 @@ func (t *attachTestTerminal) Flush() error { | |||
| 62 | 86 | return nil | |
| 63 | 87 | } | |
| 64 | 88 | ||
| 65 | - | func (t *attachTestTerminal) BeginOutput(ports.UIContext) {} | |
| 89 | + | func (t *attachTestTerminal) BeginOutput(ctx ports.UIContext) { | |
| 90 | + | t.mu.Lock() | |
| 91 | + | if ctx.AttachmentHandle == "" && len(t.publications) != 0 { | |
| 92 | + | ctx.AttachmentHandle = t.publications[len(t.publications)-1].AttachmentHandle | |
| 93 | + | } | |
| 94 | + | if ctx.Generation == 0 && len(t.publications) != 0 { | |
| 95 | + | ctx.Generation = t.publications[len(t.publications)-1].Generation | |
| 96 | + | } | |
| 97 | + | if t.txDepth == 0 { | |
| 98 | + | t.pending = ctx | |
| 99 | + | t.txFailed = false | |
| 100 | + | t.events = append(t.events, attachTestEvent{kind: "begin", status: ctx.Status, state: ctx.OutputState}) | |
| 101 | + | } | |
| 102 | + | t.txDepth++ | |
| 103 | + | t.mu.Unlock() | |
| 104 | + | } | |
| 66 | 105 | ||
| 67 | 106 | func (t *attachTestTerminal) EndOutput(success bool) { | |
| 68 | 107 | t.mu.Lock() | |
| 69 | 108 | defer t.mu.Unlock() | |
| 70 | 109 | t.successes = append(t.successes, success) | |
| 110 | + | if t.txDepth == 0 { | |
| 111 | + | return | |
| 112 | + | } | |
| 113 | + | if !success { | |
| 114 | + | t.txFailed = true | |
| 115 | + | } | |
| 116 | + | t.txDepth-- | |
| 117 | + | if t.txDepth != 0 { | |
| 118 | + | return | |
| 119 | + | } | |
| 120 | + | if t.txFailed { | |
| 121 | + | t.available = false | |
| 122 | + | t.pending = ports.UIContext{} | |
| 123 | + | return | |
| 124 | + | } | |
| 125 | + | t.publications = append(t.publications, t.pending) | |
| 126 | + | t.revision++ | |
| 127 | + | t.events = append(t.events, attachTestEvent{kind: "drain", status: t.pending.Status, state: t.pending.OutputState}) | |
| 128 | + | t.available = true | |
| 129 | + | t.pending = ports.UIContext{} | |
| 71 | 130 | } | |
| 72 | 131 | ||
| 73 | 132 | func (t *attachTestTerminal) PublishContext(uiContext ports.UIContext) error { | |
| 74 | - | if t.publishEntered != nil { | |
| 133 | + | if t.publishMatch == nil || t.publishMatch(uiContext) { | |
| 134 | + | if t.publishEntered != nil { | |
| 135 | + | select { | |
| 136 | + | case t.publishEntered <- struct{}{}: | |
| 137 | + | default: | |
| 138 | + | } | |
| 139 | + | } | |
| 140 | + | if t.publishRelease != nil { | |
| 141 | + | <-t.publishRelease | |
| 142 | + | } | |
| 143 | + | } | |
| 144 | + | t.mu.Lock() | |
| 145 | + | defer t.mu.Unlock() | |
| 146 | + | if !t.available { | |
| 147 | + | return ports.ErrUIUnavailable | |
| 148 | + | } | |
| 149 | + | if uiContext.AttachmentHandle == "" && len(t.publications) != 0 { | |
| 150 | + | uiContext.AttachmentHandle = t.publications[len(t.publications)-1].AttachmentHandle | |
| 151 | + | } | |
| 152 | + | if uiContext.Generation == 0 && len(t.publications) != 0 { | |
| 153 | + | uiContext.Generation = t.publications[len(t.publications)-1].Generation | |
| 154 | + | } | |
| 155 | + | t.publications = append(t.publications, uiContext) | |
| 156 | + | t.revision++ | |
| 157 | + | t.events = append(t.events, attachTestEvent{kind: "publish", status: uiContext.Status, state: uiContext.OutputState}) | |
| 158 | + | if uiContext.Status == ports.UIStatusDetached && t.detachedSignals != nil { | |
| 75 | 159 | select { | |
| 76 | - | case t.publishEntered <- struct{}{}: | |
| 160 | + | case t.detachedSignals <- struct{}{}: | |
| 77 | 161 | default: | |
| 78 | 162 | } | |
| 79 | 163 | } | |
| 80 | - | if t.publishRelease != nil { | |
| 81 | - | <-t.publishRelease | |
| 164 | + | return nil | |
| 165 | + | } | |
| 166 | + | ||
| 167 | + | func (t *attachTestTerminal) Snapshot() (ports.UISnapshot, error) { | |
| 168 | + | t.mu.Lock() | |
| 169 | + | defer t.mu.Unlock() | |
| 170 | + | var ctx ports.UIContext | |
| 171 | + | if len(t.publications) != 0 { | |
| 172 | + | ctx = t.publications[len(t.publications)-1] | |
| 82 | 173 | } | |
| 174 | + | if !t.available || t.revision == 0 { | |
| 175 | + | return ports.UISnapshot{}, ports.ErrUIUnavailable | |
| 176 | + | } | |
| 177 | + | return ports.UISnapshot{Revision: t.revision, Context: ctx}, nil | |
| 178 | + | } | |
| 179 | + | ||
| 180 | + | func (t *attachTestTerminal) Changes() <-chan struct{} { | |
| 83 | 181 | t.mu.Lock() | |
| 84 | 182 | defer t.mu.Unlock() | |
| 85 | - | t.publications = append(t.publications, uiContext) | |
| 86 | - | return nil | |
| 183 | + | if t.changes == nil { | |
| 184 | + | t.changes = make(chan struct{}) | |
| 185 | + | } | |
| 186 | + | return t.changes | |
| 187 | + | } | |
| 188 | + | ||
| 189 | + | func (t *attachTestTerminal) publicationList() []ports.UIContext { | |
| 190 | + | t.mu.Lock() | |
| 191 | + | defer t.mu.Unlock() | |
| 192 | + | return append([]ports.UIContext(nil), t.publications...) | |
| 193 | + | } | |
| 194 | + | ||
| 195 | + | // eventIndex reports the position of the first matching transaction event, or | |
| 196 | + | // -1 when it never happened. | |
| 197 | + | func (t *attachTestTerminal) eventIndex(kind string, status ports.UIPresentationStatus, state uint64) int { | |
| 198 | + | t.mu.Lock() | |
| 199 | + | defer t.mu.Unlock() | |
| 200 | + | for i, event := range t.events { | |
| 201 | + | if event.kind == kind && event.status == status && event.state == state { | |
| 202 | + | return i | |
| 203 | + | } | |
| 204 | + | } | |
| 205 | + | return -1 | |
| 87 | 206 | } | |
| 88 | 207 | ||
| 89 | 208 | func (t *attachTestTerminal) written() string { | |
| @@ -329,6 +448,14 @@ func awaitAttachedState(t *testing.T, sup *Supervisor) { | |||
| 329 | 448 | require.Eventually(t, func() bool { return sup.State().Presentation == PresentAttached }, 5*time.Second, time.Millisecond) | |
| 330 | 449 | } | |
| 331 | 450 | ||
| 451 | + | // authorityState reports the current foreground slot state without racing the | |
| 452 | + | // supervisor's own transitions. | |
| 453 | + | func (a *attachmentAuthority) authorityState() attachmentAuthorityState { | |
| 454 | + | a.mu.Lock() | |
| 455 | + | defer a.mu.Unlock() | |
| 456 | + | return a.state | |
| 457 | + | } | |
| 458 | + | ||
| 332 | 459 | // awaitPickerState waits until the supervisor is back in the picker. | |
| 333 | 460 | func awaitPickerState(t *testing.T, sup *Supervisor) { | |
| 334 | 461 | t.Helper() | |
| @@ -376,6 +503,360 @@ func TestSupervisorReducerAttachmentPresentation(t *testing.T) { | |||
| 376 | 503 | // TestSupervisorAttachmentLocalRemoteParity proves the same supervisor path | |
| 377 | 504 | // opens the exact resolved request for a local and a remote selection, with no | |
| 378 | 505 | // dialer choice in the client. | |
| 506 | + | func TestSupervisorUIActionBindingUsesSingleAttachmentPump(t *testing.T) { | |
| 507 | + | picker := newAttachTestPicker() | |
| 508 | + | var stream *sessionTestStream | |
| 509 | + | var streamMu sync.Mutex | |
| 510 | + | var ui *UI | |
| 511 | + | harness := startAttachHarnessConfig(t, picker, func(cfg *SupervisorConfig) { | |
| 512 | + | ui = NewUI(cfg.Terminal.(ports.UIState), cfg.Clock) | |
| 513 | + | cfg.UI = ui | |
| 514 | + | }) | |
| 515 | + | harness.service.setOpenStream(func(context.Context, ports.BrokerOpenStreamRequest) (ports.BrokerLogicalConnection, error) { | |
| 516 | + | admitted := newSessionTestStream() | |
| 517 | + | streamMu.Lock() | |
| 518 | + | stream = admitted | |
| 519 | + | streamMu.Unlock() | |
| 520 | + | return admitted, nil | |
| 521 | + | }) | |
| 522 | + | ||
| 523 | + | // Picker ownership has no UI binding, so this would fail if actions could | |
| 524 | + | // collide with or bypass the picker's exclusive input claim. | |
| 525 | + | _, err := ui.Action(t.Context(), ports.UIActionRequest{Attachment: ui.Handle(), Generation: 1, Text: "picker"}) | |
| 526 | + | var pickerErr *ports.UIError | |
| 527 | + | require.ErrorAs(t, err, &pickerErr) | |
| 528 | + | require.Equal(t, ports.UIErrUnavailable, pickerErr.Code) | |
| 529 | + | ||
| 530 | + | picker.commit(sessionTestRequest(true)) | |
| 531 | + | require.Eventually(t, func() bool { | |
| 532 | + | streamMu.Lock() | |
| 533 | + | defer streamMu.Unlock() | |
| 534 | + | return stream != nil && len(stream.messages()) > 0 | |
| 535 | + | }, 5*time.Second, time.Millisecond) | |
| 536 | + | streamMu.Lock() | |
| 537 | + | admitted := stream | |
| 538 | + | streamMu.Unlock() | |
| 539 | + | deliverReadyStream(t, admitted) | |
| 540 | + | awaitAttachedState(t, harness.sup) | |
| 541 | + | attachedOutput := sessionTestOutput(2, "\x1b[Hattached") | |
| 542 | + | attachedOutput.Full = false | |
| 543 | + | attachedOutput.Base = 1 | |
| 544 | + | attachedOutput.New = 2 | |
| 545 | + | admitted.deliver(attachedOutput) | |
| 546 | + | ||
| 547 | + | var snapshot ports.UISnapshot | |
| 548 | + | require.Eventually(t, func() bool { | |
| 549 | + | var captureErr error | |
| 550 | + | snapshot, captureErr = ui.Capture(ui.Handle()) | |
| 551 | + | return captureErr == nil && snapshot.Context.Status == ports.UIStatusAttached | |
| 552 | + | }, 5*time.Second, time.Millisecond) | |
| 553 | + | require.Equal(t, ports.UIStatusAttached, snapshot.Context.Status) | |
| 554 | + | require.NotZero(t, snapshot.Context.Generation) | |
| 555 | + | actionDone := make(chan error, 1) | |
| 556 | + | go func() { | |
| 557 | + | _, actionErr := ui.Action(t.Context(), ports.UIActionRequest{Attachment: ui.Handle(), Generation: snapshot.Context.Generation, Text: "automated"}) | |
| 558 | + | actionDone <- actionErr | |
| 559 | + | }() | |
| 560 | + | var actionID uint64 | |
| 561 | + | require.Eventually(t, func() bool { | |
| 562 | + | var fenced bool | |
| 563 | + | for _, message := range admitted.messages() { | |
| 564 | + | switch typed := message.(type) { | |
| 565 | + | case protocol.Input: | |
| 566 | + | if string(typed.Data) == "automated" { | |
| 567 | + | actionID = typed.ActionID | |
| 568 | + | } | |
| 569 | + | case protocol.UIFence: | |
| 570 | + | fenced = actionID != 0 && typed.ActionID == actionID | |
| 571 | + | } | |
| 572 | + | } | |
| 573 | + | return fenced | |
| 574 | + | }, 5*time.Second, time.Millisecond, "admitted actions must be followed by their UI fence") | |
| 575 | + | processedOutput := sessionTestOutput(3, "\x1b[Hprocessed") | |
| 576 | + | processedOutput.Full = false | |
| 577 | + | processedOutput.Base = 2 | |
| 578 | + | processedOutput.New = 3 | |
| 579 | + | processedOutput.Context.Publication = 3 | |
| 580 | + | admitted.deliver(processedOutput) | |
| 581 | + | var boundary ports.UIActionResult | |
| 582 | + | require.Eventually(t, func() bool { | |
| 583 | + | ui.mu.Lock() | |
| 584 | + | boundary = ui.boundary | |
| 585 | + | ui.mu.Unlock() | |
| 586 | + | return boundary.Revision != 0 && boundary.Context.OutputState == processedOutput.New && boundary.Context.ViewPublication == processedOutput.Context.Publication | |
| 587 | + | }, 5*time.Second, time.Millisecond, "the action-specific output must be the committed processing boundary") | |
| 588 | + | admitted.deliver(protocol.UIReceipt{ActionID: actionID, Epoch: 1, State: 3, ViewPublication: 3, Outcome: protocol.UIReceiptProcessed}) | |
| 589 | + | select { | |
| 590 | + | case actionErr := <-actionDone: | |
| 591 | + | require.NoError(t, actionErr, "receipt routing must complete the admitted action as processed") | |
| 592 | + | case <-time.After(5 * time.Second): | |
| 593 | + | t.Fatal("processed UI action did not complete") | |
| 594 | + | } | |
| 595 | + | ||
| 596 | + | // Ending the stream releases before picker reacquisition. A stale request | |
| 597 | + | // would be admitted here if finalization forgot to retire the binding. | |
| 598 | + | admitted.deliver(protocol.Detached{}) | |
| 599 | + | awaitPickerState(t, harness.sup) | |
| 600 | + | ui.mu.Lock() | |
| 601 | + | require.Nil(t, ui.input, "attachment finalization must revoke the UI pump binding") | |
| 602 | + | require.Zero(t, ui.consumer, "attachment finalization must revoke the UI consumer") | |
| 603 | + | require.Nil(t, ui.foreground, "attachment finalization must retire the UI foreground") | |
| 604 | + | ui.mu.Unlock() | |
| 605 | + | _, err = ui.Action(t.Context(), ports.UIActionRequest{Attachment: ui.Handle(), Generation: snapshot.Context.Generation, Text: "stale"}) | |
| 606 | + | var staleErr *ports.UIError | |
| 607 | + | require.ErrorAs(t, err, &staleErr) | |
| 608 | + | require.Equal(t, ports.UIErrUnavailable, staleErr.Code) | |
| 609 | + | ||
| 610 | + | // Reconnecting installs a fresh UI-owned generation. The prior generation | |
| 611 | + | // remains stale even while another attachment is actionable. | |
| 612 | + | reconnected := newSessionTestStream() | |
| 613 | + | harness.service.setOpenStream(func(context.Context, ports.BrokerOpenStreamRequest) (ports.BrokerLogicalConnection, error) { | |
| 614 | + | return reconnected, nil | |
| 615 | + | }) | |
| 616 | + | picker.commit(sessionTestRequest(true)) | |
| 617 | + | awaitHello(t, reconnected) | |
| 618 | + | deliverReadyStream(t, reconnected) | |
| 619 | + | awaitAttachedState(t, harness.sup) | |
| 620 | + | reconnectedOutput := sessionTestOutput(2, "\x1b[Hreconnected") | |
| 621 | + | reconnectedOutput.Full = false | |
| 622 | + | reconnectedOutput.Base = 1 | |
| 623 | + | reconnectedOutput.New = 2 | |
| 624 | + | reconnected.deliver(reconnectedOutput) | |
| 625 | + | require.Eventually(t, func() bool { | |
| 626 | + | current, captureErr := ui.Capture(ui.Handle()) | |
| 627 | + | return captureErr == nil && current.Context.Status == ports.UIStatusAttached && current.Context.Generation != snapshot.Context.Generation | |
| 628 | + | }, 5*time.Second, time.Millisecond) | |
| 629 | + | _, err = ui.Action(t.Context(), ports.UIActionRequest{Attachment: ui.Handle(), Generation: snapshot.Context.Generation, Text: "still-stale"}) | |
| 630 | + | staleErr = nil | |
| 631 | + | require.ErrorAs(t, err, &staleErr) | |
| 632 | + | require.Equal(t, ports.UIErrStaleAttachment, staleErr.Code) | |
| 633 | + | select { | |
| 634 | + | case <-actionDone: // completion may be receipt-driven or retired on release. | |
| 635 | + | default: | |
| 636 | + | } | |
| 637 | + | } | |
| 638 | + | ||
| 639 | + | // TestSupervisorUIActionReleaseWaitsForInFlightOutput pins GO-201: finalization | |
| 640 | + | // must not release the UI foreground binding while a foreground output | |
| 641 | + | // transaction is in flight. The release publishes Detached, so releasing it | |
| 642 | + | // early would publish an unattached presentation while a newer session frame | |
| 643 | + | // was still being written, and would return the picker claim before the last | |
| 644 | + | // frame committed. The mutation this test pins moves releaseForeground before | |
| 645 | + | // lease.stop in attachmentForeground.finalize. | |
| 646 | + | func TestSupervisorUIActionReleaseWaitsForInFlightOutput(t *testing.T) { | |
| 647 | + | const inFlightState = 2 | |
| 648 | + | picker := newAttachTestPicker() | |
| 649 | + | var ( | |
| 650 | + | streamMu sync.Mutex | |
| 651 | + | stream *sessionTestStream | |
| 652 | + | ui *UI | |
| 653 | + | ) | |
| 654 | + | harness := startAttachHarnessConfig(t, picker, func(cfg *SupervisorConfig) { | |
| 655 | + | ui = NewUI(cfg.Terminal.(ports.UIState), cfg.Clock) | |
| 656 | + | cfg.UI = ui | |
| 657 | + | }) | |
| 658 | + | harness.service.setOpenStream(func(context.Context, ports.BrokerOpenStreamRequest) (ports.BrokerLogicalConnection, error) { | |
| 659 | + | admitted := newSessionTestStream() | |
| 660 | + | streamMu.Lock() | |
| 661 | + | stream = admitted | |
| 662 | + | streamMu.Unlock() | |
| 663 | + | return admitted, nil | |
| 664 | + | }) | |
| 665 | + | ||
| 666 | + | // Installed before the commit, so the supervisor's publication goroutine | |
| 667 | + | // always observes them: only the attached frame with inFlightState blocks. | |
| 668 | + | terminal := harness.terminal | |
| 669 | + | entered := make(chan struct{}, 1) | |
| 670 | + | release := make(chan struct{}) | |
| 671 | + | var releaseOnce sync.Once | |
| 672 | + | releaseOutput := func() { releaseOnce.Do(func() { close(release) }) } | |
| 673 | + | defer releaseOutput() | |
| 674 | + | detached := make(chan struct{}, 4) | |
| 675 | + | terminal.publishMatch = func(ctx ports.UIContext) bool { | |
| 676 | + | return ctx.Status == ports.UIStatusAttached && ctx.OutputState == inFlightState | |
| 677 | + | } | |
| 678 | + | terminal.publishEntered = entered | |
| 679 | + | terminal.publishRelease = release | |
| 680 | + | terminal.detachedSignals = detached | |
| 681 | + | ||
| 682 | + | picker.commit(sessionTestRequest(true)) | |
| 683 | + | awaitStreamHello(t, &streamMu, &stream) | |
| 684 | + | streamMu.Lock() | |
| 685 | + | admitted := stream | |
| 686 | + | streamMu.Unlock() | |
| 687 | + | deliverReadyStream(t, admitted) | |
| 688 | + | awaitAttachedState(t, harness.sup) | |
| 689 | + | ||
| 690 | + | // Block one foreground output transaction after BeginOutput and before its | |
| 691 | + | // commit, so the send lease is held while finalization begins. | |
| 692 | + | inFlight := sessionTestOutput(inFlightState, "\x1b[Hin-flight") | |
| 693 | + | inFlight.Full = false | |
| 694 | + | inFlight.Base = 1 | |
| 695 | + | inFlight.New = inFlightState | |
| 696 | + | admitted.deliver(inFlight) | |
| 697 | + | select { | |
| 698 | + | case <-entered: | |
| 699 | + | case <-time.After(5 * time.Second): | |
| 700 | + | t.Fatal("the foreground output transaction never reached its publication boundary") | |
| 701 | + | } | |
| 702 | + | require.Equal(t, -1, terminal.eventIndex("drain", ports.UIStatusAttached, inFlightState), "the in-flight frame must still hold its transaction") | |
| 703 | + | ||
| 704 | + | // Cancel the attachment while that transaction is in flight. | |
| 705 | + | harness.service.lose(errors.New("attachment lost")) | |
| 706 | + | require.Eventually(t, func() bool { | |
| 707 | + | return harness.sup.attachments.authority.authorityState() == authorityFinalizing | |
| 708 | + | }, 5*time.Second, time.Millisecond, "attachment finalization must begin while the frame is in flight") | |
| 709 | + | ||
| 710 | + | // Finalization cannot complete, and Detached cannot be published, until the | |
| 711 | + | // in-flight transaction drains. | |
| 712 | + | require.Never(t, func() bool { | |
| 713 | + | select { | |
| 714 | + | case <-detached: | |
| 715 | + | return true | |
| 716 | + | default: | |
| 717 | + | return false | |
| 718 | + | } | |
| 719 | + | }, 50*time.Millisecond, time.Millisecond, "Detached must not publish while a foreground transaction is in flight") | |
| 720 | + | require.Equal(t, authorityFinalizing, harness.sup.attachments.authority.authorityState(), "finalize still waits on the output transaction") | |
| 721 | + | require.False(t, admitted.closedNow(), "the retired stream is released only after the drain") | |
| 722 | + | require.False(t, picker.owns(), "the picker claim returns only after the drain") | |
| 723 | + | require.Equal(t, PresentAttached, harness.sup.State().Presentation, "the attachment is still presented while its last frame drains") | |
| 724 | + | ui.mu.Lock() | |
| 725 | + | require.NotNil(t, ui.input, "the UI foreground binding is released only after the drain") | |
| 726 | + | require.NotNil(t, ui.foreground, "the UI foreground context is retired only after the drain") | |
| 727 | + | ui.mu.Unlock() | |
| 728 | + | ||
| 729 | + | releaseOutput() | |
| 730 | + | select { | |
| 731 | + | case <-detached: | |
| 732 | + | case <-time.After(5 * time.Second): | |
| 733 | + | t.Fatal("attachment finalization never published Detached") | |
| 734 | + | } | |
| 735 | + | ||
| 736 | + | // The newer frame commits before Detached, and Detached carries the | |
| 737 | + | // committed frame rather than the superseded one. | |
| 738 | + | drained := terminal.eventIndex("drain", ports.UIStatusAttached, inFlightState) | |
| 739 | + | published := terminal.eventIndex("publish", ports.UIStatusDetached, inFlightState) | |
| 740 | + | require.GreaterOrEqual(t, drained, 0, "the in-flight frame must commit once it drains") | |
| 741 | + | require.Greater(t, published, drained, "Detached must be published only after the newer frame commits") | |
| 742 | + | publications := terminal.publicationList() | |
| 743 | + | require.GreaterOrEqual(t, len(publications), 2) | |
| 744 | + | last := publications[len(publications)-1] | |
| 745 | + | require.Equal(t, ports.UIStatusDetached, last.Status) | |
| 746 | + | require.Equal(t, uint64(inFlightState), last.OutputState, "Detached carries the committed newer frame") | |
| 747 | + | prior := publications[len(publications)-2] | |
| 748 | + | require.Equal(t, ports.UIStatusAttached, prior.Status) | |
| 749 | + | require.Equal(t, uint64(inFlightState), prior.OutputState) | |
| 750 | + | ||
| 751 | + | // Finalization completes and hands the picker claim back. | |
| 752 | + | require.Eventually(t, func() bool { | |
| 753 | + | ui.mu.Lock() | |
| 754 | + | released := ui.input == nil && ui.consumer == 0 && ui.foreground == nil | |
| 755 | + | ui.mu.Unlock() | |
| 756 | + | return released && picker.owns() && harness.sup.attachments.authority.authorityState() == authorityIdle && admitted.closedNow() | |
| 757 | + | }, 5*time.Second, time.Millisecond, "finalization must complete and return the picker claim") | |
| 758 | + | require.Equal(t, PresentPicker, harness.sup.State().Presentation) | |
| 759 | + | } | |
| 760 | + | ||
| 761 | + | func TestSupervisorUIViewUpdateCommitsActionBoundaryWithoutBytes(t *testing.T) { | |
| 762 | + | picker := newAttachTestPicker() | |
| 763 | + | var ui *UI | |
| 764 | + | harness := startAttachHarnessConfig(t, picker, func(cfg *SupervisorConfig) { | |
| 765 | + | ui = NewUI(cfg.Terminal.(ports.UIState), cfg.Clock) | |
| 766 | + | cfg.UI = ui | |
| 767 | + | }) | |
| 768 | + | stream := newSessionTestStream() | |
| 769 | + | harness.service.setOpenStream(func(context.Context, ports.BrokerOpenStreamRequest) (ports.BrokerLogicalConnection, error) { | |
| 770 | + | return stream, nil | |
| 771 | + | }) | |
| 772 | + | picker.commit(sessionTestRequest(true)) | |
| 773 | + | awaitHello(t, stream) | |
| 774 | + | deliverReadyStream(t, stream) | |
| 775 | + | awaitAttachedState(t, harness.sup) | |
| 776 | + | initialOutput := sessionTestOutput(2, "\x1b[Hmetadata-base") | |
| 777 | + | initialOutput.Full = false | |
| 778 | + | initialOutput.Base = 1 | |
| 779 | + | initialOutput.New = 2 | |
| 780 | + | initialOutput.Context.Publication = 2 | |
| 781 | + | stream.deliver(initialOutput) | |
| 782 | + | ||
| 783 | + | var snapshot ports.UISnapshot | |
| 784 | + | require.Eventually(t, func() bool { | |
| 785 | + | var err error | |
| 786 | + | snapshot, err = ui.Capture(ui.Handle()) | |
| 787 | + | return err == nil && snapshot.Context.Status == ports.UIStatusAttached && snapshot.Context.OutputState == 2 | |
| 788 | + | }, 5*time.Second, time.Millisecond) | |
| 789 | + | actionDone := make(chan error, 1) | |
| 790 | + | go func() { | |
| 791 | + | _, err := ui.Action(t.Context(), ports.UIActionRequest{Attachment: ui.Handle(), Generation: snapshot.Context.Generation, Text: "metadata"}) | |
| 792 | + | actionDone <- err | |
| 793 | + | }() | |
| 794 | + | var actionID uint64 | |
| 795 | + | require.Eventually(t, func() bool { | |
| 796 | + | for _, message := range stream.messages() { | |
| 797 | + | if fence, ok := message.(protocol.UIFence); ok { | |
| 798 | + | actionID = fence.ActionID | |
| 799 | + | return actionID != 0 | |
| 800 | + | } | |
| 801 | + | } | |
| 802 | + | return false | |
| 803 | + | }, 5*time.Second, time.Millisecond) | |
| 804 | + | view := protocol.ViewContext{Publication: 3, Route: snapshot.Context.Route, TabID: "metadata-tab", FocusedPaneID: "metadata-pane"} | |
| 805 | + | view.Route.Target = protocol.ExactSessionTarget{LifecycleID: domain.SessionLifecycleID{1}, SessionName: "alpha"} | |
| 806 | + | stream.deliver(protocol.UIViewUpdate{Epoch: 1, State: 2, Context: view}) | |
| 807 | + | require.Eventually(t, func() bool { | |
| 808 | + | current, err := ui.Capture(ui.Handle()) | |
| 809 | + | return err == nil && current.Context.ViewPublication == 3 && current.Context.TabID == "metadata-tab" | |
| 810 | + | }, 5*time.Second, time.Millisecond, "metadata-only update must commit through the foreground transaction") | |
| 811 | + | stream.deliver(protocol.UIReceipt{ActionID: actionID, Epoch: 1, State: 2, ViewPublication: 3, Outcome: protocol.UIReceiptProcessed}) | |
| 812 | + | select { | |
| 813 | + | case err := <-actionDone: | |
| 814 | + | require.NoError(t, err) | |
| 815 | + | case <-time.After(5 * time.Second): | |
| 816 | + | t.Fatal("metadata-only committed boundary did not complete action") | |
| 817 | + | } | |
| 818 | + | } | |
| 819 | + | ||
| 820 | + | func TestSupervisorUIViewUpdateRequestsOneCoalescedReset(t *testing.T) { | |
| 821 | + | picker := newAttachTestPicker() | |
| 822 | + | harness := startAttachHarness(t, picker) | |
| 823 | + | stream := newSessionTestStream() | |
| 824 | + | harness.service.setOpenStream(func(context.Context, ports.BrokerOpenStreamRequest) (ports.BrokerLogicalConnection, error) { | |
| 825 | + | return stream, nil | |
| 826 | + | }) | |
| 827 | + | picker.commit(sessionTestRequest(true)) | |
| 828 | + | awaitHello(t, stream) | |
| 829 | + | deliverReadyStream(t, stream) | |
| 830 | + | awaitAttachedState(t, harness.sup) | |
| 831 | + | bad := protocol.UIViewUpdate{Epoch: 9, State: 9, Context: protocol.ViewContext{Publication: 2, Route: protocol.CommittedRouteIdentity{Target: protocol.ExactSessionTarget{LifecycleID: domain.SessionLifecycleID{1}, SessionName: "alpha"}}, TabID: "future", FocusedPaneID: "future"}} | |
| 832 | + | stream.deliver(bad) | |
| 833 | + | stream.deliver(bad) | |
| 834 | + | require.Eventually(t, func() bool { | |
| 835 | + | count := 0 | |
| 836 | + | for _, message := range stream.messages() { | |
| 837 | + | if _, ok := message.(protocol.OutputResetRequest); ok { | |
| 838 | + | count++ | |
| 839 | + | } | |
| 840 | + | } | |
| 841 | + | return count == 1 | |
| 842 | + | }, 5*time.Second, time.Millisecond, "future view dependencies must request one coalesced reset") | |
| 843 | + | } | |
| 844 | + | ||
| 845 | + | func TestSupervisorWithoutUIDoesNotBindActions(t *testing.T) { | |
| 846 | + | picker := newAttachTestPicker() | |
| 847 | + | harness := startAttachHarness(t, picker) | |
| 848 | + | require.Nil(t, harness.sup.cfg.UI, "the optional seam must preserve publication-only composition") | |
| 849 | + | require.Nil(t, harness.sup.attachments.actionUI, "mutation that binds an implicit UI changes no-UI behavior") | |
| 850 | + | } | |
| 851 | + | ||
| 852 | + | func TestUIActionWithoutInputReturnsUnavailable(t *testing.T) { | |
| 853 | + | ui := NewUI(&attachTestTerminal{}, newSupervisorTestClock()) | |
| 854 | + | _, err := ui.Action(t.Context(), ports.UIActionRequest{Attachment: ui.Handle(), Generation: 1, Text: "x"}) | |
| 855 | + | var uiErr *ports.UIError | |
| 856 | + | require.ErrorAs(t, err, &uiErr) | |
| 857 | + | require.Equal(t, ports.UIErrUnavailable, uiErr.Code) | |
| 858 | + | } | |
| 859 | + | ||
| 379 | 860 | func TestSupervisorAttachmentLocalRemoteParity(t *testing.T) { | |
| 380 | 861 | for _, local := range []bool{true, false} { | |
| 381 | 862 | name := "remote" | |
| @@ -31,9 +31,33 @@ type attachTestTerminal struct { | |||||
| 31 | flushes int | 31 | flushes int | ||
| 32 | restores int | 32 | restores int | ||
| 33 | publications []ports.UIContext | 33 | publications []ports.UIContext | ||
| 34 | + | pending ports.UIContext | |||
| 35 | + | revision uint64 | |||
| 36 | + | txDepth int | |||
| 37 | + | txFailed bool | |||
| 38 | + | available bool | |||
| 34 | successes []bool | 39 | successes []bool | ||
| 35 | publishEntered chan struct{} | 40 | publishEntered chan struct{} | ||
| 36 | publishRelease <-chan struct{} | 41 | publishRelease <-chan struct{} | ||
| 42 | + | // publishMatch narrows the publication block to one matching transaction. | |||
| 43 | + | // A nil predicate blocks every publication, which is what the existing | |||
| 44 | + | // initial-publication tests rely on. | |||
| 45 | + | publishMatch func(ports.UIContext) bool | |||
| 46 | + | // detachedSignals receives one signal per committed Detached publication, so | |||
| 47 | + | // a test observes attachment release without polling. | |||
| 48 | + | detachedSignals chan struct{} | |||
| 49 | + | // events records the transaction order: each begin, each publication, and | |||
| 50 | + | // each successful drain. A test proves release ordering against it instead | |||
| 51 | + | // of inferring it from separate slices. | |||
| 52 | + | events []attachTestEvent | |||
| 53 | + | changes chan struct{} | |||
| 54 | + | } | |||
| 55 | + | ||||
| 56 | + | // attachTestEvent is one ordered terminal transaction event. | |||
| 57 | + | type attachTestEvent struct { | |||
| 58 | + | kind string // begin | publish | drain | |||
| 59 | + | status ports.UIPresentationStatus | |||
| 60 | + | state uint64 | |||
| 37 | } | 61 | } | ||
| 38 | 62 | ||||
| 39 | func (t *attachTestTerminal) EnterRaw() (func() error, error) { | 63 | func (t *attachTestTerminal) EnterRaw() (func() error, error) { | ||
| @@ -62,28 +86,123 @@ func (t *attachTestTerminal) Flush() error { | |||||
| 62 | return nil | 86 | return nil | ||
| 63 | } | 87 | } | ||
| 64 | 88 | ||||
| 65 | - | func (t *attachTestTerminal) BeginOutput(ports.UIContext) {} | 89 | + | func (t *attachTestTerminal) BeginOutput(ctx ports.UIContext) { |
| 90 | + | t.mu.Lock() | |||
| 91 | + | if ctx.AttachmentHandle == "" && len(t.publications) != 0 { | |||
| 92 | + | ctx.AttachmentHandle = t.publications[len(t.publications)-1].AttachmentHandle | |||
| 93 | + | } | |||
| 94 | + | if ctx.Generation == 0 && len(t.publications) != 0 { | |||
| 95 | + | ctx.Generation = t.publications[len(t.publications)-1].Generation | |||
| 96 | + | } | |||
| 97 | + | if t.txDepth == 0 { | |||
| 98 | + | t.pending = ctx | |||
| 99 | + | t.txFailed = false | |||
| 100 | + | t.events = append(t.events, attachTestEvent{kind: "begin", status: ctx.Status, state: ctx.OutputState}) | |||
| 101 | + | } | |||
| 102 | + | t.txDepth++ | |||
| 103 | + | t.mu.Unlock() | |||
| 104 | + | } | |||
| 66 | 105 | ||||
| 67 | func (t *attachTestTerminal) EndOutput(success bool) { | 106 | func (t *attachTestTerminal) EndOutput(success bool) { | ||
| 68 | t.mu.Lock() | 107 | t.mu.Lock() | ||
| 69 | defer t.mu.Unlock() | 108 | defer t.mu.Unlock() | ||
| 70 | t.successes = append(t.successes, success) | 109 | t.successes = append(t.successes, success) | ||
| 110 | + | if t.txDepth == 0 { | |||
| 111 | + | return | |||
| 112 | + | } | |||
| 113 | + | if !success { | |||
| 114 | + | t.txFailed = true | |||
| 115 | + | } | |||
| 116 | + | t.txDepth-- | |||
| 117 | + | if t.txDepth != 0 { | |||
| 118 | + | return | |||
| 119 | + | } | |||
| 120 | + | if t.txFailed { | |||
| 121 | + | t.available = false | |||
| 122 | + | t.pending = ports.UIContext{} | |||
| 123 | + | return | |||
| 124 | + | } | |||
| 125 | + | t.publications = append(t.publications, t.pending) | |||
| 126 | + | t.revision++ | |||
| 127 | + | t.events = append(t.events, attachTestEvent{kind: "drain", status: t.pending.Status, state: t.pending.OutputState}) | |||
| 128 | + | t.available = true | |||
| 129 | + | t.pending = ports.UIContext{} | |||
| 71 | } | 130 | } | ||
| 72 | 131 | ||||
| 73 | func (t *attachTestTerminal) PublishContext(uiContext ports.UIContext) error { | 132 | func (t *attachTestTerminal) PublishContext(uiContext ports.UIContext) error { | ||
| 74 | - | if t.publishEntered != nil { | 133 | + | if t.publishMatch == nil || t.publishMatch(uiContext) { |
| 134 | + | if t.publishEntered != nil { | |||
| 135 | + | select { | |||
| 136 | + | case t.publishEntered <- struct{}{}: | |||
| 137 | + | default: | |||
| 138 | + | } | |||
| 139 | + | } | |||
| 140 | + | if t.publishRelease != nil { | |||
| 141 | + | <-t.publishRelease | |||
| 142 | + | } | |||
| 143 | + | } | |||
| 144 | + | t.mu.Lock() | |||
| 145 | + | defer t.mu.Unlock() | |||
| 146 | + | if !t.available { | |||
| 147 | + | return ports.ErrUIUnavailable | |||
| 148 | + | } | |||
| 149 | + | if uiContext.AttachmentHandle == "" && len(t.publications) != 0 { | |||
| 150 | + | uiContext.AttachmentHandle = t.publications[len(t.publications)-1].AttachmentHandle | |||
| 151 | + | } | |||
| 152 | + | if uiContext.Generation == 0 && len(t.publications) != 0 { | |||
| 153 | + | uiContext.Generation = t.publications[len(t.publications)-1].Generation | |||
| 154 | + | } | |||
| 155 | + | t.publications = append(t.publications, uiContext) | |||
| 156 | + | t.revision++ | |||
| 157 | + | t.events = append(t.events, attachTestEvent{kind: "publish", status: uiContext.Status, state: uiContext.OutputState}) | |||
| 158 | + | if uiContext.Status == ports.UIStatusDetached && t.detachedSignals != nil { | |||
| 75 | select { | 159 | select { | ||
| 76 | - | case t.publishEntered <- struct{}{}: | 160 | + | case t.detachedSignals <- struct{}{}: |
| 77 | default: | 161 | default: | ||
| 78 | } | 162 | } | ||
| 79 | } | 163 | } | ||
| 80 | - | if t.publishRelease != nil { | 164 | + | return nil |
| 81 | - | <-t.publishRelease | 165 | + | } |
| 166 | + | ||||
| 167 | + | func (t *attachTestTerminal) Snapshot() (ports.UISnapshot, error) { | |||
| 168 | + | t.mu.Lock() | |||
| 169 | + | defer t.mu.Unlock() | |||
| 170 | + | var ctx ports.UIContext | |||
| 171 | + | if len(t.publications) != 0 { | |||
| 172 | + | ctx = t.publications[len(t.publications)-1] | |||
| 82 | } | 173 | } | ||
| 174 | + | if !t.available || t.revision == 0 { | |||
| 175 | + | return ports.UISnapshot{}, ports.ErrUIUnavailable | |||
| 176 | + | } | |||
| 177 | + | return ports.UISnapshot{Revision: t.revision, Context: ctx}, nil | |||
| 178 | + | } | |||
| 179 | + | ||||
| 180 | + | func (t *attachTestTerminal) Changes() <-chan struct{} { | |||
| 83 | t.mu.Lock() | 181 | t.mu.Lock() | ||
| 84 | defer t.mu.Unlock() | 182 | defer t.mu.Unlock() | ||
| 85 | - | t.publications = append(t.publications, uiContext) | 183 | + | if t.changes == nil { |
| 86 | - | return nil | 184 | + | t.changes = make(chan struct{}) |
| 185 | + | } | |||
| 186 | + | return t.changes | |||
| 187 | + | } | |||
| 188 | + | ||||
| 189 | + | func (t *attachTestTerminal) publicationList() []ports.UIContext { | |||
| 190 | + | t.mu.Lock() | |||
| 191 | + | defer t.mu.Unlock() | |||
| 192 | + | return append([]ports.UIContext(nil), t.publications...) | |||
| 193 | + | } | |||
| 194 | + | ||||
| 195 | + | // eventIndex reports the position of the first matching transaction event, or | |||
| 196 | + | // -1 when it never happened. | |||
| 197 | + | func (t *attachTestTerminal) eventIndex(kind string, status ports.UIPresentationStatus, state uint64) int { | |||
| 198 | + | t.mu.Lock() | |||
| 199 | + | defer t.mu.Unlock() | |||
| 200 | + | for i, event := range t.events { | |||
| 201 | + | if event.kind == kind && event.status == status && event.state == state { | |||
| 202 | + | return i | |||
| 203 | + | } | |||
| 204 | + | } | |||
| 205 | + | return -1 | |||
| 87 | } | 206 | } | ||
| 88 | 207 | ||||
| 89 | func (t *attachTestTerminal) written() string { | 208 | func (t *attachTestTerminal) written() string { | ||
| @@ -329,6 +448,14 @@ func awaitAttachedState(t *testing.T, sup *Supervisor) { | |||||
| 329 | require.Eventually(t, func() bool { return sup.State().Presentation == PresentAttached }, 5*time.Second, time.Millisecond) | 448 | require.Eventually(t, func() bool { return sup.State().Presentation == PresentAttached }, 5*time.Second, time.Millisecond) | ||
| 330 | } | 449 | } | ||
| 331 | 450 | ||||
| 451 | + | // authorityState reports the current foreground slot state without racing the | |||
| 452 | + | // supervisor's own transitions. | |||
| 453 | + | func (a *attachmentAuthority) authorityState() attachmentAuthorityState { | |||
| 454 | + | a.mu.Lock() | |||
| 455 | + | defer a.mu.Unlock() | |||
| 456 | + | return a.state | |||
| 457 | + | } | |||
| 458 | + | ||||
| 332 | // awaitPickerState waits until the supervisor is back in the picker. | 459 | // awaitPickerState waits until the supervisor is back in the picker. | ||
| 333 | func awaitPickerState(t *testing.T, sup *Supervisor) { | 460 | func awaitPickerState(t *testing.T, sup *Supervisor) { | ||
| 334 | t.Helper() | 461 | t.Helper() | ||
| @@ -376,6 +503,360 @@ func TestSupervisorReducerAttachmentPresentation(t *testing.T) { | |||||
| 376 | // TestSupervisorAttachmentLocalRemoteParity proves the same supervisor path | 503 | // TestSupervisorAttachmentLocalRemoteParity proves the same supervisor path | ||
| 377 | // opens the exact resolved request for a local and a remote selection, with no | 504 | // opens the exact resolved request for a local and a remote selection, with no | ||
| 378 | // dialer choice in the client. | 505 | // dialer choice in the client. | ||
| 506 | + | func TestSupervisorUIActionBindingUsesSingleAttachmentPump(t *testing.T) { | |||
| 507 | + | picker := newAttachTestPicker() | |||
| 508 | + | var stream *sessionTestStream | |||
| 509 | + | var streamMu sync.Mutex | |||
| 510 | + | var ui *UI | |||
| 511 | + | harness := startAttachHarnessConfig(t, picker, func(cfg *SupervisorConfig) { | |||
| 512 | + | ui = NewUI(cfg.Terminal.(ports.UIState), cfg.Clock) | |||
| 513 | + | cfg.UI = ui | |||
| 514 | + | }) | |||
| 515 | + | harness.service.setOpenStream(func(context.Context, ports.BrokerOpenStreamRequest) (ports.BrokerLogicalConnection, error) { | |||
| 516 | + | admitted := newSessionTestStream() | |||
| 517 | + | streamMu.Lock() | |||
| 518 | + | stream = admitted | |||
| 519 | + | streamMu.Unlock() | |||
| 520 | + | return admitted, nil | |||
| 521 | + | }) | |||
| 522 | + | ||||
| 523 | + | // Picker ownership has no UI binding, so this would fail if actions could | |||
| 524 | + | // collide with or bypass the picker's exclusive input claim. | |||
| 525 | + | _, err := ui.Action(t.Context(), ports.UIActionRequest{Attachment: ui.Handle(), Generation: 1, Text: "picker"}) | |||
| 526 | + | var pickerErr *ports.UIError | |||
| 527 | + | require.ErrorAs(t, err, &pickerErr) | |||
| 528 | + | require.Equal(t, ports.UIErrUnavailable, pickerErr.Code) | |||
| 529 | + | ||||
| 530 | + | picker.commit(sessionTestRequest(true)) | |||
| 531 | + | require.Eventually(t, func() bool { | |||
| 532 | + | streamMu.Lock() | |||
| 533 | + | defer streamMu.Unlock() | |||
| 534 | + | return stream != nil && len(stream.messages()) > 0 | |||
| 535 | + | }, 5*time.Second, time.Millisecond) | |||
| 536 | + | streamMu.Lock() | |||
| 537 | + | admitted := stream | |||
| 538 | + | streamMu.Unlock() | |||
| 539 | + | deliverReadyStream(t, admitted) | |||
| 540 | + | awaitAttachedState(t, harness.sup) | |||
| 541 | + | attachedOutput := sessionTestOutput(2, "\x1b[Hattached") | |||
| 542 | + | attachedOutput.Full = false | |||
| 543 | + | attachedOutput.Base = 1 | |||
| 544 | + | attachedOutput.New = 2 | |||
| 545 | + | admitted.deliver(attachedOutput) | |||
| 546 | + | ||||
| 547 | + | var snapshot ports.UISnapshot | |||
| 548 | + | require.Eventually(t, func() bool { | |||
| 549 | + | var captureErr error | |||
| 550 | + | snapshot, captureErr = ui.Capture(ui.Handle()) | |||
| 551 | + | return captureErr == nil && snapshot.Context.Status == ports.UIStatusAttached | |||
| 552 | + | }, 5*time.Second, time.Millisecond) | |||
| 553 | + | require.Equal(t, ports.UIStatusAttached, snapshot.Context.Status) | |||
| 554 | + | require.NotZero(t, snapshot.Context.Generation) | |||
| 555 | + | actionDone := make(chan error, 1) | |||
| 556 | + | go func() { | |||
| 557 | + | _, actionErr := ui.Action(t.Context(), ports.UIActionRequest{Attachment: ui.Handle(), Generation: snapshot.Context.Generation, Text: "automated"}) | |||
| 558 | + | actionDone <- actionErr | |||
| 559 | + | }() | |||
| 560 | + | var actionID uint64 | |||
| 561 | + | require.Eventually(t, func() bool { | |||
| 562 | + | var fenced bool | |||
| 563 | + | for _, message := range admitted.messages() { | |||
| 564 | + | switch typed := message.(type) { | |||
| 565 | + | case protocol.Input: | |||
| 566 | + | if string(typed.Data) == "automated" { | |||
| 567 | + | actionID = typed.ActionID | |||
| 568 | + | } | |||
| 569 | + | case protocol.UIFence: | |||
| 570 | + | fenced = actionID != 0 && typed.ActionID == actionID | |||
| 571 | + | } | |||
| 572 | + | } | |||
| 573 | + | return fenced | |||
| 574 | + | }, 5*time.Second, time.Millisecond, "admitted actions must be followed by their UI fence") | |||
| 575 | + | processedOutput := sessionTestOutput(3, "\x1b[Hprocessed") | |||
| 576 | + | processedOutput.Full = false | |||
| 577 | + | processedOutput.Base = 2 | |||
| 578 | + | processedOutput.New = 3 | |||
| 579 | + | processedOutput.Context.Publication = 3 | |||
| 580 | + | admitted.deliver(processedOutput) | |||
| 581 | + | var boundary ports.UIActionResult | |||
| 582 | + | require.Eventually(t, func() bool { | |||
| 583 | + | ui.mu.Lock() | |||
| 584 | + | boundary = ui.boundary | |||
| 585 | + | ui.mu.Unlock() | |||
| 586 | + | return boundary.Revision != 0 && boundary.Context.OutputState == processedOutput.New && boundary.Context.ViewPublication == processedOutput.Context.Publication | |||
| 587 | + | }, 5*time.Second, time.Millisecond, "the action-specific output must be the committed processing boundary") | |||
| 588 | + | admitted.deliver(protocol.UIReceipt{ActionID: actionID, Epoch: 1, State: 3, ViewPublication: 3, Outcome: protocol.UIReceiptProcessed}) | |||
| 589 | + | select { | |||
| 590 | + | case actionErr := <-actionDone: | |||
| 591 | + | require.NoError(t, actionErr, "receipt routing must complete the admitted action as processed") | |||
| 592 | + | case <-time.After(5 * time.Second): | |||
| 593 | + | t.Fatal("processed UI action did not complete") | |||
| 594 | + | } | |||
| 595 | + | ||||
| 596 | + | // Ending the stream releases before picker reacquisition. A stale request | |||
| 597 | + | // would be admitted here if finalization forgot to retire the binding. | |||
| 598 | + | admitted.deliver(protocol.Detached{}) | |||
| 599 | + | awaitPickerState(t, harness.sup) | |||
| 600 | + | ui.mu.Lock() | |||
| 601 | + | require.Nil(t, ui.input, "attachment finalization must revoke the UI pump binding") | |||
| 602 | + | require.Zero(t, ui.consumer, "attachment finalization must revoke the UI consumer") | |||
| 603 | + | require.Nil(t, ui.foreground, "attachment finalization must retire the UI foreground") | |||
| 604 | + | ui.mu.Unlock() | |||
| 605 | + | _, err = ui.Action(t.Context(), ports.UIActionRequest{Attachment: ui.Handle(), Generation: snapshot.Context.Generation, Text: "stale"}) | |||
| 606 | + | var staleErr *ports.UIError | |||
| 607 | + | require.ErrorAs(t, err, &staleErr) | |||
| 608 | + | require.Equal(t, ports.UIErrUnavailable, staleErr.Code) | |||
| 609 | + | ||||
| 610 | + | // Reconnecting installs a fresh UI-owned generation. The prior generation | |||
| 611 | + | // remains stale even while another attachment is actionable. | |||
| 612 | + | reconnected := newSessionTestStream() | |||
| 613 | + | harness.service.setOpenStream(func(context.Context, ports.BrokerOpenStreamRequest) (ports.BrokerLogicalConnection, error) { | |||
| 614 | + | return reconnected, nil | |||
| 615 | + | }) | |||
| 616 | + | picker.commit(sessionTestRequest(true)) | |||
| 617 | + | awaitHello(t, reconnected) | |||
| 618 | + | deliverReadyStream(t, reconnected) | |||
| 619 | + | awaitAttachedState(t, harness.sup) | |||
| 620 | + | reconnectedOutput := sessionTestOutput(2, "\x1b[Hreconnected") | |||
| 621 | + | reconnectedOutput.Full = false | |||
| 622 | + | reconnectedOutput.Base = 1 | |||
| 623 | + | reconnectedOutput.New = 2 | |||
| 624 | + | reconnected.deliver(reconnectedOutput) | |||
| 625 | + | require.Eventually(t, func() bool { | |||
| 626 | + | current, captureErr := ui.Capture(ui.Handle()) | |||
| 627 | + | return captureErr == nil && current.Context.Status == ports.UIStatusAttached && current.Context.Generation != snapshot.Context.Generation | |||
| 628 | + | }, 5*time.Second, time.Millisecond) | |||
| 629 | + | _, err = ui.Action(t.Context(), ports.UIActionRequest{Attachment: ui.Handle(), Generation: snapshot.Context.Generation, Text: "still-stale"}) | |||
| 630 | + | staleErr = nil | |||
| 631 | + | require.ErrorAs(t, err, &staleErr) | |||
| 632 | + | require.Equal(t, ports.UIErrStaleAttachment, staleErr.Code) | |||
| 633 | + | select { | |||
| 634 | + | case <-actionDone: // completion may be receipt-driven or retired on release. | |||
| 635 | + | default: | |||
| 636 | + | } | |||
| 637 | + | } | |||
| 638 | + | ||||
| 639 | + | // TestSupervisorUIActionReleaseWaitsForInFlightOutput pins GO-201: finalization | |||
| 640 | + | // must not release the UI foreground binding while a foreground output | |||
| 641 | + | // transaction is in flight. The release publishes Detached, so releasing it | |||
| 642 | + | // early would publish an unattached presentation while a newer session frame | |||
| 643 | + | // was still being written, and would return the picker claim before the last | |||
| 644 | + | // frame committed. The mutation this test pins moves releaseForeground before | |||
| 645 | + | // lease.stop in attachmentForeground.finalize. | |||
| 646 | + | func TestSupervisorUIActionReleaseWaitsForInFlightOutput(t *testing.T) { | |||
| 647 | + | const inFlightState = 2 | |||
| 648 | + | picker := newAttachTestPicker() | |||
| 649 | + | var ( | |||
| 650 | + | streamMu sync.Mutex | |||
| 651 | + | stream *sessionTestStream | |||
| 652 | + | ui *UI | |||
| 653 | + | ) | |||
| 654 | + | harness := startAttachHarnessConfig(t, picker, func(cfg *SupervisorConfig) { | |||
| 655 | + | ui = NewUI(cfg.Terminal.(ports.UIState), cfg.Clock) | |||
| 656 | + | cfg.UI = ui | |||
| 657 | + | }) | |||
| 658 | + | harness.service.setOpenStream(func(context.Context, ports.BrokerOpenStreamRequest) (ports.BrokerLogicalConnection, error) { | |||
| 659 | + | admitted := newSessionTestStream() | |||
| 660 | + | streamMu.Lock() | |||
| 661 | + | stream = admitted | |||
| 662 | + | streamMu.Unlock() | |||
| 663 | + | return admitted, nil | |||
| 664 | + | }) | |||
| 665 | + | ||||
| 666 | + | // Installed before the commit, so the supervisor's publication goroutine | |||
| 667 | + | // always observes them: only the attached frame with inFlightState blocks. | |||
| 668 | + | terminal := harness.terminal | |||
| 669 | + | entered := make(chan struct{}, 1) | |||
| 670 | + | release := make(chan struct{}) | |||
| 671 | + | var releaseOnce sync.Once | |||
| 672 | + | releaseOutput := func() { releaseOnce.Do(func() { close(release) }) } | |||
| 673 | + | defer releaseOutput() | |||
| 674 | + | detached := make(chan struct{}, 4) | |||
| 675 | + | terminal.publishMatch = func(ctx ports.UIContext) bool { | |||
| 676 | + | return ctx.Status == ports.UIStatusAttached && ctx.OutputState == inFlightState | |||
| 677 | + | } | |||
| 678 | + | terminal.publishEntered = entered | |||
| 679 | + | terminal.publishRelease = release | |||
| 680 | + | terminal.detachedSignals = detached | |||
| 681 | + | ||||
| 682 | + | picker.commit(sessionTestRequest(true)) | |||
| 683 | + | awaitStreamHello(t, &streamMu, &stream) | |||
| 684 | + | streamMu.Lock() | |||
| 685 | + | admitted := stream | |||
| 686 | + | streamMu.Unlock() | |||
| 687 | + | deliverReadyStream(t, admitted) | |||
| 688 | + | awaitAttachedState(t, harness.sup) | |||
| 689 | + | ||||
| 690 | + | // Block one foreground output transaction after BeginOutput and before its | |||
| 691 | + | // commit, so the send lease is held while finalization begins. | |||
| 692 | + | inFlight := sessionTestOutput(inFlightState, "\x1b[Hin-flight") | |||
| 693 | + | inFlight.Full = false | |||
| 694 | + | inFlight.Base = 1 | |||
| 695 | + | inFlight.New = inFlightState | |||
| 696 | + | admitted.deliver(inFlight) | |||
| 697 | + | select { | |||
| 698 | + | case <-entered: | |||
| 699 | + | case <-time.After(5 * time.Second): | |||
| 700 | + | t.Fatal("the foreground output transaction never reached its publication boundary") | |||
| 701 | + | } | |||
| 702 | + | require.Equal(t, -1, terminal.eventIndex("drain", ports.UIStatusAttached, inFlightState), "the in-flight frame must still hold its transaction") | |||
| 703 | + | ||||
| 704 | + | // Cancel the attachment while that transaction is in flight. | |||
| 705 | + | harness.service.lose(errors.New("attachment lost")) | |||
| 706 | + | require.Eventually(t, func() bool { | |||
| 707 | + | return harness.sup.attachments.authority.authorityState() == authorityFinalizing | |||
| 708 | + | }, 5*time.Second, time.Millisecond, "attachment finalization must begin while the frame is in flight") | |||
| 709 | + | ||||
| 710 | + | // Finalization cannot complete, and Detached cannot be published, until the | |||
| 711 | + | // in-flight transaction drains. | |||
| 712 | + | require.Never(t, func() bool { | |||
| 713 | + | select { | |||
| 714 | + | case <-detached: | |||
| 715 | + | return true | |||
| 716 | + | default: | |||
| 717 | + | return false | |||
| 718 | + | } | |||
| 719 | + | }, 50*time.Millisecond, time.Millisecond, "Detached must not publish while a foreground transaction is in flight") | |||
| 720 | + | require.Equal(t, authorityFinalizing, harness.sup.attachments.authority.authorityState(), "finalize still waits on the output transaction") | |||
| 721 | + | require.False(t, admitted.closedNow(), "the retired stream is released only after the drain") | |||
| 722 | + | require.False(t, picker.owns(), "the picker claim returns only after the drain") | |||
| 723 | + | require.Equal(t, PresentAttached, harness.sup.State().Presentation, "the attachment is still presented while its last frame drains") | |||
| 724 | + | ui.mu.Lock() | |||
| 725 | + | require.NotNil(t, ui.input, "the UI foreground binding is released only after the drain") | |||
| 726 | + | require.NotNil(t, ui.foreground, "the UI foreground context is retired only after the drain") | |||
| 727 | + | ui.mu.Unlock() | |||
| 728 | + | ||||
| 729 | + | releaseOutput() | |||
| 730 | + | select { | |||
| 731 | + | case <-detached: | |||
| 732 | + | case <-time.After(5 * time.Second): | |||
| 733 | + | t.Fatal("attachment finalization never published Detached") | |||
| 734 | + | } | |||
| 735 | + | ||||
| 736 | + | // The newer frame commits before Detached, and Detached carries the | |||
| 737 | + | // committed frame rather than the superseded one. | |||
| 738 | + | drained := terminal.eventIndex("drain", ports.UIStatusAttached, inFlightState) | |||
| 739 | + | published := terminal.eventIndex("publish", ports.UIStatusDetached, inFlightState) | |||
| 740 | + | require.GreaterOrEqual(t, drained, 0, "the in-flight frame must commit once it drains") | |||
| 741 | + | require.Greater(t, published, drained, "Detached must be published only after the newer frame commits") | |||
| 742 | + | publications := terminal.publicationList() | |||
| 743 | + | require.GreaterOrEqual(t, len(publications), 2) | |||
| 744 | + | last := publications[len(publications)-1] | |||
| 745 | + | require.Equal(t, ports.UIStatusDetached, last.Status) | |||
| 746 | + | require.Equal(t, uint64(inFlightState), last.OutputState, "Detached carries the committed newer frame") | |||
| 747 | + | prior := publications[len(publications)-2] | |||
| 748 | + | require.Equal(t, ports.UIStatusAttached, prior.Status) | |||
| 749 | + | require.Equal(t, uint64(inFlightState), prior.OutputState) | |||
| 750 | + | ||||
| 751 | + | // Finalization completes and hands the picker claim back. | |||
| 752 | + | require.Eventually(t, func() bool { | |||
| 753 | + | ui.mu.Lock() | |||
| 754 | + | released := ui.input == nil && ui.consumer == 0 && ui.foreground == nil | |||
| 755 | + | ui.mu.Unlock() | |||
| 756 | + | return released && picker.owns() && harness.sup.attachments.authority.authorityState() == authorityIdle && admitted.closedNow() | |||
| 757 | + | }, 5*time.Second, time.Millisecond, "finalization must complete and return the picker claim") | |||
| 758 | + | require.Equal(t, PresentPicker, harness.sup.State().Presentation) | |||
| 759 | + | } | |||
| 760 | + | ||||
| 761 | + | func TestSupervisorUIViewUpdateCommitsActionBoundaryWithoutBytes(t *testing.T) { | |||
| 762 | + | picker := newAttachTestPicker() | |||
| 763 | + | var ui *UI | |||
| 764 | + | harness := startAttachHarnessConfig(t, picker, func(cfg *SupervisorConfig) { | |||
| 765 | + | ui = NewUI(cfg.Terminal.(ports.UIState), cfg.Clock) | |||
| 766 | + | cfg.UI = ui | |||
| 767 | + | }) | |||
| 768 | + | stream := newSessionTestStream() | |||
| 769 | + | harness.service.setOpenStream(func(context.Context, ports.BrokerOpenStreamRequest) (ports.BrokerLogicalConnection, error) { | |||
| 770 | + | return stream, nil | |||
| 771 | + | }) | |||
| 772 | + | picker.commit(sessionTestRequest(true)) | |||
| 773 | + | awaitHello(t, stream) | |||
| 774 | + | deliverReadyStream(t, stream) | |||
| 775 | + | awaitAttachedState(t, harness.sup) | |||
| 776 | + | initialOutput := sessionTestOutput(2, "\x1b[Hmetadata-base") | |||
| 777 | + | initialOutput.Full = false | |||
| 778 | + | initialOutput.Base = 1 | |||
| 779 | + | initialOutput.New = 2 | |||
| 780 | + | initialOutput.Context.Publication = 2 | |||
| 781 | + | stream.deliver(initialOutput) | |||
| 782 | + | ||||
| 783 | + | var snapshot ports.UISnapshot | |||
| 784 | + | require.Eventually(t, func() bool { | |||
| 785 | + | var err error | |||
| 786 | + | snapshot, err = ui.Capture(ui.Handle()) | |||
| 787 | + | return err == nil && snapshot.Context.Status == ports.UIStatusAttached && snapshot.Context.OutputState == 2 | |||
| 788 | + | }, 5*time.Second, time.Millisecond) | |||
| 789 | + | actionDone := make(chan error, 1) | |||
| 790 | + | go func() { | |||
| 791 | + | _, err := ui.Action(t.Context(), ports.UIActionRequest{Attachment: ui.Handle(), Generation: snapshot.Context.Generation, Text: "metadata"}) | |||
| 792 | + | actionDone <- err | |||
| 793 | + | }() | |||
| 794 | + | var actionID uint64 | |||
| 795 | + | require.Eventually(t, func() bool { | |||
| 796 | + | for _, message := range stream.messages() { | |||
| 797 | + | if fence, ok := message.(protocol.UIFence); ok { | |||
| 798 | + | actionID = fence.ActionID | |||
| 799 | + | return actionID != 0 | |||
| 800 | + | } | |||
| 801 | + | } | |||
| 802 | + | return false | |||
| 803 | + | }, 5*time.Second, time.Millisecond) | |||
| 804 | + | view := protocol.ViewContext{Publication: 3, Route: snapshot.Context.Route, TabID: "metadata-tab", FocusedPaneID: "metadata-pane"} | |||
| 805 | + | view.Route.Target = protocol.ExactSessionTarget{LifecycleID: domain.SessionLifecycleID{1}, SessionName: "alpha"} | |||
| 806 | + | stream.deliver(protocol.UIViewUpdate{Epoch: 1, State: 2, Context: view}) | |||
| 807 | + | require.Eventually(t, func() bool { | |||
| 808 | + | current, err := ui.Capture(ui.Handle()) | |||
| 809 | + | return err == nil && current.Context.ViewPublication == 3 && current.Context.TabID == "metadata-tab" | |||
| 810 | + | }, 5*time.Second, time.Millisecond, "metadata-only update must commit through the foreground transaction") | |||
| 811 | + | stream.deliver(protocol.UIReceipt{ActionID: actionID, Epoch: 1, State: 2, ViewPublication: 3, Outcome: protocol.UIReceiptProcessed}) | |||
| 812 | + | select { | |||
| 813 | + | case err := <-actionDone: | |||
| 814 | + | require.NoError(t, err) | |||
| 815 | + | case <-time.After(5 * time.Second): | |||
| 816 | + | t.Fatal("metadata-only committed boundary did not complete action") | |||
| 817 | + | } | |||
| 818 | + | } | |||
| 819 | + | ||||
| 820 | + | func TestSupervisorUIViewUpdateRequestsOneCoalescedReset(t *testing.T) { | |||
| 821 | + | picker := newAttachTestPicker() | |||
| 822 | + | harness := startAttachHarness(t, picker) | |||
| 823 | + | stream := newSessionTestStream() | |||
| 824 | + | harness.service.setOpenStream(func(context.Context, ports.BrokerOpenStreamRequest) (ports.BrokerLogicalConnection, error) { | |||
| 825 | + | return stream, nil | |||
| 826 | + | }) | |||
| 827 | + | picker.commit(sessionTestRequest(true)) | |||
| 828 | + | awaitHello(t, stream) | |||
| 829 | + | deliverReadyStream(t, stream) | |||
| 830 | + | awaitAttachedState(t, harness.sup) | |||
| 831 | + | bad := protocol.UIViewUpdate{Epoch: 9, State: 9, Context: protocol.ViewContext{Publication: 2, Route: protocol.CommittedRouteIdentity{Target: protocol.ExactSessionTarget{LifecycleID: domain.SessionLifecycleID{1}, SessionName: "alpha"}}, TabID: "future", FocusedPaneID: "future"}} | |||
| 832 | + | stream.deliver(bad) | |||
| 833 | + | stream.deliver(bad) | |||
| 834 | + | require.Eventually(t, func() bool { | |||
| 835 | + | count := 0 | |||
| 836 | + | for _, message := range stream.messages() { | |||
| 837 | + | if _, ok := message.(protocol.OutputResetRequest); ok { | |||
| 838 | + | count++ | |||
| 839 | + | } | |||
| 840 | + | } | |||
| 841 | + | return count == 1 | |||
| 842 | + | }, 5*time.Second, time.Millisecond, "future view dependencies must request one coalesced reset") | |||
| 843 | + | } | |||
| 844 | + | ||||
| 845 | + | func TestSupervisorWithoutUIDoesNotBindActions(t *testing.T) { | |||
| 846 | + | picker := newAttachTestPicker() | |||
| 847 | + | harness := startAttachHarness(t, picker) | |||
| 848 | + | require.Nil(t, harness.sup.cfg.UI, "the optional seam must preserve publication-only composition") | |||
| 849 | + | require.Nil(t, harness.sup.attachments.actionUI, "mutation that binds an implicit UI changes no-UI behavior") | |||
| 850 | + | } | |||
| 851 | + | ||||
| 852 | + | func TestUIActionWithoutInputReturnsUnavailable(t *testing.T) { | |||
| 853 | + | ui := NewUI(&attachTestTerminal{}, newSupervisorTestClock()) | |||
| 854 | + | _, err := ui.Action(t.Context(), ports.UIActionRequest{Attachment: ui.Handle(), Generation: 1, Text: "x"}) | |||
| 855 | + | var uiErr *ports.UIError | |||
| 856 | + | require.ErrorAs(t, err, &uiErr) | |||
| 857 | + | require.Equal(t, ports.UIErrUnavailable, uiErr.Code) | |||
| 858 | + | } | |||
| 859 | + | ||||
| 379 | func TestSupervisorAttachmentLocalRemoteParity(t *testing.T) { | 860 | func TestSupervisorAttachmentLocalRemoteParity(t *testing.T) { | ||
| 380 | for _, local := range []bool{true, false} { | 861 | for _, local := range []bool{true, false} { | ||
| 381 | name := "remote" | 862 | name := "remote" | ||
M internal/usecase/client/ui_actions.go +33 -0
| @@ -27,6 +27,29 @@ func (u *UI) bindForeground(ctx context.Context, input *terminalInputPump, consu | |||
| 27 | 27 | return u.generation | |
| 28 | 28 | } | |
| 29 | 29 | ||
| 30 | + | // releaseForeground retires exactly the binding installed for generation. It | |
| 31 | + | // is called when the supervisor begins finalizing the attachment, before the | |
| 32 | + | // pump claim can return to the picker. Detached is the honest publication while | |
| 33 | + | // no session owns input; it also gives observers a committed status that cannot | |
| 34 | + | // be mistaken for an actionable attachment. | |
| 35 | + | func (u *UI) releaseForeground(generation uint64) { | |
| 36 | + | u.mu.Lock() | |
| 37 | + | if generation != u.generation { | |
| 38 | + | u.mu.Unlock() | |
| 39 | + | return | |
| 40 | + | } | |
| 41 | + | u.input = nil | |
| 42 | + | u.consumer = 0 | |
| 43 | + | u.foreground = nil | |
| 44 | + | u.boundary = ports.UIActionResult{} | |
| 45 | + | if u.pending != 0 { | |
| 46 | + | u.finishLocked(u.pending, ports.UIActionOutcomeUnknown, ports.UIActionResult{}) | |
| 47 | + | } | |
| 48 | + | u.signalLocked() | |
| 49 | + | u.mu.Unlock() | |
| 50 | + | u.status(ports.UIStatusDetached) | |
| 51 | + | } | |
| 52 | + | ||
| 30 | 53 | func (u *UI) accept(id, generation uint64) bool { | |
| 31 | 54 | u.mu.Lock() | |
| 32 | 55 | defer u.mu.Unlock() | |
| @@ -120,6 +143,12 @@ func (u *UI) Action(ctx context.Context, request ports.UIActionRequest) (ports.U | |||
| 120 | 143 | if request.Attachment != u.handle { | |
| 121 | 144 | return ports.UIActionResult{}, &ports.UIError{Code: ports.UIErrStaleAttachment} | |
| 122 | 145 | } | |
| 146 | + | u.mu.Lock() | |
| 147 | + | available := u.input != nil | |
| 148 | + | u.mu.Unlock() | |
| 149 | + | if !available { | |
| 150 | + | return ports.UIActionResult{}, &ports.UIError{Code: ports.UIErrUnavailable} | |
| 151 | + | } | |
| 123 | 152 | snapshot, err := u.state.Snapshot() | |
| 124 | 153 | if err != nil { | |
| 125 | 154 | return ports.UIActionResult{}, err | |
| @@ -140,6 +169,10 @@ func (u *UI) Action(ctx context.Context, request ports.UIActionRequest) (ports.U | |||
| 140 | 169 | return ports.UIActionResult{}, &ports.UIError{Code: ports.UIErrInvalidRequest} | |
| 141 | 170 | } | |
| 142 | 171 | u.mu.Lock() | |
| 172 | + | if u.input == nil { | |
| 173 | + | u.mu.Unlock() | |
| 174 | + | return ports.UIActionResult{}, &ports.UIError{Code: ports.UIErrUnavailable} | |
| 175 | + | } | |
| 143 | 176 | if request.Generation != u.generation { | |
| 144 | 177 | u.mu.Unlock() | |
| 145 | 178 | return ports.UIActionResult{}, &ports.UIError{Code: ports.UIErrStaleAttachment} | |
| @@ -27,6 +27,29 @@ func (u *UI) bindForeground(ctx context.Context, input *terminalInputPump, consu | |||||
| 27 | return u.generation | 27 | return u.generation | ||
| 28 | } | 28 | } | ||
| 29 | 29 | ||||
| 30 | + | // releaseForeground retires exactly the binding installed for generation. It | |||
| 31 | + | // is called when the supervisor begins finalizing the attachment, before the | |||
| 32 | + | // pump claim can return to the picker. Detached is the honest publication while | |||
| 33 | + | // no session owns input; it also gives observers a committed status that cannot | |||
| 34 | + | // be mistaken for an actionable attachment. | |||
| 35 | + | func (u *UI) releaseForeground(generation uint64) { | |||
| 36 | + | u.mu.Lock() | |||
| 37 | + | if generation != u.generation { | |||
| 38 | + | u.mu.Unlock() | |||
| 39 | + | return | |||
| 40 | + | } | |||
| 41 | + | u.input = nil | |||
| 42 | + | u.consumer = 0 | |||
| 43 | + | u.foreground = nil | |||
| 44 | + | u.boundary = ports.UIActionResult{} | |||
| 45 | + | if u.pending != 0 { | |||
| 46 | + | u.finishLocked(u.pending, ports.UIActionOutcomeUnknown, ports.UIActionResult{}) | |||
| 47 | + | } | |||
| 48 | + | u.signalLocked() | |||
| 49 | + | u.mu.Unlock() | |||
| 50 | + | u.status(ports.UIStatusDetached) | |||
| 51 | + | } | |||
| 52 | + | ||||
| 30 | func (u *UI) accept(id, generation uint64) bool { | 53 | func (u *UI) accept(id, generation uint64) bool { | ||
| 31 | u.mu.Lock() | 54 | u.mu.Lock() | ||
| 32 | defer u.mu.Unlock() | 55 | defer u.mu.Unlock() | ||
| @@ -120,6 +143,12 @@ func (u *UI) Action(ctx context.Context, request ports.UIActionRequest) (ports.U | |||||
| 120 | if request.Attachment != u.handle { | 143 | if request.Attachment != u.handle { | ||
| 121 | return ports.UIActionResult{}, &ports.UIError{Code: ports.UIErrStaleAttachment} | 144 | return ports.UIActionResult{}, &ports.UIError{Code: ports.UIErrStaleAttachment} | ||
| 122 | } | 145 | } | ||
| 146 | + | u.mu.Lock() | |||
| 147 | + | available := u.input != nil | |||
| 148 | + | u.mu.Unlock() | |||
| 149 | + | if !available { | |||
| 150 | + | return ports.UIActionResult{}, &ports.UIError{Code: ports.UIErrUnavailable} | |||
| 151 | + | } | |||
| 123 | snapshot, err := u.state.Snapshot() | 152 | snapshot, err := u.state.Snapshot() | ||
| 124 | if err != nil { | 153 | if err != nil { | ||
| 125 | return ports.UIActionResult{}, err | 154 | return ports.UIActionResult{}, err | ||
| @@ -140,6 +169,10 @@ func (u *UI) Action(ctx context.Context, request ports.UIActionRequest) (ports.U | |||||
| 140 | return ports.UIActionResult{}, &ports.UIError{Code: ports.UIErrInvalidRequest} | 169 | return ports.UIActionResult{}, &ports.UIError{Code: ports.UIErrInvalidRequest} | ||
| 141 | } | 170 | } | ||
| 142 | u.mu.Lock() | 171 | u.mu.Lock() | ||
| 172 | + | if u.input == nil { | |||
| 173 | + | u.mu.Unlock() | |||
| 174 | + | return ports.UIActionResult{}, &ports.UIError{Code: ports.UIErrUnavailable} | |||
| 175 | + | } | |||
| 143 | if request.Generation != u.generation { | 176 | if request.Generation != u.generation { | ||
| 144 | u.mu.Unlock() | 177 | u.mu.Unlock() | ||
| 145 | return ports.UIActionResult{}, &ports.UIError{Code: ports.UIErrStaleAttachment} | 178 | return ports.UIActionResult{}, &ports.UIError{Code: ports.UIErrStaleAttachment} | ||