Add independent Remote input helper with EOF and watchdog cleanup

internal/platform/xorg/remote_helper_native_test.go

@@ -0,0 +1,316 @@
+// Copyright (c) 2026 Storm Dragon
+// Authors: Storm Dragon, Angus Kola
+// SPDX-License-Identifier: GPL-3.0-or-later
+
+package xorg
+
+import (
+	"bytes"
+	"io"
+	"os"
+	"os/exec"
+	"syscall"
+	"testing"
+	"time"
+
+	"git.stormux.org/storm/cthulhu/internal/remoteinput"
+)
+
+// The writer owns both write ends exclusively. The supervising test owns the
+// helper process so it can observe cleanup and reap it after writer SIGKILL.
+func TestNativeRemoteKeyControlWriter(t *testing.T) {
+	if os.Args[len(os.Args)-1] != "remote-key-control-writer" {
+		return
+	}
+	if os.Getenv("CTHULHU_NATIVE_REMOTE_KEYS") != "1" {
+		os.Exit(1)
+	}
+	control := os.NewFile(3, "private-control-writer")
+	lifetime := os.NewFile(4, "private-lifetime-writer")
+	_, err := io.Copy(control, os.Stdin)
+	control.Close()
+	lifetime.Close()
+	if err != nil {
+		os.Exit(1)
+	}
+	os.Exit(0)
+}
+
+type nativeHelper struct {
+	writer, helper         *exec.Cmd
+	input                  io.WriteCloser
+	output                 io.ReadCloser
+	writerDone, helperDone chan struct{}
+	writerErr, helperErr   error
+	stopped                bool
+	stderr                 bytes.Buffer
+	probe                  *remoteKeyTarget
+	shift, control         uint16
+}
+
+func startNativeHelper(t *testing.T) *nativeHelper {
+	t.Helper()
+	path := os.Getenv("CTHULHU_REMOTE_INPUT_HELPER")
+	if path == "" {
+		t.Fatal("requires helper built by private native_remote_keys.py")
+	}
+	probe := remoteNativeTarget(t)
+	shift, err := probe.backend.Resolve("Shift_L")
+	if err != nil {
+		t.Fatal(err)
+	}
+	controlCode, err := probe.backend.Resolve("Control_R")
+	if err != nil {
+		t.Fatal(err)
+	}
+	keyRead, keyWrite, err := os.Pipe()
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer keyRead.Close()
+	defer keyWrite.Close()
+	lifeRead, lifeWrite, err := os.Pipe()
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer lifeRead.Close()
+	defer lifeWrite.Close()
+	f := &nativeHelper{probe: probe, shift: shift, control: controlCode,
+		writerDone: make(chan struct{}), helperDone: make(chan struct{})}
+	f.helper = exec.Command(path, "--epoch", "7")
+	f.helper.Stdin = keyRead
+	f.helper.ExtraFiles = []*os.File{lifeRead}
+	f.helper.SysProcAttr = &syscall.SysProcAttr{Setpgid: true}
+	f.helper.Stderr = &f.stderr
+	ackRead, ackWrite, err := os.Pipe()
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer ackWrite.Close()
+	t.Cleanup(func() { ackRead.Close() })
+	// Wait must not close the read end before the final cleanup ACK is read.
+	// A caller-owned pipe retains buffered records after helper process exit.
+	f.output, f.helper.Stdout = ackRead, ackWrite
+	if err := f.helper.Start(); err != nil {
+		t.Fatal(err)
+	}
+	go func() { f.helperErr = f.helper.Wait(); close(f.helperDone) }()
+	t.Cleanup(func() {
+		if f.writer != nil && f.writer.Process != nil {
+			if f.stopped {
+				f.writer.Process.Signal(syscall.SIGCONT)
+			}
+			f.input.Close()
+			select {
+			case <-f.writerDone:
+			case <-time.After(5 * time.Second):
+				f.writer.Process.Kill()
+				<-f.writerDone
+			}
+		}
+		select {
+		case <-f.helperDone:
+		default:
+			f.helper.Process.Signal(syscall.SIGTERM)
+			select {
+			case <-f.helperDone:
+			case <-time.After(5 * time.Second):
+				f.helper.Process.Kill()
+				<-f.helperDone
+				t.Error("private helper did not complete cleanup on termination")
+			}
+		}
+		f.output.Close()
+		// Emergency teardown for a failing fixture only, on this private Xvfb
+		// server after both processes exited. These codes are fixture-owned.
+		for _, code := range []uint16{f.shift, f.control} {
+			if held, err := probe.backend.Held(code); err == nil && held {
+				probe.backend.Set(code, false)
+				t.Error("helper left a fixture key held")
+			}
+		}
+	})
+	executable, err := os.Executable()
+	if err != nil {
+		t.Fatal(err)
+	}
+	f.writer = exec.Command(executable, "-test.run=^TestNativeRemoteKeyControlWriter$", "--", "remote-key-control-writer")
+	f.writer.ExtraFiles = []*os.File{keyWrite, lifeWrite}
+	f.writer.SysProcAttr = &syscall.SysProcAttr{Setpgid: true}
+	f.input, err = f.writer.StdinPipe()
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := f.writer.Start(); err != nil {
+		t.Fatal(err)
+	}
+	go func() { f.writerErr = f.writer.Wait(); close(f.writerDone) }()
+	writerGroup, err := syscall.Getpgid(f.writer.Process.Pid)
+	if err != nil {
+		t.Fatal(err)
+	}
+	helperGroup, err := syscall.Getpgid(f.helper.Process.Pid)
+	if err != nil || writerGroup == helperGroup || helperGroup != f.helper.Process.Pid {
+		t.Fatal("input helper was not isolated from its writer process group")
+	}
+	// Deferred closes now remove every supervisor copy of both write ends.
+	return f
+}
+
+func (f *nativeHelper) ack(t *testing.T) remoteinput.Ack {
+	t.Helper()
+	result := make(chan remoteinput.Ack, 1)
+	go func() {
+		frame := make([]byte, remoteinput.AckSize)
+		if _, err := io.ReadFull(f.output, frame); err != nil {
+			result <- remoteinput.Ack{}
+			return
+		}
+		ack, _ := remoteinput.DecodeAck(frame)
+		result <- ack
+	}()
+	select {
+	case ack := <-result:
+		if ack.Kind == 0 || ack.Epoch != 7 {
+			t.Fatal("missing or invalid native helper acknowledgement")
+		}
+		return ack
+	case <-time.After(5 * time.Second):
+		t.Fatal("private helper acknowledgement timed out")
+		return remoteinput.Ack{}
+	}
+}
+
+func (f *nativeHelper) send(t *testing.T, operation remoteinput.Operation) {
+	t.Helper()
+	frame, err := remoteinput.EncodeOperation(operation)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if _, err := f.input.Write(frame); err != nil {
+		t.Fatal(err)
+	}
+}
+
+func (f *nativeHelper) hold(t *testing.T) {
+	t.Helper()
+	if ack := f.ack(t); ack.Kind != remoteinput.Ready {
+		t.Fatal("helper did not become ready")
+	}
+	for index, name := range []string{"Shift_L", "Control_R"} {
+		f.send(t, remoteinput.Operation{Kind: remoteinput.Down, Epoch: 7, Sequence: uint64(index + 1), Name: name})
+		if ack := f.ack(t); ack.Kind != remoteinput.Accepted || ack.Sequence != uint64(index+1) {
+			t.Fatal("helper did not apply its owned modifier", ack)
+		}
+	}
+	remoteNativeHeld(t, f.probe, f.shift, true)
+	remoteNativeHeld(t, f.probe, f.control, true)
+}
+
+func (f *nativeHelper) exited(t *testing.T, success bool) {
+	t.Helper()
+	select {
+	case <-f.helperDone:
+		if (f.helperErr == nil) != success {
+			t.Fatalf("unexpected native helper exit: %v; diagnostics: %s", f.helperErr, f.stderr.String())
+		}
+	case <-time.After(5 * time.Second):
+		t.Fatal("native helper process did not exit")
+	}
+	remoteNativeHeld(t, f.probe, f.shift, false)
+	remoteNativeHeld(t, f.probe, f.control, false)
+}
+
+func TestNativeRemoteKeyHelperRetirement(t *testing.T) {
+	for _, reason := range []string{"close", "writer-kill", "writer-stop", "signal", "replay", "epoch"} {
+		t.Run(reason, func(t *testing.T) {
+			remoteNativeThread(t)
+			f := startNativeHelper(t)
+			f.hold(t)
+			wantReason, success := remoteinput.ReasonClose, true
+			switch reason {
+			case "close":
+				f.send(t, remoteinput.Operation{Kind: remoteinput.Close, Epoch: 7, Sequence: 3})
+			case "writer-kill":
+				if err := f.writer.Process.Kill(); err != nil {
+					t.Fatal(err)
+				}
+				wantReason = remoteinput.ReasonEOF
+			case "writer-stop":
+				if err := f.writer.Process.Signal(syscall.SIGSTOP); err != nil {
+					t.Fatal(err)
+				}
+				f.stopped = true
+				wantReason, success = remoteinput.ReasonExpired, false
+			case "signal":
+				// A partial key frame cannot delay the independent signal path.
+				if _, err := f.input.Write([]byte{remoteinput.Version}); err != nil {
+					t.Fatal(err)
+				}
+				if err := f.helper.Process.Signal(syscall.SIGTERM); err != nil {
+					t.Fatal(err)
+				}
+				wantReason = remoteinput.ReasonCancelled
+			case "replay":
+				f.send(t, remoteinput.Operation{Kind: remoteinput.Heartbeat, Epoch: 7, Sequence: 2})
+				wantReason, success = remoteinput.ReasonProtocol, false
+			case "epoch":
+				f.send(t, remoteinput.Operation{Kind: remoteinput.Heartbeat, Epoch: 8, Sequence: 3})
+				wantReason, success = remoteinput.ReasonProtocol, false
+			}
+			if ack := f.ack(t); ack.Kind != remoteinput.Closed || ack.Reason != wantReason || ack.CleanupFailed {
+				t.Fatal("helper retirement did not confirm native cleanup", ack)
+			}
+			f.exited(t, success)
+		})
+	}
+}
+
+func TestNativeRemoteKeyHelperHeartbeats(t *testing.T) {
+	remoteNativeThread(t)
+	f := startNativeHelper(t)
+	f.hold(t)
+	ticker := time.NewTicker(500 * time.Millisecond)
+	defer ticker.Stop()
+	for sequence := uint64(3); sequence <= 10; sequence++ {
+		<-ticker.C // Four seconds exceeds the initial three-second lifetime.
+		f.send(t, remoteinput.Operation{Kind: remoteinput.Heartbeat, Epoch: 7, Sequence: sequence})
+		if ack := f.ack(t); ack.Kind != remoteinput.Accepted || ack.Sequence != sequence {
+			t.Fatal("healthy heartbeat failed to retain the helper", ack)
+		}
+	}
+	f.send(t, remoteinput.Operation{Kind: remoteinput.Close, Epoch: 7, Sequence: 11})
+	if ack := f.ack(t); ack.Kind != remoteinput.Closed || ack.CleanupFailed {
+		t.Fatal("healthy helper did not clean up")
+	}
+	f.exited(t, true)
+}
+
+func TestNativeRemoteKeyHelperBlockedOutput(t *testing.T) {
+	remoteNativeThread(t)
+	f := startNativeHelper(t)
+	f.hold(t)
+	// Fill the actual inherited stdout pipe and bounded acknowledgement queue.
+	// The helper must release its keys and close both pollable pipe directions.
+	writesDone := make(chan error, 1)
+	go func() {
+		for sequence := uint64(3); sequence < 20000; sequence++ {
+			frame, _ := remoteinput.EncodeOperation(remoteinput.Operation{Kind: remoteinput.Heartbeat, Epoch: 7, Sequence: sequence})
+			if _, err := f.input.Write(frame); err != nil {
+				writesDone <- err
+				return
+			}
+		}
+		writesDone <- nil
+	}()
+	f.exited(t, false)
+	select {
+	case err := <-writesDone:
+		if err == nil {
+			t.Fatal("helper accepted unbounded output without acknowledgement consumption")
+		}
+	case <-time.After(5 * time.Second):
+		f.input.Close()
+		t.Fatal("helper exit left the private writer blocked")
+	}
+}

internal/platform/xorg/remote_key_helper.go

@@ -0,0 +1,39 @@
+// Copyright (c) 2026 Storm Dragon
+// Authors: Storm Dragon, Angus Kola
+// SPDX-License-Identifier: GPL-3.0-or-later
+
+package xorg
+
+import (
+	"context"
+	"io"
+	"runtime"
+
+	"git.stormux.org/storm/cthulhu/internal/remoteinput"
+)
+
+type remoteHelperTarget struct{ target *remoteKeyTarget }
+
+func (target remoteHelperTarget) Apply(name string, pressed bool) error {
+	return target.target.apply(name, pressed)
+}
+func (target remoteHelperTarget) Close() error { return target.target.close() }
+
+// ServeRemoteKeyHelper is for the independent source-built helper executable,
+// never the reader event loop. It supplies no consent or production mailbox
+// attachment. Physical collision/provenance and native policy gates remain open.
+func ServeRemoteKeyHelper(ctx context.Context, epoch uint64, control, lifetime io.ReadCloser, output io.WriteCloser) error {
+	if epoch == 0 || ctx == nil || ctx.Err() != nil || control == nil || lifetime == nil || output == nil {
+		return remoteinput.ErrProtocol
+	}
+	runtime.LockOSThread()
+	defer runtime.UnlockOSThread()
+	target, err := openRemoteKeyTarget()
+	if err != nil {
+		control.Close()
+		lifetime.Close()
+		output.Close()
+		return remoteinput.ErrTarget
+	}
+	return remoteinput.Run(ctx, epoch, control, lifetime, output, remoteHelperTarget{target: target})
+}

internal/remoteinput/pipe.go

@@ -0,0 +1,44 @@
+// Copyright (c) 2026 Storm Dragon
+// Authors: Storm Dragon, Angus Kola
+// SPDX-License-Identifier: GPL-3.0-or-later
+
+package remoteinput
+
+import (
+	"os"
+	"syscall"
+)
+
+// AdoptPipe transfers an inherited pipe to an owned nonblocking duplicate so
+// Go's poller can interrupt blocked I/O on Close. Terminals/regular files and
+// wrong-direction descriptors are rejected; a helper must have private control
+// pipes. Failure leaves the original file owned by its caller.
+func AdoptPipe(file *os.File, write bool) (*os.File, error) {
+	if file == nil {
+		return nil, ErrProtocol
+	}
+	info, err := file.Stat()
+	if err != nil || info.Mode()&os.ModeNamedPipe == 0 {
+		return nil, ErrProtocol
+	}
+	flags, _, failure := syscall.Syscall(syscall.SYS_FCNTL, file.Fd(), syscall.F_GETFL, 0)
+	want := syscall.O_RDONLY
+	if write {
+		want = syscall.O_WRONLY
+	}
+	if failure != 0 || int(flags)&syscall.O_ACCMODE != want {
+		return nil, ErrProtocol
+	}
+	fd, err := syscall.Dup(int(file.Fd()))
+	if err != nil {
+		return nil, ErrProtocol
+	}
+	syscall.CloseOnExec(fd)
+	if err := syscall.SetNonblock(fd, true); err != nil {
+		syscall.Close(fd)
+		return nil, ErrProtocol
+	}
+	owned := os.NewFile(uintptr(fd), "remote-input-pipe")
+	file.Close()
+	return owned, nil
+}

internal/remoteinput/pipe_test.go

@@ -0,0 +1,56 @@
+// Copyright (c) 2026 Storm Dragon
+// Authors: Storm Dragon, Angus Kola
+// SPDX-License-Identifier: GPL-3.0-or-later
+
+package remoteinput
+
+import (
+	"errors"
+	"os"
+	"testing"
+	"time"
+)
+
+func TestAdoptPipeInterruptsInheritedRead(t *testing.T) {
+	reader, writer, err := os.Pipe()
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer writer.Close()
+	if _, err := AdoptPipe(reader, true); !errors.Is(err, ErrProtocol) {
+		t.Fatal("read-only descriptor was accepted as helper output")
+	}
+	if _, err := AdoptPipe(writer, false); !errors.Is(err, ErrProtocol) {
+		t.Fatal("write-only descriptor was accepted as helper control")
+	}
+	owned, err := AdoptPipe(reader, false)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer owned.Close()
+	done := make(chan error, 1)
+	go func() { var value [1]byte; _, err := owned.Read(value[:]); done <- err }()
+	owned.Close()
+	select {
+	case err := <-done:
+		if err == nil {
+			t.Fatal("closed inherited pipe kept reading")
+		}
+	case <-time.After(time.Second):
+		t.Fatal("Close did not interrupt inherited read")
+	}
+	if _, err := reader.Stat(); err == nil {
+		t.Fatal("adoption left the original descriptor open")
+	}
+	file, err := os.CreateTemp(t.TempDir(), "not-a-pipe")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer file.Close()
+	if _, err := AdoptPipe(file, false); !errors.Is(err, ErrProtocol) {
+		t.Fatal("helper accepted a regular file")
+	}
+	if _, err := AdoptPipe(nil, false); !errors.Is(err, ErrProtocol) {
+		t.Fatal("helper accepted a missing descriptor")
+	}
+}

internal/remoteinput/protocol.go

@@ -0,0 +1,138 @@
+// Copyright (c) 2026 Storm Dragon
+// Authors: Storm Dragon, Angus Kola
+// SPDX-License-Identifier: GPL-3.0-or-later
+
+// Package remoteinput implements private, bounded input-helper IPC. Possession
+// of the inherited control pipe is not reader consent; only a future reader
+// owner may launch it for an already authorized lease.
+package remoteinput
+
+import (
+	"encoding/binary"
+	"errors"
+
+	"git.stormux.org/storm/cthulhu/internal/input"
+)
+
+const (
+	Version       = 1
+	OperationSize = 52
+	AckSize       = 20
+
+	Heartbeat byte = 1
+	Down      byte = 2
+	Up        byte = 3
+	Close     byte = 4
+
+	Ready    byte = 1
+	Accepted byte = 2
+	Closed   byte = 3
+
+	ReasonClose     byte = 1
+	ReasonEOF       byte = 2
+	ReasonExpired   byte = 3
+	ReasonProtocol  byte = 4
+	ReasonTarget    byte = 5
+	ReasonOutput    byte = 6
+	ReasonCancelled byte = 7
+)
+
+var ErrProtocol = errors.New("invalid private remote input protocol")
+
+type Operation struct {
+	Kind            byte
+	Epoch, Sequence uint64
+	Name            string
+}
+
+func EncodeOperation(operation Operation) ([]byte, error) {
+	if operation.Epoch == 0 || operation.Sequence == 0 || len(operation.Name) > 32 {
+		return nil, ErrProtocol
+	}
+	switch operation.Kind {
+	case Down, Up:
+		if !input.ValidRemoteKeyName(operation.Name) {
+			return nil, ErrProtocol
+		}
+	case Heartbeat, Close:
+		if operation.Name != "" {
+			return nil, ErrProtocol
+		}
+	default:
+		return nil, ErrProtocol
+	}
+	frame := make([]byte, OperationSize)
+	frame[0], frame[1], frame[2] = Version, operation.Kind, byte(len(operation.Name))
+	binary.BigEndian.PutUint64(frame[4:12], operation.Epoch)
+	binary.BigEndian.PutUint64(frame[12:20], operation.Sequence)
+	copy(frame[20:], operation.Name)
+	return frame, nil
+}
+
+func DecodeOperation(frame []byte) (Operation, error) {
+	if len(frame) != OperationSize || frame[0] != Version || frame[3] != 0 || frame[2] > 32 {
+		return Operation{}, ErrProtocol
+	}
+	length := int(frame[2])
+	for _, padding := range frame[20+length:] {
+		if padding != 0 {
+			return Operation{}, ErrProtocol
+		}
+	}
+	operation := Operation{Kind: frame[1], Epoch: binary.BigEndian.Uint64(frame[4:12]),
+		Sequence: binary.BigEndian.Uint64(frame[12:20]), Name: string(frame[20 : 20+length])}
+	if _, err := EncodeOperation(operation); err != nil {
+		return Operation{}, err
+	}
+	return operation, nil
+}
+
+// Accepted acknowledges a completed target Apply, not application delivery.
+// Only Closed with CleanupFailed=false confirms successful target cleanup.
+type Ack struct {
+	Kind, Reason    byte
+	CleanupFailed   bool
+	Epoch, Sequence uint64
+}
+
+func EncodeAck(ack Ack) ([]byte, error) {
+	if ack.Epoch == 0 {
+		return nil, ErrProtocol
+	}
+	switch ack.Kind {
+	case Ready:
+		if ack.Sequence != 0 || ack.Reason != 0 || ack.CleanupFailed {
+			return nil, ErrProtocol
+		}
+	case Accepted:
+		if ack.Sequence == 0 || ack.Reason != 0 || ack.CleanupFailed {
+			return nil, ErrProtocol
+		}
+	case Closed:
+		if ack.Reason < ReasonClose || ack.Reason > ReasonCancelled {
+			return nil, ErrProtocol
+		}
+	default:
+		return nil, ErrProtocol
+	}
+	frame := make([]byte, AckSize)
+	frame[0], frame[1], frame[2] = Version, ack.Kind, ack.Reason
+	if ack.CleanupFailed {
+		frame[3] = 1
+	}
+	binary.BigEndian.PutUint64(frame[4:12], ack.Epoch)
+	binary.BigEndian.PutUint64(frame[12:20], ack.Sequence)
+	return frame, nil
+}
+
+func DecodeAck(frame []byte) (Ack, error) {
+	if len(frame) != AckSize || frame[0] != Version || frame[3] > 1 {
+		return Ack{}, ErrProtocol
+	}
+	ack := Ack{Kind: frame[1], Reason: frame[2], CleanupFailed: frame[3] != 0,
+		Epoch: binary.BigEndian.Uint64(frame[4:12]), Sequence: binary.BigEndian.Uint64(frame[12:20])}
+	if _, err := EncodeAck(ack); err != nil {
+		return Ack{}, err
+	}
+	return ack, nil
+}

internal/remoteinput/protocol_test.go

@@ -0,0 +1,89 @@
+// Copyright (c) 2026 Storm Dragon
+// Authors: Storm Dragon, Angus Kola
+// SPDX-License-Identifier: GPL-3.0-or-later
+
+package remoteinput
+
+import (
+	"encoding/binary"
+	"errors"
+	"testing"
+)
+
+func TestHelperOperationCodec(t *testing.T) {
+	for _, operation := range []Operation{
+		{Kind: Down, Epoch: 7, Sequence: 1, Name: "Control_R"},
+		{Kind: Up, Epoch: 7, Sequence: 2, Name: "KP_Enter"},
+		{Kind: Heartbeat, Epoch: 7, Sequence: ^uint64(0)},
+		{Kind: Close, Epoch: ^uint64(0), Sequence: 1},
+	} {
+		frame, err := EncodeOperation(operation)
+		if err != nil || len(frame) != OperationSize || frame[0] != 1 || frame[1] != operation.Kind ||
+			binary.BigEndian.Uint64(frame[4:12]) != operation.Epoch || binary.BigEndian.Uint64(frame[12:20]) != operation.Sequence {
+			t.Fatal("wrong fixed operation representation", err)
+		}
+		if decoded, err := DecodeOperation(frame); err != nil || decoded != operation {
+			t.Fatal("operation identity changed", err)
+		}
+	}
+	for _, operation := range []Operation{
+		{Kind: 0, Epoch: 7, Sequence: 1}, {Kind: Heartbeat, Sequence: 1},
+		{Kind: Heartbeat, Epoch: 7}, {Kind: Down, Epoch: 7, Sequence: 1},
+		{Kind: Up, Epoch: 7, Sequence: 1, Name: "private unknown"},
+		{Kind: Down, Epoch: 7, Sequence: 1, Name: "a\x00"},
+		{Kind: Heartbeat, Epoch: 7, Sequence: 1, Name: "a"},
+		{Kind: Close, Epoch: 7, Sequence: 1, Name: "a"},
+	} {
+		if _, err := EncodeOperation(operation); !errors.Is(err, ErrProtocol) {
+			t.Fatal("invalid private operation was encoded")
+		}
+	}
+	frame, _ := EncodeOperation(Operation{Kind: Down, Epoch: 7, Sequence: 1, Name: "a"})
+	for _, index := range []int{0, 1, 2, 3, 21, OperationSize - 1} {
+		bad := append([]byte(nil), frame...)
+		bad[index] = 255
+		if value, err := DecodeOperation(bad); !errors.Is(err, ErrProtocol) || value != (Operation{}) {
+			t.Fatal("malformed version/kind/length/reserved/padding produced a partial operation")
+		}
+	}
+	for _, bad := range [][]byte{nil, frame[:OperationSize-1], append(frame, 0)} {
+		if _, err := DecodeOperation(bad); !errors.Is(err, ErrProtocol) {
+			t.Fatal("nonfixed frame was accepted")
+		}
+	}
+}
+
+func TestHelperAckCodec(t *testing.T) {
+	for _, ack := range []Ack{
+		{Kind: Ready, Epoch: 7}, {Kind: Accepted, Epoch: 7, Sequence: 1},
+		{Kind: Closed, Epoch: 7, Sequence: 2, Reason: ReasonClose},
+		{Kind: Closed, Epoch: 7, Sequence: ^uint64(0), Reason: ReasonEOF, CleanupFailed: true},
+	} {
+		frame, err := EncodeAck(ack)
+		if err != nil || len(frame) != AckSize {
+			t.Fatal("acknowledgement was not fixed size", err)
+		}
+		if decoded, err := DecodeAck(frame); err != nil || decoded != ack {
+			t.Fatal("acknowledgement lost cleanup/sequence identity", err)
+		}
+	}
+	for _, ack := range []Ack{
+		{Kind: Ready}, {Kind: Ready, Epoch: 7, Sequence: 1},
+		{Kind: Ready, Epoch: 7, CleanupFailed: true}, {Kind: Accepted, Epoch: 7},
+		{Kind: Accepted, Epoch: 7, Sequence: 1, Reason: ReasonClose},
+		{Kind: Closed, Epoch: 7}, {Kind: Closed, Epoch: 7, Reason: 255},
+		{Kind: 255, Epoch: 7},
+	} {
+		if _, err := EncodeAck(ack); !errors.Is(err, ErrProtocol) {
+			t.Fatal("invalid acknowledgement was encoded")
+		}
+	}
+	frame, _ := EncodeAck(Ack{Kind: Closed, Epoch: 7, Reason: ReasonEOF})
+	for _, index := range []int{0, 1, 2, 3} {
+		bad := append([]byte(nil), frame...)
+		bad[index] = 255
+		if value, err := DecodeAck(bad); !errors.Is(err, ErrProtocol) || value != (Ack{}) {
+			t.Fatal("invalid acknowledgement flags or version were accepted")
+		}
+	}
+}

internal/remoteinput/runner.go

@@ -0,0 +1,260 @@
+// Copyright (c) 2026 Storm Dragon
+// Authors: Storm Dragon, Angus Kola
+// SPDX-License-Identifier: GPL-3.0-or-later
+
+package remoteinput
+
+import (
+	"context"
+	"errors"
+	"io"
+	"sync"
+	"time"
+)
+
+const Watchdog = 3 * time.Second
+const ackQueueSize = 32
+const finalAckTimeout = 250 * time.Millisecond
+
+var (
+	ErrExpired = errors.New("remote input helper heartbeat expired")
+	ErrTarget  = errors.New("remote input helper target failed")
+	ErrOutput  = errors.New("remote input helper acknowledgements unavailable")
+	ErrCleanup = errors.New("remote input helper cleanup failed")
+	errEOF     = errors.New("remote input helper owner closed")
+)
+
+// Target methods run on Run's caller, never on its I/O goroutines. Close must
+// retire keys before destroying native state. A blocked native method cannot
+// be interrupted here: watchdog recovery requires a responsive native server.
+type Target interface {
+	Apply(name string, pressed bool) error
+	Close() error
+}
+
+type ackWrite struct {
+	ack  Ack
+	done chan error
+}
+
+// Run owns valid supplied streams and target through cleanup and I/O join.
+// Stream Close must interrupt blocked reads/writes; AdoptPipe supplies that
+// property for inherited Go file descriptors. The caller supplies one fixed
+// nonzero epoch and must not launch a replacement until cleanup is confirmed.
+func Run(ctx context.Context, epoch uint64, control, lifetime io.ReadCloser, output io.WriteCloser, target Target) error {
+	return run(ctx, epoch, control, lifetime, output, target, Watchdog)
+}
+
+func run(ctx context.Context, epoch uint64, control, lifetime io.ReadCloser, output io.WriteCloser, target Target, timeout time.Duration) error {
+	if ctx == nil || epoch == 0 || control == nil || lifetime == nil || output == nil || target == nil || timeout <= 0 {
+		return ErrProtocol
+	}
+	ownerContext, cancelOwner := context.WithCancelCause(ctx)
+	defer cancelOwner(context.Canceled)
+	ioContext, stopIO := context.WithCancel(context.Background())
+	results := make(chan Operation, 1)
+	acks := make(chan ackWrite, ackQueueSize)
+	writeFailure := make(chan struct{}, 1)
+	var workers sync.WaitGroup
+	workers.Add(3)
+	go func() {
+		defer workers.Done()
+		// This pipe never carries data. Its EOF cannot sit behind queued keys.
+		var value [1]byte
+		n, err := lifetime.Read(value[:])
+		if n == 0 && err == io.EOF {
+			cancelOwner(errEOF)
+		} else {
+			cancelOwner(ErrProtocol)
+		}
+	}()
+	go func() {
+		defer workers.Done()
+		for {
+			frame := make([]byte, OperationSize)
+			_, err := io.ReadFull(control, frame)
+			var operation Operation
+			if err == nil {
+				operation, err = DecodeOperation(frame)
+			}
+			if err != nil {
+				if err == io.EOF {
+					cancelOwner(errEOF)
+				} else {
+					cancelOwner(ErrProtocol)
+				}
+				return
+			}
+			select {
+			case results <- operation:
+			case <-ioContext.Done():
+				return
+			}
+		}
+	}()
+	go func() {
+		defer workers.Done()
+		for {
+			select {
+			case <-ioContext.Done():
+				return
+			case item := <-acks:
+				if ioContext.Err() != nil {
+					return
+				}
+				frame, err := EncodeAck(item.ack)
+				if err == nil {
+					err = writeAll(output, frame)
+				}
+				if item.done != nil {
+					item.done <- err
+				}
+				if err != nil {
+					cancelOwner(ErrOutput)
+					writeFailure <- struct{}{}
+					return
+				}
+			}
+		}
+	}()
+	defer func() {
+		stopIO()
+		_ = control.Close()
+		_ = lifetime.Close()
+		_ = output.Close()
+		workers.Wait()
+	}()
+	enqueue := func(ack Ack, done chan error) bool {
+		select {
+		case acks <- ackWrite{ack: ack, done: done}:
+			return true
+		default:
+			return false
+		}
+	}
+	deadline := time.Now().Add(timeout)
+	timer := time.NewTimer(timeout)
+	defer timer.Stop()
+	var sequence uint64
+	var cause error
+	reason := ReasonEOF
+	ownerStopped := func() {
+		cause = context.Cause(ownerContext)
+		switch cause {
+		case errEOF:
+			cause, reason = nil, ReasonEOF
+		case ErrProtocol:
+			reason = ReasonProtocol
+		case ErrOutput:
+			reason = ReasonOutput
+		default:
+			cause = ctx.Err()
+			if cause == nil {
+				cause = context.Canceled
+			}
+			reason = ReasonCancelled
+		}
+	}
+	enqueue(Ack{Kind: Ready, Epoch: epoch}, nil)
+loop:
+	for {
+		select {
+		case <-ownerContext.Done():
+			ownerStopped()
+			break loop
+		case <-writeFailure:
+			cause, reason = ErrOutput, ReasonOutput
+			break loop
+		case <-timer.C:
+			if time.Now().Before(deadline) {
+				timer.Reset(time.Until(deadline))
+				continue
+			}
+			cause, reason = ErrExpired, ReasonExpired
+			break loop
+		case operation := <-results:
+			// Check before every frame, including a heartbeat. Busy input or a
+			// late heartbeat cannot win a select race against expired authority.
+			if ownerContext.Err() != nil {
+				ownerStopped()
+				break loop
+			}
+			if !time.Now().Before(deadline) {
+				cause, reason = ErrExpired, ReasonExpired
+				break loop
+			}
+			if operation.Epoch != epoch || operation.Sequence <= sequence {
+				cause, reason = ErrProtocol, ReasonProtocol
+				break loop
+			}
+			sequence = operation.Sequence
+			switch operation.Kind {
+			case Close:
+				reason = ReasonClose
+				break loop
+			case Heartbeat:
+				deadline = time.Now().Add(timeout)
+				timer.Reset(timeout)
+			case Down, Up:
+				if err := target.Apply(operation.Name, operation.Kind == Down); err != nil {
+					cause, reason = ErrTarget, ReasonTarget
+					break loop
+				}
+			}
+			if ownerContext.Err() != nil {
+				ownerStopped()
+				break loop
+			}
+			if !time.Now().Before(deadline) {
+				cause, reason = ErrExpired, ReasonExpired
+				break loop
+			}
+			if !enqueue(Ack{Kind: Accepted, Epoch: epoch, Sequence: sequence}, nil) {
+				cause, reason = ErrOutput, ReasonOutput
+				break loop
+			}
+		}
+	}
+	// No acknowledgement or blocked pipe can postpone native cleanup.
+	_ = control.Close()
+	cleanup := target.Close()
+	done := make(chan error, 1)
+	if enqueue(Ack{Kind: Closed, Reason: reason, CleanupFailed: cleanup != nil, Epoch: epoch, Sequence: sequence}, done) {
+		flush := time.NewTimer(finalAckTimeout)
+		select {
+		case err := <-done:
+			if err != nil && cause == nil {
+				cause = ErrOutput
+			}
+		case <-writeFailure:
+			if cause == nil {
+				cause = ErrOutput
+			}
+		case <-flush.C:
+			if cause == nil {
+				cause = ErrOutput
+			}
+		}
+		flush.Stop()
+	} else if cause == nil {
+		cause = ErrOutput
+	}
+	if cleanup != nil {
+		return ErrCleanup
+	}
+	return cause
+}
+
+func writeAll(writer io.Writer, frame []byte) error {
+	for len(frame) > 0 {
+		n, err := writer.Write(frame)
+		if err != nil {
+			return err
+		}
+		if n <= 0 || n > len(frame) {
+			return io.ErrShortWrite
+		}
+		frame = frame[n:]
+	}
+	return nil
+}

internal/remoteinput/runner_test.go

@@ -0,0 +1,318 @@
+// Copyright (c) 2026 Storm Dragon
+// Authors: Storm Dragon, Angus Kola
+// SPDX-License-Identifier: GPL-3.0-or-later
+
+package remoteinput
+
+import (
+	"context"
+	"errors"
+	"io"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+)
+
+type helperTarget struct {
+	mu                       sync.Mutex
+	keys                     []Operation
+	failApply, failClose     bool
+	closeCount               int
+	applied, closed, blocked chan struct{}
+	gate                     chan struct{}
+	applyOnce, closeOnce     sync.Once
+}
+
+func (target *helperTarget) Apply(name string, pressed bool) error {
+	target.mu.Lock()
+	kind := Up
+	if pressed {
+		kind = Down
+	}
+	target.keys = append(target.keys, Operation{Kind: kind, Name: name})
+	gate, fail := target.gate, target.failApply
+	target.mu.Unlock()
+	target.applyOnce.Do(func() { close(target.applied) })
+	if gate != nil {
+		close(target.blocked)
+		<-gate
+	}
+	if fail {
+		return errors.New("private target input")
+	}
+	return nil
+}
+func (target *helperTarget) Close() error {
+	target.mu.Lock()
+	target.closeCount++
+	fail := target.failClose
+	target.mu.Unlock()
+	target.closeOnce.Do(func() { close(target.closed) })
+	if fail {
+		return errors.New("private cleanup diagnostics")
+	}
+	return nil
+}
+
+type helperFixture struct {
+	control, life *io.PipeWriter
+	output        *io.PipeReader
+	target        *helperTarget
+	cancel        context.CancelFunc
+	done          chan struct{}
+	err           error
+}
+
+func startHelper(t *testing.T, timeout time.Duration) *helperFixture {
+	t.Helper()
+	reader, control := io.Pipe()
+	lifetime, life := io.Pipe()
+	output, writer := io.Pipe()
+	ctx, cancel := context.WithCancel(t.Context())
+	target := &helperTarget{applied: make(chan struct{}), closed: make(chan struct{}), blocked: make(chan struct{})}
+	f := &helperFixture{control: control, life: life, output: output, target: target, cancel: cancel, done: make(chan struct{})}
+	go func() {
+		f.err = run(ctx, 7, reader, lifetime, writer, target, timeout)
+		close(f.done)
+	}()
+	t.Cleanup(func() {
+		cancel()
+		control.Close()
+		life.Close()
+		output.Close()
+		select {
+		case <-f.done:
+		case <-time.After(2 * time.Second):
+			t.Error("helper I/O workers did not join")
+		}
+	})
+	return f
+}
+
+func (f *helperFixture) ack(t *testing.T) Ack {
+	t.Helper()
+	result := make(chan Ack, 1)
+	go func() {
+		frame := make([]byte, AckSize)
+		if _, err := io.ReadFull(f.output, frame); err != nil {
+			result <- Ack{}
+			return
+		}
+		ack, _ := DecodeAck(frame)
+		result <- ack
+	}()
+	select {
+	case ack := <-result:
+		if ack.Kind == 0 {
+			t.Fatal("invalid/missing helper acknowledgement")
+		}
+		return ack
+	case <-time.After(2 * time.Second):
+		t.Fatal("helper acknowledgement timed out")
+		return Ack{}
+	}
+}
+
+func (f *helperFixture) send(t *testing.T, operation Operation) {
+	t.Helper()
+	frame, err := EncodeOperation(operation)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := writeAll(f.control, frame); err != nil {
+		t.Fatal(err)
+	}
+}
+
+func (f *helperFixture) finish(t *testing.T, want error) {
+	t.Helper()
+	select {
+	case <-f.done:
+		if !errors.Is(f.err, want) {
+			t.Fatalf("helper result = %v; want %v", f.err, want)
+		}
+		f.target.mu.Lock()
+		defer f.target.mu.Unlock()
+		if f.target.closeCount != 1 {
+			t.Fatal("target cleanup was not performed exactly once")
+		}
+	case <-time.After(2 * time.Second):
+		t.Fatal("helper cleanup did not finish")
+	}
+}
+
+func TestHelperRunnerOrderedClose(t *testing.T) {
+	f := startHelper(t, time.Second)
+	if ack := f.ack(t); ack.Kind != Ready || ack.Epoch != 7 || ack.Sequence != 0 {
+		t.Fatal("wrong helper readiness identity")
+	}
+	for index, kind := range []byte{Down, Heartbeat, Up, Close} {
+		operation := Operation{Kind: kind, Epoch: 7, Sequence: uint64(index + 1)}
+		if kind == Down || kind == Up {
+			operation.Name = "Control_R"
+		}
+		if index == 0 {
+			frame, _ := EncodeOperation(operation)
+			if err := writeAll(f.control, frame[:7]); err != nil {
+				t.Fatal(err)
+			}
+			if err := writeAll(f.control, frame[7:]); err != nil {
+				t.Fatal(err)
+			}
+		} else {
+			f.send(t, operation)
+		}
+		ack := f.ack(t)
+		if ack.Sequence != operation.Sequence || ack.Epoch != 7 || ack.CleanupFailed {
+			t.Fatal("helper lost epoch, sequence or cleanup identity")
+		}
+		if kind == Close {
+			if ack.Kind != Closed || ack.Reason != ReasonClose {
+				t.Fatal("close was acknowledged before cleanup")
+			}
+		} else if ack.Kind != Accepted {
+			t.Fatal("valid operation was not accepted")
+		}
+	}
+	f.finish(t, nil)
+	if len(f.target.keys) != 2 || f.target.keys[0].Kind != Down || f.target.keys[1].Kind != Up {
+		t.Fatal("helper changed input order or synthesized keys")
+	}
+}
+
+func TestHelperRunnerRejectsStaleAndMalformed(t *testing.T) {
+	for _, failure := range []string{"epoch", "replay", "partial", "reserved", "lifetime-data"} {
+		t.Run(failure, func(t *testing.T) {
+			f := startHelper(t, time.Second)
+			f.ack(t)
+			f.send(t, Operation{Kind: Down, Epoch: 7, Sequence: 1, Name: "Shift_L"})
+			f.ack(t)
+			op := Operation{Kind: Heartbeat, Epoch: 7, Sequence: 2}
+			switch failure {
+			case "epoch":
+				op.Epoch++
+				f.send(t, op)
+			case "replay":
+				op.Sequence = 1
+				f.send(t, op)
+			case "partial":
+				if _, err := f.control.Write([]byte{Version}); err != nil {
+					t.Fatal(err)
+				}
+				f.control.Close()
+			case "reserved":
+				frame, _ := EncodeOperation(op)
+				frame[3] = 1
+				if err := writeAll(f.control, frame); err != nil {
+					t.Fatal(err)
+				}
+			case "lifetime-data":
+				if _, err := f.life.Write([]byte{1}); err != nil {
+					t.Fatal(err)
+				}
+			}
+			if ack := f.ack(t); ack.Kind != Closed || ack.Reason != ReasonProtocol || ack.CleanupFailed {
+				t.Fatal("invalid input did not retire with cleanup", ack)
+			}
+			f.finish(t, ErrProtocol)
+			if len(f.target.keys) != 1 {
+				t.Fatal("invalid input reached the target")
+			}
+		})
+	}
+}
+
+func TestHelperRunnerLifetimeEOFCancelsPartialControl(t *testing.T) {
+	f := startHelper(t, time.Second)
+	f.ack(t)
+	f.send(t, Operation{Kind: Down, Epoch: 7, Sequence: 1, Name: "Shift_L"})
+	f.ack(t)
+	if _, err := f.control.Write([]byte{Version}); err != nil {
+		t.Fatal(err)
+	}
+	f.life.Close() // Control remains open with an incomplete frame.
+	if ack := f.ack(t); ack.Kind != Closed || ack.Reason != ReasonEOF || ack.CleanupFailed {
+		t.Fatal("lifeline EOF waited for incomplete control data", ack)
+	}
+	f.finish(t, nil)
+}
+
+func TestHelperRunnerCancellationDiscardsQueuedDown(t *testing.T) {
+	f := startHelper(t, time.Second)
+	f.ack(t)
+	gate := make(chan struct{})
+	f.target.mu.Lock()
+	f.target.gate = gate
+	f.target.mu.Unlock()
+	f.send(t, Operation{Kind: Down, Epoch: 7, Sequence: 1, Name: "Shift_L"})
+	<-f.target.blocked
+	f.send(t, Operation{Kind: Down, Epoch: 7, Sequence: 2, Name: "a"})
+	f.cancel() // Synchronously invalidates the owner's context.
+	close(gate)
+	if ack := f.ack(t); ack.Kind != Closed || ack.Reason != ReasonCancelled {
+		t.Fatal("cancelled key was acknowledged")
+	}
+	f.finish(t, context.Canceled)
+	if len(f.target.keys) != 1 || f.target.keys[0].Name != "Shift_L" {
+		t.Fatal("queued down ran after cancellation")
+	}
+}
+
+func TestHelperRunnerBlockedAcksDoNotPostponeWatchdog(t *testing.T) {
+	f := startHelper(t, 50*time.Millisecond)
+	// Leave Ready unread, blocking the writer before it can write any ACK.
+	f.send(t, Operation{Kind: Down, Epoch: 7, Sequence: 1, Name: "Shift_L"})
+	select {
+	case <-f.target.closed:
+	case <-time.After(time.Second):
+		t.Fatal("blocked output postponed native cleanup")
+	}
+	f.finish(t, ErrExpired)
+}
+
+func TestHelperRunnerKeyTrafficDoesNotRenewWatchdog(t *testing.T) {
+	f := startHelper(t, 100*time.Millisecond)
+	f.ack(t)
+	deadline := time.Now().Add(time.Second)
+	for sequence := uint64(1); ; sequence++ {
+		if time.Now().After(deadline) {
+			t.Fatal("key-only traffic renewed helper authority")
+		}
+		frame, _ := EncodeOperation(Operation{Kind: Down, Epoch: 7, Sequence: sequence, Name: "a"})
+		err := writeAll(f.control, frame)
+		ack := f.ack(t)
+		if ack.Kind == Closed {
+			if ack.Reason != ReasonExpired {
+				t.Fatal("key-only traffic stopped for an unexpected reason", ack)
+			}
+			break
+		}
+		if err != nil || ack.Kind != Accepted {
+			t.Fatal("unexpected key-only acknowledgement", err, ack)
+		}
+	}
+	f.finish(t, ErrExpired)
+}
+
+func TestHelperRunnerTargetAndCleanupErrorsArePrivate(t *testing.T) {
+	for _, cleanupFailure := range []bool{false, true} {
+		f := startHelper(t, time.Second)
+		f.ack(t)
+		f.target.mu.Lock()
+		f.target.failApply, f.target.failClose = true, cleanupFailure
+		f.target.mu.Unlock()
+		f.send(t, Operation{Kind: Down, Epoch: 7, Sequence: 1, Name: "a"})
+		if ack := f.ack(t); ack.Kind != Closed || ack.Reason != ReasonTarget || ack.CleanupFailed != cleanupFailure {
+			t.Fatal("native failure lost its cleanup status", ack)
+		}
+		want := ErrTarget
+		if cleanupFailure {
+			want = ErrCleanup
+		}
+		f.finish(t, want)
+		if strings.Contains(f.err.Error(), "private") {
+			t.Fatal("target error leaked input or diagnostics")
+		}
+	}
+}