Distinguish input reply backpressure from receiver lifecycle

Changed

 README.md                      |  3 ++-
 docs/api.md                    |  6 ++++++
 docs/design.md                 |  9 +++++++--
 scripts/test-input-reply-queue | 10 ++++++++++
 src/c/api.h                    |  3 +++
 src/c/host.c                   | 33 +++++++++++++++++++++++++++++++--
 src/sigil/wm/bridge.sgl        | 10 +++++++++-
 src/sigil/wm/input.sgl         |  7 ++++---
 src/sigil/wm/runtime.sgl       |  6 +++++-
 test/input-queue-gate.c        | 34 ++++++++++++++++++++++++++++++++++
 test/input-reply-queue.sgl     | 28 ++++++++++++++++++++++++----
 11 files changed, 135 insertions(+), 14 deletions(-)

Diff

README.mdmodified

@@ -88,7 +88,8 @@ composed-output captures and PNGs under `build/receipts/`.

or motion hook is returning its decision. A test-only native gate pauses

consumption, and another scheduler task releases it to verify cooperative

backpressure, accepted replies and subsequent input progress. It also checks

−91

replies after shutdown with an empty ring. Set `SIGIL` to a runtime with FFI

+91

replies before native readiness and after shutdown with an empty ring, and

+92

checks diagnostic rate limiting. Set `SIGIL` to a runtime with FFI

loading support; logs are retained under `build/receipts/input-reply-queue/`.

`scripts/test-lock-presentation` uses a test-only presentation barrier to check

docs/api.mdmodified

@@ -203,6 +203,12 @@ target or output changes. Its `moved?:` field distinguishes input motion from

203

scene refresh; coordinates use `x-milli:` and `y-milli:` fields. Observation

204

chooses no focus and can run independently of interception. Disabled observation

205

and absent motion hooks generate no ordinary motion callback to Sigil.

+206

Input replies retry cooperatively when the command ring is full or native

+207

startup is still in progress. The hook runs once; its copied disposition stays

+208

in FIFO order with earlier commands. A terminated receiver ends retry without

+209

raising a queue-full error. Rate-limited `INPUT_REPLY` lines distinguish

+210

`queue-full`, `receiver-starting` and `receiver-stopped` with depth and input kind.

+211

212

Intercepted input decisions have a 32 ms native observation budget and 10 ms

213

timer granularity. An intercepted key/button over that budget is logged and

214

keeps waiting for policy; it is never forwarded by default. Missing motion

docs/design.mdmodified

@@ -104,8 +104,13 @@ and focus commands. When it fills, Sigil retains the computed reply and sleeps

104

cooperatively for 1 ms between enqueue attempts, without repeating the hook.

105

Keeping the reply in this FIFO preserves earlier hook commands, including focus

106

authorization. A reserved queue or priority lane could overtake those commands.

−107

Retry ends when the reply is accepted or the native receiver has shut down;

−108

shutdown cannot leave an active intercepted input waiting for that reply.

+107

Replies also wait while the native receiver is starting. Retry ends when the

+108

reply is accepted or the receiver has shut down; shutdown cannot leave an active

+109

intercepted input waiting for that reply. `wm_send` retains its boolean contract;

+110

`wm_input_reply` distinguishes accepted, full, starting and stopped results.

+111

`INPUT_REPLY` diagnostics include the case, wait/drop action, queue depth, input

+112

kind and native state. Repeated failures are limited to one line per second for

+113

each case and kind, with the suppressed attempt count included on the next line.

114

115

Intercepted decisions are observed after 32 ms, with at most one 10 ms timer

116

interval of granularity. An unanswered intercepted press is logged and keeps

scripts/test-input-reply-queuemodified

@@ -15,6 +15,15 @@ for kind in 30 31 32; do

cat "$log"

grep -q "^INPUT_REPLY_QUEUE=PASS kind=$kind " "$log"

! grep -q 'event handler failed' "$log"

+18

[[ $(grep -c '^INPUT_REPLY case=queue-full action=wait depth=1024 ' "$log") == 1 ]]

+19

done

+20

for kind in 30 31 32; do

+21

log=build/receipts/input-reply-queue/starting-$kind.log

+22

  WM_INPUT_MODE=starting WM_INPUT_KIND=$kind timeout --foreground 15 "${SIGIL:-sigil}" -L src eval -f test/input-reply-queue.sgl > "$log" 2>&1 || { cat "$log"; exit 1; }

+23

cat "$log"

+24

grep -q "^INPUT_STARTING=PASS kind=$kind " "$log"

+25

! grep -q 'event handler failed' "$log"

+26

[[ $(grep -c '^INPUT_REPLY case=receiver-starting action=wait depth=0 ' "$log") == 1 ]]

done

for kind in 30 31 32; do

log=build/receipts/input-reply-queue/shutdown-$kind.log

@@ -22,4 +31,5 @@ for kind in 30 31 32; do

cat "$log"

grep -q "^INPUT_SHUTDOWN=PASS kind=$kind" "$log"

! grep -q 'event handler failed' "$log"

+34

[[ $(grep -c '^INPUT_REPLY case=receiver-stopped action=drop depth=0 ' "$log") == 1 ]]

done

src/c/api.hmodified

@@ -31,6 +31,9 @@ int wm_event_fd(void);

int wm_next(struct wm_message *event);

int wm_event_text(void *bytes, int capacity);

int wm_send(const struct wm_message *command);

+34

enum wm_send_result { WM_SEND_INVALID=-2, WM_SEND_STOPPED=-1, WM_SEND_FULL=0, WM_SEND_ACCEPTED=1, WM_SEND_STARTING=2 };

+35

/* Reply status preserves startup waits without changing wm_send's boolean contract. */

+36

int wm_input_reply(const struct wm_message *command,int input_kind);

int wm_submit(int64_t kind, int64_t token, const int64_t *words, int64_t count);

/* Pixel metadata: x,y,logical width/height,pixel width/height,stride,stack.

* stack is -1 bottom, 0 preserve, 1 top, 2 ordered insertion (options required).

src/c/host.cmodified

@@ -2470,14 +2470,43 @@ int wm_next(struct wm_message *event) {

2470

}

2471

return 0;

2472

}

+2473

static int send_owned_result(struct wm_message m,unsigned *depth,int *state) {

+2474

*state=wm_state();

+2475

if(*state!=2 || bridge.joined) {

+2476

*depth=atomic_load(&bridge.commands.write)-atomic_load(&bridge.commands.read);

+2477

return *state==1 && !bridge.joined ? WM_SEND_STARTING : WM_SEND_STOPPED;

+2478

}

+2479

if(!push(&bridge.commands,m)) { *depth=QUEUE_SIZE; return WM_SEND_FULL; }

+2480

wake(bridge.command_fd); return WM_SEND_ACCEPTED;

+2481

}

2482

static int send_owned(struct wm_message m) {

−2474

if(wm_state()!=2 || bridge.joined || !push(&bridge.commands,m)) return 0;

−2475

wake(bridge.command_fd); return 1;

+2483

unsigned depth;

+2484

int state;

+2485

return send_owned_result(m,&depth,&state)==WM_SEND_ACCEPTED;

2486

}

2487

int wm_send(const struct wm_message *m) {

2488

if(!m || m->word[0]>=90 || m->word[0]<=0) return 0;

2489

return send_owned(*m);

2490

}

+2491

int wm_input_reply(const struct wm_message *m,int kind) {

+2492

/* Repeated retries must not turn backpressure into logging pressure. */

+2493

static struct { uint64_t next,suppressed; } reports[3][3];

+2494

if(!m || m->word[0]!=11 || kind<30 || kind>32) return WM_SEND_INVALID;

+2495

unsigned depth;

+2496

int state,result=send_owned_result(*m,&depth,&state);

+2497

if(result==WM_SEND_ACCEPTED) return result;

+2498

unsigned reason=result==WM_SEND_FULL ? 0 : result==WM_SEND_STARTING ? 1 : 2;

+2499

uint64_t now=now_ms();

+2500

if(now>=reports[kind-30][reason].next) {

+2501

fprintf(stderr,"INPUT_REPLY case=%s action=%s depth=%u kind=%d state=%d suppressed=%llu\n",

+2502

result==WM_SEND_FULL ? "queue-full" : result==WM_SEND_STARTING ? "receiver-starting" : "receiver-stopped",

+2503

result==WM_SEND_STOPPED ? "drop" : "wait",depth,kind,state,

+2504

(unsigned long long)reports[kind-30][reason].suppressed);

+2505

reports[kind-30][reason].next=now+1000;

+2506

reports[kind-30][reason].suppressed=0;

+2507

} else reports[kind-30][reason].suppressed++;

+2508

return result;

+2509

}

2510

int wm_submit_traced(int64_t kind,int64_t token,const int64_t *words,int64_t count,const uint64_t *causes,int trace_count) {

2511

if(!words || count<1 || count>65536 || kind<1 || kind>15) return 0;

2512

if(trace_count<0 || trace_count>TRACE_CAUSES || (trace_count && !causes)) return 0;

src/sigil/wm/bridge.sglmodified

@@ -3,7 +3,7 @@

(import (sigil ffi) (sigil process) (sigil async) (sigil io) (sigil wm trace))

  (export native-start! native-options! native-socket-name! native-decoration-default! native-state native-socket native-statistic native-join!

current-trace-correlation native-trace-enable! native-trace-dropped

−6

          native-latency-mark! native-command! native-submit! native-pixels! native-event-text next-native-event wait-native-event!)

          native-latency-mark! native-command! native-input-reply! native-submit! native-pixels! native-event-text next-native-event wait-native-event!)

(begin

(define native (c-library (or (getenv "SIGIL_WM_LIBRARY") "./build/libsigil-wm.so")))

(define native-start! (c-function native "wm_start" '() ffi/int))

@@ -20,6 +20,7 @@

(define native-join! (c-function native "wm_join" '() ffi/int))

(define event-fd (c-function native "wm_event_fd" '() ffi/int))

(define send (c-function native "wm_send" (list ffi/pointer) ffi/int))

+23

(define input-reply (c-function native "wm_input_reply" (list ffi/pointer ffi/int) ffi/int))

(define next (c-function native "wm_next" (list ffi/pointer) ffi/int))

(define event-text (c-function native "wm_event_text" (list ffi/pointer ffi/int) ffi/int))

@@ -68,6 +69,13 @@

(pointer-set! storage ffi/uint64 120 (current-trace-correlation)))

(= (send storage) 1))))

+72

(define (native-input-reply! words kind)

+73

(unless (<= (length words) 16) (error "command too large"))

+74

(with-words words 128 (lambda (storage)

+75

(when (not (= (current-trace-correlation) 0))

+76

(pointer-set! storage ffi/uint64 120 (current-trace-correlation)))

+77

(input-reply storage kind))))

+78

(define (native-submit! kind token words)

(let ((causes (trace-causes)))

(with-words words (* 8 (max 1 (length words)))

src/sigil/wm/input.sglmodified

@@ -109,9 +109,10 @@

109

(input-event-modifiers replacement))))

110

;; Preserve command order while yielding until the native queue has space.

111

(let retry ()

−112

(unless (or (host-send! host reply) (not (host-running? host)))

−113

(sleep .001)

−114

(retry))))))

+112

(case (host-input-reply! host reply (input-event-kind event))

+113

((0 2) (sleep .001) (retry))

+114

((1 -1) #t)

+115

(else (error "invalid input reply")))))))

116

117

(define (install-input-dispatch! host)

118

(or (wm-host-input-handler host)

src/sigil/wm/runtime.sglmodified

@@ -3,7 +3,7 @@

(import (sigil math) (sigil struct) (sigil io) (sigil wm bridge) (sigil wm model) (sigil wm trace))

(export start-host! stop-host! run-host! poll-host! dispatch-host-event!

on-event! remove-handler! host-snapshot host-socket host-statistic host-running?

−6

host-send! host-submit! host-next-token!

host-send! host-input-reply! host-submit! host-next-token!

wm-host? wm-host-key-hook set-wm-host-key-hook!

wm-host-button-hook set-wm-host-button-hook!

wm-host-motion-hook set-wm-host-motion-hook!

@@ -40,6 +40,10 @@

(unless (wm-host? host) (error "expected a host"))

(native-command! words))

+43

(define (host-input-reply! host words kind)

+44

(unless (wm-host? host) (error "expected a host"))

+45

(native-input-reply! words kind))

+46

(define (host-submit! host kind token words)

(unless (wm-host? host) (error "expected a host"))

(native-submit! kind token words))

test/input-queue-gate.cmodified

@@ -1,13 +1,20 @@

/* SPDX-License-Identifier: BSD-3-Clause */

#include <unistd.h>

#include <wayland-server-core.h>

static ssize_t gated_read(int fd,void *buffer,size_t count);

static const char *gated_socket(struct wl_display *display);

#define read(...) gated_read(__VA_ARGS__)

#define wm_send unrecorded_send

#define wm_input_reply unrecorded_input_reply

#define wl_display_add_socket_auto gated_socket

#include "../src/c/host.c"

+11

#undef wl_display_add_socket_auto

+12

#undef wm_input_reply

#undef wm_send

#undef read

static atomic_bool hold_commands,commands_held;

+17

static atomic_int starting_kind;

static unsigned reply_attempts,reply_refused,reply_accepted;

int wm_send(const struct wm_message *message) {

@@ -20,6 +27,33 @@ int wm_send(const struct wm_message *message) {

return result;

}

+30

int wm_input_reply(const struct wm_message *message,int kind) {

+31

int result=unrecorded_input_reply(message,kind);

+32

reply_attempts++;

+33

if(result==WM_SEND_ACCEPTED) reply_accepted++;

+34

else reply_refused++;

+35

return result;

+36

}

+37

+38

int wm_test_start_gate(int kind) {

+39

atomic_store(&starting_kind,kind);

+40

return 1;

+41

}

+42

+43

/* A copied startup observation exercises retries before the ready event. */

+44

static const char *gated_socket(struct wl_display *display) {

+45

int kind=atomic_load(&starting_kind);

+46

if(kind) {

+47

emit((struct wm_message){.word={kind,777,1,now_ms(),kind==30 ? 30 : kind==31 ? 272 : 0,1,0,0,1,1}});

+48

uint64_t deadline=now_ms()+3000;

+49

while(atomic_load(&starting_kind) && now_ms()<deadline) {

+50

struct timespec delay={.tv_nsec=500000};

+51

nanosleep(&delay,NULL);

+52

}

+53

}

+54

return wl_display_add_socket_auto(display);

+55

}

+56

int wm_test_reply_stat(int index) {

switch(index) {

case 0: return reply_attempts;

test/input-reply-queue.sglmodified

@@ -1,6 +1,6 @@

(import (sigil async) (sigil io) (sigil ffi) (sigil process) (sigil time)

(sigil wm host) (sigil wm input)

−3

(only (sigil wm runtime) host-send! wm-host-stopped?))

(only (sigil wm runtime) host-send! wm-host-stopped? set-wm-host-motion-hook!))

(define (check value message) (unless value (error message)))

(define (now-ms) (inexact->exact (round (* 1000 (/ (current-jiffy) (jiffies-per-second))))))

(define (await predicate)

@@ -13,8 +13,10 @@

(define queued (c-function fixture "wm_test_queued" (list ffi/int) ffi/int))

(define depth (c-function fixture "wm_test_depth" '() ffi/int))

(define reply-stat (c-function fixture "wm_test_reply_stat" (list ffi/int) ffi/int))

+16

(define start-gate (c-function fixture "wm_test_start_gate" (list ffi/int) ffi/int))

(define mode (or (getenv "WM_INPUT_MODE") "overflow"))

(define kind (string->number (or (getenv "WM_INPUT_KIND") "30")))

+19

(when (equal? mode "starting") (start-gate kind))

(define host (start-host!))

(define ready? #f)

(define completed '())

@@ -31,12 +33,17 @@

(set! observed (cons event observed))

(unless saved

(set! saved event)

−34

(if (equal? mode "shutdown")

+36

(cond

+37

((equal? mode "starting")

+38

(set! start (now-ms))

+39

(go (sleep .075) (start-gate 0) (set! released? #t)))

+40

((equal? mode "shutdown")

(begin

(check (stop-host! host) "stop command refused")

(await (lambda () (not (host-running? host))))

(check (= (depth) 0) "shutdown command queue not empty")

−39

(display "SHUTDOWN_REPLY queued=0 reply-kind=") (display kind) (newline))

+45

(display "SHUTDOWN_REPLY queued=0 reply-kind=") (display kind) (newline)))

+46

(else

(begin

(check (= (gate 1) 1) "native command gate failed")

(let loop () (when (set-seat-capabilities! host 3) (loop)))

@@ -46,8 +53,12 @@

(display (input-event-kind event)) (newline)

(set! start (now-ms))

(set! filled? #t)

−49

(go (sleep .075) (gate 0) (set! released? #t)))))

+56

(go (sleep .075) (gate 0) (set! released? #t))))))

(drop))

+58

(set-key-hook! host hook)

+59

(set-button-hook! host hook)

+60

(when (= kind 32) (set-motion-hook! host hook 0 0 '(0 0 800 600)))

+61

(when (and (= kind 32) (equal? mode "starting")) (set-wm-host-motion-hook! host hook))

(with-async

(go (run-host! host))

(await (lambda () ready?))

@@ -63,6 +74,15 @@

(check first "input emulation refused")

(await (lambda () saved))

(cond

+77

((equal? mode "starting")

+78

(await (lambda () (memv first completed)))

+79

(check released? "startup reply did not yield to readiness")

+80

(check (= (length observed) 2) "startup hook repeated or input lost")

+81

(check (> (reply-stat 1) 0) "starting receiver never refused reply")

+82

(check (< (reply-stat 0) 200) "startup reply spun without sleeping")

+83

(check (= (reply-stat 2) 2) "startup reply dropped before readiness")

+84

(display "INPUT_STARTING=PASS kind=") (display kind)

+85

(display " elapsed-ms=") (display (- (now-ms) start)) (newline))

((equal? mode "shutdown")

(await (lambda () (wm-host-stopped? host)))

(check (= (reply-stat 0) 1) "shutdown reply retried without a receiver")