feat: complete Windows client control and recovery paths

internal/client/spool/events.go

@@ -267,14 +267,25 @@ func (store *Store) AssignSendWindow(ctx context.Context, issueUUID domain.UUID,
 		return nil, err
 	}
 	defer tx.Rollback()
-	var next uint64
-	err = tx.QueryRowContext(ctx, `SELECT next_event_seq FROM commands WHERE issue_uuid = ?`, issueUUID[:]).Scan(&next)
+	var next, lastAck, assignedBytes uint64
+	err = tx.QueryRowContext(ctx, `SELECT next_event_seq, last_server_ack,
+COALESCE((SELECT sum(length(payload)) FROM events
+  WHERE issue_uuid = commands.issue_uuid AND event_seq IS NOT NULL AND event_seq > commands.last_server_ack), 0)
+FROM commands WHERE issue_uuid = ?`, issueUUID[:]).Scan(&next, &lastAck, &assignedBytes)
 	if err == sql.ErrNoRows {
 		return nil, ErrUnknownCommand
 	}
 	if err != nil {
 		return nil, err
 	}
+	// Assigned rows remain pinned until cumulative acknowledgement.  Never
+	// assign another row while that durable send window is full; otherwise a
+	// burst of local events could grow the pinned set without bound even though
+	// callers pass a per-command byte limit on every invocation.
+	if assignedBytes >= maximumStoredBytes {
+		return eventsBySequence(ctx, tx, issueUUID, 0)
+	}
+	maximumStoredBytes -= assignedBytes
 	rows, err := tx.QueryContext(ctx, `SELECT local_ordinal, length(payload) FROM events WHERE issue_uuid = ? AND event_seq IS NULL ORDER BY local_ordinal LIMIT ?`, issueUUID[:], maximumEvents)
 	if err != nil {
 		return nil, err
@@ -475,6 +486,14 @@ func (store *Store) RecoverLaunchUncertainty(ctx context.Context, now time.Time)
 // crash between a terminal marker and its public event from creating a state
 // that can be replayed as a second execution.
 func (store *Store) AppendLifecycle(ctx context.Context, issueUUID domain.UUID, phase uint32, revision uint64, detail string, observedAt time.Time) (Event, error) {
+	return store.AppendLifecycleWithIdentity(ctx, issueUUID, phase, revision, detail, observedAt, nil)
+}
+
+// AppendLifecycleWithIdentity is the lifecycle boundary used by the client
+// runtime when the Windows supervisor has captured an immutable selection
+// record.  Keeping the legacy wrapper above preserves the small store API for
+// recovery and tests that have no platform identity to attach.
+func (store *Store) AppendLifecycleWithIdentity(ctx context.Context, issueUUID domain.UUID, phase uint32, revision uint64, detail string, observedAt time.Time, identity *rvboxv1.WindowsExecutionIdentity) (Event, error) {
 	if !validUUID(issueUUID) || phase == 0 || phase > 11 || observedAt.IsZero() || revision == 0 {
 		return Event{}, errors.New("invalid lifecycle event")
 	}
@@ -503,7 +522,7 @@ func (store *Store) AppendLifecycle(ctx context.Context, issueUUID domain.UUID,
 	if !domain.CanTransition(rvboxv1.CommandLifecycle(current), rvboxv1.CommandLifecycle(phase)) {
 		return Event{}, fmt.Errorf("invalid lifecycle transition %s -> %s", rvboxv1.CommandLifecycle(current), rvboxv1.CommandLifecycle(phase))
 	}
-	lifecycle := &rvboxv1.LifecycleChange{Lifecycle: rvboxv1.CommandLifecycle(phase), CommandRevision: revision, Detail: detail}
+	lifecycle := &rvboxv1.LifecycleChange{Lifecycle: rvboxv1.CommandLifecycle(phase), CommandRevision: revision, Detail: detail, WindowsExecutionIdentity: identity}
 	payload, err := proto.MarshalOptions{Deterministic: true}.Marshal(&rvboxv1.CommandEvent{IssueUuid: issueUUID.String(), ObservedAt: timestamppb.New(observedAt), Payload: &rvboxv1.CommandEvent_Lifecycle{Lifecycle: lifecycle}})
 	if err != nil {
 		return Event{}, err

internal/client/spool/spool_test.go

@@ -13,6 +13,7 @@ import (
 	rvboxv1 "github.com/rvbox/rvbox/gen/go/rvbox/v1"
 
 	"github.com/rvbox/rvbox/internal/domain"
+	"google.golang.org/protobuf/proto"
 )
 
 func TestExecutionSpecIsDurableAndQuotaCounted_HP_DISPATCH_07(t *testing.T) {
@@ -188,6 +189,38 @@ func TestSpoolAcceptanceSequencingAndAck_HP_CLIENT_07(t *testing.T) {
 	}
 }
 
+func TestSendWindowPinsUnacknowledgedBytes_BH_CLIENT_13(t *testing.T) {
+	t.Parallel()
+	ctx := context.Background()
+	store := openTestStore(t, ctx, filepath.Join(t.TempDir(), "spool"), DefaultTombstoneLimit)
+	issue := testUUID(t, "019c46f1-1d02-7000-8000-000000000014")
+	command := testCommand(issue, []byte("send-window"))
+	now := time.Date(2026, time.September, 6, 12, 0, 0, 0, time.UTC)
+	if _, err := store.AcceptCommand(ctx, command, now); err != nil {
+		t.Fatal(err)
+	}
+	for index, payload := range [][]byte{[]byte("first"), []byte("second")} {
+		if _, err := store.AppendEvent(ctx, issue, EventInput{Kind: uint32(index + 1), Compression: 1, Payload: payload, CreatedAt: now.Add(time.Duration(index) * time.Second)}); err != nil {
+			t.Fatal(err)
+		}
+	}
+	assigned, err := store.AssignSendWindow(ctx, issue, 1, uint64(len("first")))
+	if err != nil || len(assigned) != 1 || assigned[0].EventSeq != 1 {
+		t.Fatalf("first bounded window = %#v, %v", assigned, err)
+	}
+	assigned, err = store.AssignSendWindow(ctx, issue, 1, uint64(len("first")))
+	if err != nil || len(assigned) != 1 || assigned[0].EventSeq != 1 {
+		t.Fatalf("full pinned window assigned more data = %#v, %v", assigned, err)
+	}
+	if err := store.Ack(ctx, issue, 1); err != nil {
+		t.Fatal(err)
+	}
+	assigned, err = store.AssignSendWindow(ctx, issue, 1, uint64(len("second")))
+	if err != nil || len(assigned) != 1 || assigned[0].EventSeq != 2 {
+		t.Fatalf("window after cumulative ack = %#v, %v", assigned, err)
+	}
+}
+
 func TestAppendLifecycleAtomicallyUpdatesPhaseAndEvent_HP_CLIENT_11(t *testing.T) {
 	t.Parallel()
 	ctx := context.Background()
@@ -199,7 +232,9 @@ func TestAppendLifecycleAtomicallyUpdatesPhaseAndEvent_HP_CLIENT_11(t *testing.T
 	if _, err := store.AcceptCommand(ctx, command, now); err != nil {
 		t.Fatal(err)
 	}
-	if _, err := store.AppendLifecycle(ctx, issue, uint32(rvboxv1.CommandLifecycle_COMMAND_RUNNING), 1, "launch authorized", now); err != nil {
+	effective := rvboxv1.WindowsExecutionContext_WINDOWS_EXECUTION_CONTEXT_ACTIVE_USER
+	identity := &rvboxv1.WindowsExecutionIdentity{EffectiveContext: &effective, SessionId: ptrUint32(1), SessionUserSid: "S-1-5-21-user", EffectiveUserSid: "S-1-5-21-user", AttemptedContexts: []rvboxv1.WindowsExecutionContext{effective}, SelectionDetail: "selected"}
+	if _, err := store.AppendLifecycleWithIdentity(ctx, issue, uint32(rvboxv1.CommandLifecycle_COMMAND_RUNNING), 1, "launch authorized", now, identity); err != nil {
 		t.Fatal(err)
 	}
 	if _, err := store.AppendLifecycle(ctx, issue, uint32(rvboxv1.CommandLifecycle_COMMAND_SUCCEEDED), 1, "exit 0", now.Add(time.Second)); err != nil {
@@ -223,8 +258,14 @@ func TestAppendLifecycleAtomicallyUpdatesPhaseAndEvent_HP_CLIENT_11(t *testing.T
 	if err != nil || len(events) != 2 {
 		t.Fatalf("lifecycle events = %#v, %v", events, err)
 	}
+	var decoded rvboxv1.CommandEvent
+	if err := proto.Unmarshal(events[0].Payload, &decoded); err != nil || decoded.GetLifecycle().GetWindowsExecutionIdentity().GetEffectiveContext() != effective {
+		t.Fatalf("running identity event = %v, %v", decoded.GetLifecycle(), err)
+	}
 }
 
+func ptrUint32(value uint32) *uint32 { return &value }
+
 func TestTerminalCleanupTombstonesAndConflicts_BH_CLIENT_02(t *testing.T) {
 	t.Parallel()
 	ctx := context.Background()