B3.4.1: Wire qualified Awsh private command and return exchange

runtime/envoy/awsh/launch/completion_linux.go

@@ -36,7 +36,7 @@ func (s *Session) BeginCompletion(ctx context.Context) error {
 	s.startMu.Lock()
 	defer s.startMu.Unlock()
 	deadline, ok := ctx.Deadline()
-	if !ok || ctx.Err() != nil || s.startFailed || s.active == nil || s.active.phase != startReleased || s.completion != nil {
+	if (!ok && (!s.parentControlled || ctx.Done() == nil)) || ctx.Err() != nil || s.startFailed || s.active == nil || s.active.phase != startReleased || s.completion != nil {
 		return s.failStart(fmt.Errorf("completion outside released operation or cleanup epoch"))
 	}
 	r := &completionRecord{ctx: ctx, pidfd: -1, done: make(chan struct{})}
@@ -79,6 +79,7 @@ func (s *Session) BeginCompletion(ctx context.Context) error {
 	if err = s.completionHealthy(); err != nil {
 		return s.failStart(err)
 	}
+	s.armAdmission("input_closed", s.active.request.OperationID)
 	if err = s.startWrite(protocol.PrivateInputClose{OperationID: s.active.request.OperationID, CompletionHelperPID: int64(r.pid)}); err != nil {
 		return s.failStart(err)
 	}
@@ -215,6 +216,7 @@ func (s *Session) FinishCompletion() error {
 	if err = s.completionHealthy(); err != nil {
 		return s.failStart(err)
 	}
+	s.armAdmission("idle", "")
 	if err = s.startWrite(protocol.PrivateCompleted{OperationID: s.active.request.OperationID, Status: state.Status, PhysicalCWD: state.PhysicalCWD, Inspections: plans}); err != nil {
 		return s.failStart(err)
 	}

runtime/envoy/awsh/launch/dispatch_linux.go

@@ -0,0 +1,55 @@
+//go:build linux
+
+package launch
+
+import (
+	"os"
+	"syscall"
+
+	"github.com/omry/omegaflow/runtime/envoy/protocol"
+)
+
+// This is a one-event admission projection of the locked Session phase. Publish
+// before a result/helper write can elicit a reply. The sole dispatcher consumes
+// it once, joins that write, then invokes the existing locked phase handler.
+type dispatchAdmission struct{ kind, operation string }
+
+func (s *Session) armAdmission(kind, operation string) {
+	if s.parentControlled {
+		s.admission.Store(&dispatchAdmission{kind, operation})
+	}
+}
+
+// AdmitControl refuses premature, duplicate and wrong-operation controls even
+// while the current Session method is blocked in I/O with startMu held.
+func (s *Session) AdmitControl(message protocol.PrivateMessage) bool {
+	kind, operation := "", ""
+	switch m := message.(type) {
+	case *protocol.PrivateExecute, *protocol.PrivateShutdown:
+		kind = "idle"
+	case *protocol.PrivateStartRelease:
+		kind, operation = "start_release", m.OperationID
+	case *protocol.PrivateStartedAck:
+		kind, operation = "started_ack", m.OperationID
+	case *protocol.PrivateInputClosed:
+		kind, operation = "input_closed", m.OperationID
+	default:
+		return false
+	}
+	r := s.admission.Load()
+	return r != nil && r.kind == kind && r.operation == operation && s.admission.CompareAndSwap(r, nil)
+}
+
+// AdmitSignal preserves the same one-event rule for the reserved release signal.
+func (s *Session) AdmitSignal(signal os.Signal) bool {
+	r := s.admission.Load()
+	return signal == syscall.SIGUSR1 && r != nil && r.kind == "signal" && s.admission.CompareAndSwap(r, nil)
+}
+
+// HasActiveOperation distinguishes recoverable source rejection after BeginStart
+// returns. Call only after its I/O job joins, before starting the next method.
+func (s *Session) HasActiveOperation() bool {
+	s.startMu.Lock()
+	defer s.startMu.Unlock()
+	return s.active != nil
+}

runtime/envoy/awsh/launch/dispatch_linux_test.go

@@ -0,0 +1,54 @@
+//go:build linux
+
+package launch
+
+import (
+	"sync"
+	"sync/atomic"
+	"syscall"
+	"testing"
+
+	"github.com/omry/omegaflow/runtime/envoy/protocol"
+)
+
+func TestDispatchAdmissionRejectsEarlyAndDuplicateControls(t *testing.T) {
+	s := &Session{parentControlled: true}
+	if s.AdmitControl(&protocol.PrivateInputClosed{OperationID: "op"}) || s.AdmitSignal(syscall.SIGUSR1) {
+		t.Fatal("unarmed event admitted")
+	}
+	s.armAdmission("input_closed", "op")
+	if s.AdmitControl(&protocol.PrivateContinue{OperationID: "op", GateID: "later-owner"}) {
+		t.Fatal("unsupported lifecycle control acquired admission")
+	}
+	if s.AdmitControl(&protocol.PrivateInputClosed{OperationID: "wrong"}) || s.AdmitControl(&protocol.PrivateStartedAck{OperationID: "op"}) {
+		t.Fatal("wrong identity or phase consumed admission")
+	}
+	var winners atomic.Int32
+	var join sync.WaitGroup
+	for i := 0; i < 32; i++ {
+		join.Add(1)
+		go func() {
+			defer join.Done()
+			if s.AdmitControl(&protocol.PrivateInputClosed{OperationID: "op"}) {
+				winners.Add(1)
+			}
+		}()
+	}
+	join.Wait()
+	if winners.Load() != 1 {
+		t.Fatal("one phase authorized more than one control", winners.Load())
+	}
+	s.armAdmission("signal", "op")
+	if s.AdmitSignal(syscall.SIGUSR2) || !s.AdmitSignal(syscall.SIGUSR1) || s.AdmitSignal(syscall.SIGUSR1) {
+		t.Fatal("reserved signal admission is not exact and single-use")
+	}
+	s.armAdmission("idle", "")
+	if !s.AdmitControl(&protocol.PrivateExecute{}) || s.AdmitControl(&protocol.PrivateShutdown{}) {
+		t.Fatal("idle phase admitted concurrent execute/shutdown")
+	}
+	standalone := &Session{}
+	standalone.armAdmission("idle", "")
+	if standalone.AdmitControl(&protocol.PrivateExecute{}) {
+		t.Fatal("standalone caller acquired controlled dispatch authority")
+	}
+}

runtime/envoy/awsh/launch/operation_linux.go

@@ -46,6 +46,10 @@ type startRecord struct {
 // Check uses the same exact qualified candidate contract as selected-shell launch.
 func (s *Session) BeginStart(ctx context.Context, m protocol.PrivateExecute, expected string, target bashbuild.Target, inputs map[string]string) error {
 	return s.beginStart(ctx, m, func(ctx context.Context, source protocol.HelperSourceReply) error {
+		if s.parentControlled {
+			_, err := submission.CheckControlled(ctx, source, expected, target, inputs)
+			return err
+		}
 		_, err := submission.Check(ctx, source, expected, target, inputs)
 		return err
 	})
@@ -61,7 +65,7 @@ func (s *Session) beginStart(ctx context.Context, m protocol.PrivateExecute, che
 	ready := s.ready && !s.closing
 	s.mu.Unlock()
 	deadline, ok := ctx.Deadline()
-	if !ready || !ok || ctx.Err() != nil {
+	if !ready || (!ok && (!s.parentControlled || ctx.Done() == nil)) || ctx.Err() != nil {
 		return s.failStart(fmt.Errorf("start requires ready shell and original deadline"))
 	}
 	if _, err := protocol.EncodePrivate(m, protocol.EnvoyToAwsh); err != nil {
@@ -94,6 +98,7 @@ func (s *Session) beginStart(ctx context.Context, m protocol.PrivateExecute, che
 	if err := check(ctx, source); err != nil {
 		var rejection *submission.Rejection
 		if ctx.Err() == nil && errors.As(err, &rejection) {
+			s.armAdmission("idle", "")
 			err = s.startWrite(protocol.PrivateRejected{OperationID: m.OperationID, Code: rejection.Code, Message: rejection.Message})
 			if err = finishCancellation(ctx, stop, done, err); err == nil {
 				return nil // recoverable before any reservation or trigger
@@ -195,6 +200,7 @@ func (s *Session) AcceptStartHelper() error {
 		return nil
 	}
 	r.prepared, r.preparedUnregister, r.phase, keep = c, unregister, startPrepared, true
+	s.armAdmission("start_release", r.request.OperationID)
 	if err = s.startWrite(protocol.PrivateStartPrepared{OperationID: r.request.OperationID}); err != nil {
 		return s.failStart(err)
 	}
@@ -216,6 +222,7 @@ func (s *Session) HandleStartControl(m protocol.PrivateMessage) error {
 		if r.phase != startPrepared || v.OperationID != r.request.OperationID {
 			return s.failStart(fmt.Errorf("unexpected start_release"))
 		}
+		s.armAdmission("started_ack", r.request.OperationID)
 		if err := s.startWrite(protocol.PrivateStarted{OperationID: v.OperationID}); err != nil {
 			return s.failStart(err)
 		}
@@ -231,6 +238,7 @@ func (s *Session) HandleStartControl(m protocol.PrivateMessage) error {
 			return s.failStart(err)
 		}
 		r.phase = startSignal // arm BEFORE the helper success can release Bash
+		s.armAdmission("signal", r.request.OperationID)
 		if err := helper.Reply(r.prepared, protocol.HelperStartPrepared, protocol.HelperAccepted{}); err != nil {
 			if r.preparedUnregister != nil {
 				r.preparedUnregister()
@@ -303,6 +311,7 @@ func (s *Session) startWrite(m protocol.PrivateMessage) error {
 }
 
 func (s *Session) failStart(err error) error {
+	s.admission.Store(nil)
 	s.startFailed = true
 	if s.active != nil && s.active.preparedUnregister != nil {
 		if s.active.prepared != nil {

runtime/envoy/awsh/launch/start_linux.go

@@ -10,6 +10,7 @@ import (
 	"strconv"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"syscall"
 	"time"
 	"unsafe"
@@ -45,6 +46,7 @@ type Session struct {
 	startFailed         bool
 	completion          *completionRecord
 	parentControlled    bool // supervisor owns Handoff.Control's os.File lifetime
+	admission           atomic.Pointer[dispatchAdmission]
 }
 
 // Start admits only a qualified, exact build. No executable supervise command
@@ -53,19 +55,19 @@ func Start(ctx context.Context, h Handoff, expected string, target bashbuild.Tar
 	if _, ok := ctx.Deadline(); !ok {
 		return nil, fmt.Errorf("launch requires the existing phase deadline")
 	}
-	return startQualified(ctx, h, expected, target, inputs, false)
+	return startQualified(ctx, h, expected, target, inputs, false, nil)
 }
 
 // StartControlled uses cancellation delivered by the existing Envoy control
 // channel. Envoy enforces the launch epoch; Awsh creates no lifecycle clock.
-func StartControlled(ctx context.Context, h Handoff, expected string, target bashbuild.Target, inputs map[string]string) (*Session, error) {
+func StartControlled(ctx context.Context, h Handoff, expected string, target bashbuild.Target, inputs map[string]string, publish func(*Session)) (*Session, error) {
 	if ctx.Done() == nil {
 		return nil, fmt.Errorf("controlled launch requires parent cancellation")
 	}
-	return startQualified(ctx, h, expected, target, inputs, true)
+	return startQualified(ctx, h, expected, target, inputs, true, publish)
 }
 
-func startQualified(ctx context.Context, h Handoff, expected string, target bashbuild.Target, inputs map[string]string, controlled bool) (s *Session, err error) {
+func startQualified(ctx context.Context, h Handoff, expected string, target bashbuild.Target, inputs map[string]string, controlled bool, publish func(*Session)) (s *Session, err error) {
 	owned := false
 	defer func() {
 		if controlled && err != nil && !owned {
@@ -95,7 +97,7 @@ func startQualified(ctx context.Context, h Handoff, expected string, target bash
 		}
 	}
 	owned = true // startOwned assumes result ownership, including its failures.
-	return startOwned(ctx, h, signals, controlled)
+	return startOwned(ctx, h, signals, controlled, publish)
 }
 
 // ReportLaunchFailure consumes a usable inherited result pipe only before a
@@ -127,10 +129,10 @@ func writeLaunchFailure(ctx context.Context, result helper.DeadlineWriter) {
 }
 
 func start(ctx context.Context, h Handoff, signals []uint64) (s *Session, err error) {
-	return startOwned(ctx, h, signals, false)
+	return startOwned(ctx, h, signals, false, nil)
 }
 
-func startOwned(ctx context.Context, h Handoff, signals []uint64, controlled bool) (s *Session, err error) {
+func startOwned(ctx context.Context, h Handoff, signals []uint64, controlled bool, publish func(*Session)) (s *Session, err error) {
 	deadline, ok := ctx.Deadline()
 	if !ok && !controlled {
 		return nil, fmt.Errorf("launch requires the existing phase deadline")
@@ -142,6 +144,11 @@ func startOwned(ctx context.Context, h Handoff, signals []uint64, controlled boo
 		return nil, err
 	}
 	s = &Session{Handoff: h, parentControlled: controlled}
+	if publish != nil {
+		// The decoder can observe this Session during startup, while its
+		// admission is still empty. Only atomic admission is read there.
+		publish(s)
+	}
 	// NewFile registers a nonblocking pipe with Go's poller, making the original
 	// deadline and context closure effective even when Envoy stops reading.
 	if err = syscall.SetNonblock(h.Result, true); err != nil {
@@ -253,6 +260,7 @@ func startOwned(ctx context.Context, h Handoff, signals []uint64, controlled boo
 		err = e
 		return
 	}
+	s.armAdmission("idle", "")
 	err = helper.WriteControl(ctx, s.result, frame)
 	if err == nil {
 		err = ctx.Err()
@@ -425,7 +433,7 @@ func (s *Session) startup(ctx context.Context) error {
 // active state. The start owner supplies its existing phase deadline and must
 // withhold helper success if restoration fails; that integration is separate.
 func (s *Session) RestoreWorkloadTermios(ctx context.Context) error {
-	if _, ok := ctx.Deadline(); !ok {
+	if _, ok := ctx.Deadline(); !ok && (!s.parentControlled || ctx.Done() == nil) {
 		return fmt.Errorf("restoration requires the existing phase deadline")
 	}
 	s.mu.Lock()

runtime/envoy/awsh/submission/check_linux.go

@@ -25,6 +25,18 @@ func Check(ctx context.Context, m protocol.HelperSourceReply, expected string, t
 	})
 }
 
+// CheckControlled uses cancellation from the owning Envoy epoch. It creates no
+// replacement clock and retains fresh qualification verification before forks.
+func CheckControlled(ctx context.Context, m protocol.HelperSourceReply, expected string, target bashbuild.Target, inputs map[string]string) (string, error) {
+	if ctx.Done() == nil {
+		return "", fmt.Errorf("controlled checker requires parent cancellation")
+	}
+	return checkSource(ctx, m, func() error {
+		_, err := bashbuild.LookupAwsh("/bin/bash", expected, target, inputs)
+		return err
+	})
+}
+
 // parseOutput remembers presence, never retains unbounded checker diagnostics.
 // Each command stream has its own writer; Run joins its copying goroutines.
 type parseOutput struct{ present bool }
@@ -38,6 +50,10 @@ func check(ctx context.Context, m protocol.HelperSourceReply, verify func() erro
 	if _, ok := ctx.Deadline(); !ok {
 		return "", fmt.Errorf("source checker requires the existing start deadline")
 	}
+	return checkSource(ctx, m, verify)
+}
+
+func checkSource(ctx context.Context, m protocol.HelperSourceReply, verify func() error) (string, error) {
 	frame, err := checkedFrame(m)
 	if err != nil {
 		return "", err

runtime/envoy/awsh/submission/check_linux_test.go

@@ -38,6 +38,23 @@ func TestMain(m *testing.M) {
 	os.Exit(m.Run())
 }
 
+func TestControlledCheckerRequiresCancellationAndRetainsStandaloneGuard(t *testing.T) {
+	m := example(":")
+	if _, err := CheckControlled(context.Background(), m, "unused", bashbuild.Target{}, nil); err == nil {
+		t.Fatal("controlled checker accepted an uninterruptible context")
+	}
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+	if _, err := CheckControlled(ctx, m, "unused", bashbuild.Target{}, nil); !errors.Is(err, context.Canceled) {
+		t.Fatal("cancelled controlled checker reached qualification/fork", err)
+	}
+	ctx, cancel = context.WithCancel(context.Background())
+	defer cancel()
+	if _, err := Check(ctx, m, "unused", bashbuild.Target{}, nil); err == nil || !strings.Contains(err.Error(), "deadline") {
+		t.Fatal("controlled entry relaxed standalone deadline guard", err)
+	}
+}
+
 func TestCheckFailsClosedWithoutQualification(t *testing.T) {
 	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
 	defer cancel()

runtime/envoy/awsh/supervise/control_linux_test.go

@@ -0,0 +1,34 @@
+//go:build linux
+
+package supervise
+
+import (
+	"bytes"
+	"context"
+	"testing"
+
+	"github.com/omry/omegaflow/runtime/envoy/protocol"
+)
+
+func TestReadControlsAdmitsBeforeQueue(t *testing.T) {
+	frame, err := protocol.EncodePrivate(protocol.PrivateShutdown{}, protocol.EnvoyToAwsh)
+	if err != nil {
+		t.Fatal(err)
+	}
+	for _, available := range []int{0, 1} {
+		t.Run(string(rune('0'+available)), func(t *testing.T) {
+			events := make(chan protocol.PrivateMessage, 2)
+			remaining := available
+			err := readControls(context.Background(), bytes.NewReader(append(append([]byte{}, frame...), frame...)), events, func(protocol.PrivateMessage) bool {
+				if remaining == 0 {
+					return false
+				}
+				remaining--
+				return true
+			})
+			if err == nil || err.Error() != "private request outside current phase" || len(events) != available {
+				t.Fatal("premature or duplicate control entered queue", len(events), err)
+			}
+		})
+	}
+}