Bridge controlled-session controller and workload I/O

docs/CONTROLLED_SESSION_DESIGN.md

@@ -38,9 +38,16 @@ summary: Capability-scoped execution sessions that inherit Reploy's global conta
   before container creation, creates the frozen controller plan inert, starts
   it at most once, independently observes its exit, exposes graceful and forced
   stop operations, captures the full container ID returned by creation, and
-  pins all later lifecycle operations to that exact container. Protocol/PTY
-  bridging, controlled-session networking, and lifecycle orchestration remain
-  later slices.
+  pins all later lifecycle operations to that exact container. The
+  backend-neutral session I/O bridge is implemented: it dispatches typed
+  controller requests to an injected lifecycle handler, applies only
+  lifecycle-accepted input and resize effects, forwards exact ordered PTY
+  output as protocol events, gives lifecycle events a separate prioritized
+  bounded write admission path, and reports request, backpressure, and
+  disconnect failures without owning the channel or containers. A failed event
+  write makes the framed transport terminal so a later event cannot be appended
+  to a potentially partial frame. Full lifecycle orchestration and
+  controlled-session networking remain later slices.
 - Initial runtime: Linux containers under Docker
 - Motivating clients: OmegaFlow recording, sandboxed AI agents, security
   inspection, and untrusted-code execution
@@ -543,11 +550,16 @@ If every final byte is delivered and every output surface reaches EOF before
 the deadline, Host Reploy emits `workload_outputs_finalized(drained)` only after
 all earlier output has been consumed through its flow-control window. If the
 deadline expires, or a runtime error makes complete delivery unverifiable,
-Host Reploy forcibly closes the remaining surfaces and emits
-`workload_outputs_finalized(failed, reason)`. The multiplexing layer guarantees
-that no output frame can follow either outcome. Failure is explicit and cannot
-be converted into successful completion by `complete` or terminal
-acknowledgement.
+Host Reploy forcibly closes the remaining surfaces, records the failed outcome,
+and emits `workload_outputs_finalized(failed, reason)` while the controller
+transport remains intact. Any event-frame write failure is terminal: the frame
+header or payload may have been written partially, so Host Reploy closes the
+connection and never appends another event. That failure latches
+`controller_lost`; the output-finalization failure remains in the invoking host
+operation's result even though the damaged channel cannot carry it. The
+multiplexing layer guarantees that no output frame can follow a successfully
+emitted finalization outcome. Failure cannot be converted into successful
+completion by `complete` or terminal acknowledgement.
 
 The barrier initially covers the PTY. A future workload output-file or
 output-directory contract joins the same barrier after its files are closed,

internal/controlledsession/channel.go

@@ -76,6 +76,8 @@ type PrivateChannelV1 struct {
 
 // ControllerConnectionV1 is the sole claimed controller connection. Reads and
 // writes each admit one frame at a time, so no unbounded in-memory queue forms.
+// A failed event write closes the connection because it may have left a partial
+// frame on the stream.
 type ControllerConnectionV1 struct {
 	connection net.Conn
 	readMu     sync.Mutex
@@ -234,10 +236,14 @@ func (connection *ControllerConnectionV1) WriteEvent(ctx context.Context, event
 		return WriteEventV1(connection.connection, event)
 	})
 	if err != nil {
+		closeErr := connection.Close()
 		if isControllerDisconnectV1(err) {
-			return fmt.Errorf("%w while writing event: %v", ErrControllerDisconnectedV1, err)
+			return errors.Join(
+				fmt.Errorf("%w while writing event: %v", ErrControllerDisconnectedV1, err),
+				closeErr,
+			)
 		}
-		return fmt.Errorf("write controlled-session event: %w", err)
+		return errors.Join(fmt.Errorf("write controlled-session event: %w", err), closeErr)
 	}
 	return nil
 }

internal/controlledsession/channel_linux_test.go

@@ -7,6 +7,7 @@ import (
 	"context"
 	"encoding/binary"
 	"errors"
+	"io"
 	"net"
 	"os"
 	"path/filepath"
@@ -250,11 +251,27 @@ func TestControllerConnectionV1BoundsAndSerializesFlow(t *testing.T) {
 		server, client := net.Pipe()
 		defer client.Close()
 		connection := &ControllerConnectionV1{connection: server}
-		ctx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond)
+		ctx, cancel := context.WithTimeout(context.Background(), 250*time.Millisecond)
 		defer cancel()
 		event := EventV1{Kind: EventOutputV1, Bytes: bytes.Repeat([]byte{'x'}, MaxFramePayloadV1)}
-		if err := connection.WriteEvent(ctx, event); !errors.Is(err, context.DeadlineExceeded) {
-			t.Fatalf("WriteEvent() error = %v", err)
+		writeErr := make(chan error, 1)
+		go func() { writeErr <- connection.WriteEvent(ctx, event) }()
+		header := make([]byte, frameHeaderSizeV1)
+		if _, err := io.ReadFull(client, header); err != nil {
+			t.Fatal(err)
+		}
+		if err := <-writeErr; !errors.Is(err, context.DeadlineExceeded) {
+			t.Fatalf("first WriteEvent() error = %v", err)
+		}
+		if err := connection.WriteEvent(t.Context(), EventV1{Kind: EventOutputV1, Bytes: []byte("later")}); err == nil {
+			t.Fatal("second WriteEvent() reused the connection after a partial frame")
+		}
+		remaining, err := io.ReadAll(client)
+		if err != nil {
+			t.Fatal(err)
+		}
+		if len(remaining) != 0 {
+			t.Fatalf("partial payload bytes = %d, want 0", len(remaining))
 		}
 	})