Add embedded controlled-session broker

.changes/unreleased/+controlled-session-controller-broker.yaml

@@ -0,0 +1,2 @@
+kind: Added
+body: Add the embedded `reploy controlled-session client` broker with a strict versioned controller stream and a private, bounded terminal transport for OmegaFlow-controlled workloads.

docs/CONTROLLED_SESSION_DESIGN.md

@@ -1248,6 +1248,10 @@ closes the private host connection, and therefore cannot be mistaken for
 successful completion. Repeated `terminate` remains idempotent and repeated
 valid resize requests remain ordinary operations.
 
+Each broker-to-host request write is bounded to one second. Expiration is a
+fatal transport failure, so host-side PTY backpressure cannot hold the broker's
+control loop indefinitely.
+
 The public `opened` projection contains only the operations granted to the
 controller, endpoint coordinates, terminal dimensions, and output-finalization
 timeout. It does not expose the host-internal authorization record or its
@@ -1282,6 +1286,10 @@ content, or workload filesystem.
 
 The broker emits `broker-ready` after the listener exists and then claims the
 Host Reploy channel. It allows ten seconds for the one attachment to connect.
+Request and disconnect monitoring begins as soon as that attachment is
+accepted, including while the host claim is still waiting for `opened`; input
+and the latest resize received in that interval join the same bounded
+pre-`ready` buffer. An attachment EOF in that interval fails immediately.
 Before attachment, it may hold at most one complete private-protocol output
 frame and otherwise applies backpressure to Host Reploy. Attach timeout,
 unexpected attachment exit, malformed attachment traffic, or loss before the
@@ -1293,7 +1301,9 @@ resize, and terminal-end records so terminal bytes cannot be interpreted as
 control. The attachment switches its asciinema-owned PTY to raw mode while
 running, restores it on exit, forwards ordinary Ctrl-C as byte `0x03`, and
 translates `SIGWINCH` into resize records. It writes received output bytes to
-stdout unchanged and never writes diagnostics there.
+stdout unchanged and never writes diagnostics there. Input or resize already
+in flight when host termination latches is discarded without changing the
+termination cause or reporting controller loss.
 
 For an activated session, the broker sends terminal-end only after it has
 forwarded every earlier output byte and then received the ordered

internal/cli/cli.go

@@ -19,6 +19,7 @@ import (
 
 	reploy "github.com/omry/reploy"
 	"github.com/omry/reploy/internal/blueprint"
+	"github.com/omry/reploy/internal/controlledsession"
 	"github.com/omry/reploy/internal/deploy"
 	"github.com/omry/reploy/internal/dockerdeploy"
 	"github.com/omry/reploy/internal/overrideui"
@@ -49,6 +50,7 @@ var dockerAppCommand = dockerdeploy.AppCommand
 var runOverrideEditor = overrideui.RunWithResult
 var runBuildProgress = overrideui.RunBuildProgress
 var inspectStagedOverrideValidation = dockerdeploy.InspectStagedOverrideValidation
+var runControlledSessionBroker = controlledsession.RunControllerBrokerV1
 
 func Main(args []string, stdout io.Writer, stderr io.Writer) int {
 	if message := windowsWSLBoundaryError(runtime.GOOS, os.LookupEnv, os.Getwd); message != "" {
@@ -85,6 +87,8 @@ func Main(args []string, stdout io.Writer, stderr io.Writer) int {
 		return runBlueprintValidate(args[1:], stdout, stderr)
 	case "services":
 		return runServices(args[1:], stdout, stderr)
+	case "controlled-session":
+		return runControlledSession(args[1:], stdout, stderr)
 	default:
 		if isDeploymentCommand(args[0]) {
 			return runDocker(args, stdout, stderr, globalOptions)
@@ -99,6 +103,32 @@ func Main(args []string, stdout io.Writer, stderr io.Writer) int {
 	}
 }
 
+func runControlledSession(args []string, stdout io.Writer, stderr io.Writer) int {
+	if (len(args) == 1 && isHelpArg(args[0])) || (len(args) == 2 && args[0] == "client" && isHelpArg(args[1])) {
+		printControlledSessionHelp(stdout)
+		return 0
+	}
+	if len(args) != 1 || args[0] != "client" {
+		fmt.Fprintln(stderr, "reploy controlled-session usage error: expected client")
+		printControlledSessionShortUsage(stderr)
+		return 2
+	}
+	socket := os.Getenv("REPLOY_SESSION_SOCKET")
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	err := runControlledSessionBroker(ctx, controlledsession.ControllerBrokerOptionsV1{
+		SessionSocket: socket,
+		TemporaryHome: controlledsession.ControllerTemporaryHomeV1,
+		Input:         os.Stdin,
+		Output:        stdout,
+	})
+	if err != nil {
+		fmt.Fprintf(stderr, "reploy controlled-session client error: %v\n", err)
+		return 1
+	}
+	return 0
+}
+
 func runEmbeddedServiceContainer(args []string, stdout io.Writer, stderr io.Writer, globalOptions globalDeploymentOptions) int {
 	dir := ""
 	dockerPath := ""
@@ -3174,6 +3204,8 @@ Commands:
   install      Install or update a deployed host service
   uninstall    Remove an installed host service and Docker resources
   services     List Reploy-managed services
+  controlled-session
+               Run controller-side controlled-session integration commands
   index        Manage the cached blueprint shorthand index
   version      Print version information
 
@@ -3209,6 +3241,21 @@ Run 'reploy COMMAND --help' for command-specific options.
 `, "\n"))
 }
 
+func printControlledSessionShortUsage(output io.Writer) {
+	fmt.Fprintln(output, "Usage: reploy controlled-session client")
+}
+
+func printControlledSessionHelp(output io.Writer) {
+	fmt.Fprint(output, strings.TrimLeft(`
+Usage: reploy controlled-session client
+
+Run the controller-side controlled-session broker. The command consumes the
+controller-private REPLOY_SESSION_SOCKET environment variable and exchanges
+versioned JSON Lines with the controller orchestrator on stdin and stdout.
+Human-readable diagnostics are written only to stderr.
+`, "\n"))
+}
+
 func printPackIndexShortUsage(commandName string, output io.Writer) {
 	fmt.Fprintf(output, "Usage: reploy %s COMMAND\n", commandName)
 	fmt.Fprintln(output)

internal/cli/controlled_session_test.go

@@ -0,0 +1,59 @@
+package cli
+
+import (
+	"context"
+	"errors"
+	"strings"
+	"testing"
+
+	"github.com/omry/reploy/internal/controlledsession"
+)
+
+func TestControlledSessionHelpAndUsage(t *testing.T) {
+	code, stdout, stderr := runCLI("controlled-session", "--help")
+	if code != 0 || stderr != "" || !strings.Contains(stdout, "Usage: reploy controlled-session client") || !strings.Contains(stdout, "REPLOY_SESSION_SOCKET") {
+		t.Fatalf("help code=%d stdout=%q stderr=%q", code, stdout, stderr)
+	}
+	code, stdout, stderr = runCLI("controlled-session", "client", "--help")
+	if code != 0 || stderr != "" || !strings.Contains(stdout, "Usage: reploy controlled-session client") {
+		t.Fatalf("client help code=%d stdout=%q stderr=%q", code, stdout, stderr)
+	}
+	code, stdout, stderr = runCLI("controlled-session")
+	if code != 2 || stdout != "" || !strings.Contains(stderr, "expected client") || !strings.Contains(stderr, "Usage: reploy controlled-session client") {
+		t.Fatalf("usage code=%d stdout=%q stderr=%q", code, stdout, stderr)
+	}
+	code, stdout, stderr = runCLI("--help")
+	if code != 0 || stderr != "" || !strings.Contains(stdout, "controlled-session") {
+		t.Fatalf("top-level help code=%d stdout=%q stderr=%q", code, stdout, stderr)
+	}
+}
+
+func TestControlledSessionClientDispatchesEmbeddedBroker(t *testing.T) {
+	original := runControlledSessionBroker
+	t.Cleanup(func() { runControlledSessionBroker = original })
+	t.Setenv("REPLOY_SESSION_SOCKET", "/run/reploy/session.sock")
+	called := false
+	runControlledSessionBroker = func(ctx context.Context, options controlledsession.ControllerBrokerOptionsV1) error {
+		called = true
+		if ctx == nil || ctx.Done() == nil || options.SessionSocket != "/run/reploy/session.sock" || options.TemporaryHome != controlledsession.ControllerTemporaryHomeV1 || options.Input == nil || options.Output == nil {
+			t.Fatalf("broker options = %#v", options)
+		}
+		return nil
+	}
+	code, stdout, stderr := runCLI("controlled-session", "client")
+	if code != 0 || stdout != "" || stderr != "" || !called {
+		t.Fatalf("client code=%d stdout=%q stderr=%q called=%t", code, stdout, stderr, called)
+	}
+}
+
+func TestControlledSessionClientReportsRuntimeFailureOnlyOnStderr(t *testing.T) {
+	original := runControlledSessionBroker
+	t.Cleanup(func() { runControlledSessionBroker = original })
+	runControlledSessionBroker = func(context.Context, controlledsession.ControllerBrokerOptionsV1) error {
+		return errors.New("controlled-session controller broker requires Linux")
+	}
+	code, stdout, stderr := runCLI("controlled-session", "client")
+	if code != 1 || stdout != "" || !strings.Contains(stderr, "requires Linux") {
+		t.Fatalf("failure code=%d stdout=%q stderr=%q", code, stdout, stderr)
+	}
+}

internal/controlledsession/client.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"io"
 	"net"
 	"sync"
 )
@@ -79,7 +80,14 @@ func (client *SessionClientV1) ReadEvent(ctx context.Context) (EventV1, error) {
 		return readErr
 	})
 	if err != nil {
-		return EventV1{}, errors.Join(fmt.Errorf("read controlled-session client event: %w", err), client.Close())
+		readErr := fmt.Errorf("read controlled-session client event: %w", err)
+		client.stateMu.RLock()
+		terminated := client.terminated
+		client.stateMu.RUnlock()
+		if terminated && errors.Is(err, io.EOF) {
+			return EventV1{}, readErr
+		}
+		return EventV1{}, errors.Join(readErr, client.Close())
 	}
 	if event.Kind == EventOpenedV1 {
 		return EventV1{}, errors.Join(fmt.Errorf("read controlled-session client event: opened may appear only once"), client.Close())

internal/controlledsession/client_test.go

@@ -1,7 +1,10 @@
 package controlledsession
 
 import (
+	"bytes"
 	"context"
+	"errors"
+	"io"
 	"net"
 	"reflect"
 	"strings"
@@ -170,6 +173,51 @@ func TestSessionClientV1AcknowledgesStartupFailureBeforeReady(t *testing.T) {
 	}
 }
 
+func TestSessionClientV1KeepsPostTerminatedEOFOpenForAcknowledgement(t *testing.T) {
+	var events bytes.Buffer
+	opened := testOpenedV1()
+	terminated := EventV1{Kind: EventTerminatedV1, Terminated: &ResultV1{
+		Cause:                            CauseStartupFailureV1,
+		WorkloadStatus:                   ProcessStatusV1{Kind: ProcessStatusUnknownV1},
+		WorkloadOutputFinalizationStatus: WorkloadOutputFinalizationStatusV1{Kind: WorkloadOutputFinalizationDrainedV1},
+		RuntimeObservationStatus:         RuntimeObservationStatusV1{Kind: RuntimeObservationMaintainedV1},
+		ControllerFinalizationStatus:     ControllerFinalizationStatusV1{Kind: ControllerFinalizationStartupFailedV1},
+		CleanupStatus:                    CleanupStatusV1{Kind: CleanupStatusSucceededV1},
+		RecoveryAction:                   RecoveryNoneV1,
+	}}
+	if err := WriteEventV1(&events, EventV1{Kind: EventOpenedV1, Opened: &opened}); err != nil {
+		t.Fatal(err)
+	}
+	if err := WriteEventV1(&events, terminated); err != nil {
+		t.Fatal(err)
+	}
+	connection := &scriptedSessionClientConnectionV1{reader: bytes.NewReader(events.Bytes())}
+	ctx, cancel := context.WithTimeout(t.Context(), time.Second)
+	defer cancel()
+	client, err := newSessionClientV1(ctx, connection)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer client.Close()
+	if event, err := client.ReadEvent(ctx); err != nil || !reflect.DeepEqual(event, terminated) {
+		t.Fatalf("terminated event = %#v, error = %v", event, err)
+	}
+	if _, err := client.ReadEvent(ctx); !errors.Is(err, io.EOF) {
+		t.Fatalf("post-terminated read error = %v", err)
+	}
+	if connection.closed {
+		t.Fatal("post-terminated EOF closed the connection before acknowledgement")
+	}
+	wantRequest := RequestV1{Kind: RequestAcknowledgeTerminatedV1}
+	if err := client.WriteRequest(ctx, wantRequest); err != nil {
+		t.Fatal(err)
+	}
+	request, err := ReadRequestV1(bytes.NewReader(connection.writes.Bytes()))
+	if err != nil || !reflect.DeepEqual(request, wantRequest) {
+		t.Fatalf("acknowledgement = %#v, error = %v", request, err)
+	}
+}
+
 func TestSessionClientV1RequiresOpenedFirstAndRejectsRepeatedOpened(t *testing.T) {
 	for _, test := range []struct {
 		name   string
@@ -217,3 +265,40 @@ func testOpenedV1() OpenedV1 {
 }
 
 func pointerToOpenedV1(value OpenedV1) *OpenedV1 { return &value }
+
+type scriptedSessionClientConnectionV1 struct {
+	reader *bytes.Reader
+	writes bytes.Buffer
+	closed bool
+}
+
+func (connection *scriptedSessionClientConnectionV1) Read(payload []byte) (int, error) {
+	if connection.closed {
+		return 0, net.ErrClosed
+	}
+	return connection.reader.Read(payload)
+}
+
+func (connection *scriptedSessionClientConnectionV1) Write(payload []byte) (int, error) {
+	if connection.closed {
+		return 0, net.ErrClosed
+	}
+	return connection.writes.Write(payload)
+}
+
+func (connection *scriptedSessionClientConnectionV1) Close() error {
+	connection.closed = true
+	return nil
+}
+
+func (*scriptedSessionClientConnectionV1) LocalAddr() net.Addr {
+	return &net.UnixAddr{Name: "scripted-local", Net: "unix"}
+}
+
+func (*scriptedSessionClientConnectionV1) RemoteAddr() net.Addr {
+	return &net.UnixAddr{Name: "scripted-remote", Net: "unix"}
+}
+
+func (*scriptedSessionClientConnectionV1) SetDeadline(time.Time) error      { return nil }
+func (*scriptedSessionClientConnectionV1) SetReadDeadline(time.Time) error  { return nil }
+func (*scriptedSessionClientConnectionV1) SetWriteDeadline(time.Time) error { return nil }