feat(g01): observe bounded idle drain ordering

experiments/g01-scaleset/cmd/g01-live/main.go

@@ -89,7 +89,7 @@ func runWithPreparation(args []string, in io.Reader, out io.Writer, revisionForB
 		if *approvalPath != "" || *statePath != "" || *phase != "" || workerInputs || *pairedBinding != "" {
 			return reject()
 		}
-		fmt.Fprintln(out, "Controller-only phases: create, before-ack, after-ack, before-acquire, acquire-loss, jit-loss, inspect, cleanup. Paired terminal mode uses one same-process executable with explicit worker approval/state inputs and fixed terminal sequencing; no worker launch or workflow dispatch. Live execution requires an immutable reviewed build, exact private approval and controller-side broker input.")
+		fmt.Fprintln(out, "Controller-only phases: create, before-ack, after-ack, before-acquire, acquire-loss, jit-loss, drain, inspect, cleanup. Paired terminal mode uses one same-process executable with explicit worker approval/state inputs and fixed terminal sequencing; no worker launch or workflow dispatch. Live execution requires an immutable reviewed build, exact private approval and controller-side broker input.")
 		return 0
 	}
 	modeCount := 0

experiments/g01-scaleset/livecanary/approval.go

@@ -12,7 +12,7 @@ var nonce = regexp.MustCompile(`^[a-f0-9]{32}$`)
 var sha = regexp.MustCompile(`^[a-f0-9]{40}$`)
 var actionsHost = regexp.MustCompile(`^[a-z0-9-]+(?:\.[a-z0-9-]+)*\.actions\.githubusercontent\.com$`)
 var workflowPath = regexp.MustCompile(`^\.github/workflows/[a-zA-Z0-9_-]+\.ya?ml$`)
-var phases = []string{"create", "before-ack", "after-ack", "before-acquire", "acquire-loss", "jit-loss", "inspect", "cleanup"}
+var phases = []string{"create", "before-ack", "after-ack", "before-acquire", "acquire-loss", "jit-loss", "drain", "inspect", "cleanup"}
 
 func (a Approval) setName() string    { return "g01-" + a.OwnerNonce }
 func (a Approval) workerName() string { return a.setName() + "-worker-1" }
@@ -24,7 +24,7 @@ func (a Approval) Validate(now time.Time) error {
 	if a.Organization == ".." {
 		return ErrApproval
 	}
-	if !component.MatchString(a.Organization) || !component.MatchString(a.Repository) || !component.MatchString(a.Controller) || a.Organization == "." || a.Repository == "." || a.Repository == ".." || a.RepositoryID <= 0 || a.RunnerGroupID <= 0 || !nonce.MatchString(a.OwnerNonce) || !sha.MatchString(a.HarnessSHA) || !sha.MatchString(a.WorkflowSHA) || !workflowPath.MatchString(a.WorkflowPath) || !a.ExpiresAt.After(now) || a.ExpiresAt.After(now.Add(24*time.Hour)) || len(a.ActionsHosts) == 0 || len(a.ActionsHosts) > 8 || len(a.Phases) == 0 {
+	if !component.MatchString(a.Organization) || !component.MatchString(a.Repository) || !component.MatchString(a.Controller) || a.Organization == "." || a.Repository == "." || a.Repository == ".." || a.RepositoryID <= 0 || a.RunnerGroupID <= 0 || !nonce.MatchString(a.OwnerNonce) || !sha.MatchString(a.HarnessSHA) || !sha.MatchString(a.WorkflowSHA) || !workflowPath.MatchString(a.WorkflowPath) || !a.ExpiresAt.After(now) || a.ExpiresAt.After(now.Add(24*time.Hour)) || len(a.ActionsHosts) == 0 || len(a.ActionsHosts) > 8 || len(a.Phases) == 0 || len(a.Phases) > 9 {
 		return ErrApproval
 	}
 	seen := map[string]bool{}

experiments/g01-scaleset/livecanary/baseline_execution.go

@@ -20,7 +20,17 @@ func (s *pairedBaselineScope) afterAcquire(ctx context.Context, acquisition base
 	if err != nil {
 		return err
 	}
-	capture := &baselineWireCapture{stage: "jit", setID: s.setID}
+	if s.listener == nil {
+		return ErrQuarantine
+	}
+	// AcquireJobs invokes this continuation while the listener mutex is held.
+	// wire reads the already-captured session tuple directly for this handoff;
+	// taking the listener accessors here would recursively lock that mutex.
+	capture := s.listener.wire("jit")
+	capture.runnerName = s.approval.workerName()
+	if capture.origin == "" || !capture.runtimePathPrefixSet || capture.runtimePathPrefix == "" || capture.runnerName == "" {
+		return ErrQuarantine
+	}
 	request, cancel := context.WithTimeout(ctx, operationTimeout)
 	got, callErr := s.captured.GenerateJIT(capture.context(request), s.setID, s.approval.workerName())
 	cancel()

experiments/g01-scaleset/livecanary/baseline_journal.go

@@ -116,6 +116,23 @@ func cloneEvent(e Event) Event {
 		e.Authority = &x
 	}
 	e.RequestIDs = slices.Clone(e.RequestIDs)
+	if e.Drain != nil {
+		data, _ := json.Marshal(e.Drain)
+		var x drainObservation
+		_ = json.Unmarshal(data, &x)
+		x.Ordering = slices.Clone(e.Drain.Ordering)
+		e.Drain = &x
+	}
+	if e.DrainSnapshot != nil {
+		data, _ := json.Marshal(e.DrainSnapshot)
+		var x drainSnapshot
+		_ = json.Unmarshal(data, &x)
+		if e.DrainSnapshot.Runner != nil {
+			runner := *e.DrainSnapshot.Runner
+			x.Runner = &runner
+		}
+		e.DrainSnapshot = &x
+	}
 	if e.Baseline != nil {
 		data, _ := json.Marshal(e.Baseline)
 		var x baselineRecord

experiments/g01-scaleset/livecanary/baseline_listener.go

@@ -18,23 +18,25 @@ var errBaselineCollected = errors.New("baseline callback collection complete")
 // No current phase/CLI calls this. The future pair orchestrator must prove
 // completed pairing and host preflight before invoking this experiment slice.
 type baselineListener struct {
-	finalizer              *pairedBaselineScope
-	finalizing             bool
-	pairGuard              func() error
-	mu                     sync.Mutex
-	ctx                    context.Context
-	cancel                 context.CancelFunc
-	approval               Approval
-	journal                *FileJournal
-	api                    SDKAPI
-	identity               controllerJournalIdentity
-	creation               controllerRecordRef
-	setID                  int
-	session                Session
-	initial                scaleset.RunnerScaleSetSession
-	sessionID, queue       string
-	running, used, invalid bool
-	after                  func(context.Context, baselineAcquisition) error
+	finalizer                *pairedBaselineScope
+	finalizing               bool
+	pairGuard                func() error
+	mu                       sync.Mutex
+	ctx                      context.Context
+	cancel                   context.CancelFunc
+	approval                 Approval
+	journal                  *FileJournal
+	api                      SDKAPI
+	identity                 controllerJournalIdentity
+	creation                 controllerRecordRef
+	setID                    int
+	session                  Session
+	initial                  scaleset.RunnerScaleSetSession
+	sessionID, queue, origin string
+	runtimePathPrefix        string
+	runtimePathPrefixSet     bool
+	running, used, invalid   bool
+	after                    func(context.Context, baselineAcquisition) error
 }
 
 func newBaselineListenerHeld(ctx context.Context, a Approval, j *FileJournal, api *SDKAPI, setID int) (*baselineListener, error) {
@@ -163,7 +165,25 @@ func (b *baselineListener) finish(r baselineRecord, known bool) (controllerRecor
 	return ref, nil
 }
 func (b *baselineListener) wire(stage string) *baselineWireCapture {
-	return &baselineWireCapture{stage: stage, setID: b.setID, queue: b.queue}
+	return &baselineWireCapture{stage: stage, setID: b.setID, organization: b.approval.Organization, owner: b.approval.setName(), sessionID: b.sessionID, queue: b.queue, origin: b.origin, runtimePathPrefix: b.runtimePathPrefix, runtimePathPrefixSet: b.runtimePathPrefixSet, allowedHosts: baselineWireAllowedHosts(b.approval, b.api.drainEndpointHost())}
+}
+
+func (b *baselineListener) capturedOrigin() string {
+	if b == nil {
+		return ""
+	}
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	return b.origin
+}
+
+func (b *baselineListener) capturedRuntimePathPrefix() (string, bool) {
+	if b == nil {
+		return "", false
+	}
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	return b.runtimePathPrefix, b.runtimePathPrefixSet && !b.invalid
 }
 func (b *baselineListener) run(after func(context.Context, baselineAcquisition) error) error {
 	if b == nil || b.ctx == nil || b.cancel == nil || !b.mu.TryLock() {
@@ -225,7 +245,16 @@ func (b *baselineListener) initialize() error {
 	cancel()
 	r.Set = c.set
 	r.HTTPStatus = c.status
-	known := c.observed() && r.Set.eligible(b.approval, b.setID) && ((callErr != nil && b.ctx.Err() != nil) || (callErr == nil && set != nil && set.ID == r.Set.ID && set.Name == r.Set.Name && set.RunnerGroupID == r.Set.GroupID && set.RunnerSetting.DisableUpdate && r.Set.Statistics.matches(set.Statistics)))
+	beforeOrigin := c.requestOrigin()
+	prefix, prefixKnown := c.requestRuntimePathPrefix()
+	known := c.observed() && prefixKnown && r.Set.eligible(b.approval, b.setID) && ((callErr != nil && b.ctx.Err() != nil) || (callErr == nil && set != nil && set.ID == r.Set.ID && set.Name == r.Set.Name && set.RunnerGroupID == r.Set.GroupID && set.RunnerSetting.DisableUpdate && r.Set.Statistics.matches(set.Statistics)))
+	if prefixKnown {
+		b.runtimePathPrefix = prefix
+		b.runtimePathPrefixSet = true
+	}
+	if c.observed() && beforeOrigin != "" {
+		b.origin = beforeOrigin
+	}
 	if _, err = b.finish(r, known); err != nil {
 		return err
 	}
@@ -238,16 +267,24 @@ func (b *baselineListener) initialize() error {
 	session, callErr := b.api.OpenSession(c.context(ctx), b.setID, b.approval.setName())
 	cancel()
 	sf, _, _, status := c.facts()
+	origin := c.requestOrigin()
+	sessionPrefix, sessionPrefixKnown := c.requestRuntimePathPrefix()
 	r.Session = sf
 	r.HTTPStatus = status
-	known = c.observed() && sf.eligible(b.approval, b.setID) && ((callErr != nil && b.ctx.Err() != nil) || (callErr == nil && session != nil))
+	known = c.observed() && sessionPrefixKnown && sf.eligible(b.approval, b.setID) && ((callErr != nil && b.ctx.Err() != nil) || (callErr == nil && session != nil))
 	var initial scaleset.RunnerScaleSetSession
 	if callErr == nil && session != nil {
 		initial = session.Session()
 	}
 	if known && callErr == nil {
 		known = initial.SessionID.String() == sf.SessionID && initial.OwnerName == sf.Owner && sf.Statistics.matches(initial.Statistics) && initial.MessageQueueURL != ""
 	}
+	if known && (origin == "" || !b.runtimePathPrefixSet || sessionPrefix != b.runtimePathPrefix) {
+		known = false
+	}
+	if known && origin != "" {
+		b.origin = origin
+	}
 	if callErr == nil && session != nil && sf != nil && initial.SessionID.String() == sf.SessionID && initial.OwnerName == b.approval.setName() {
 		b.session = session
 		b.sessionID = sf.SessionID
@@ -388,13 +425,14 @@ func (b *baselineListener) AcquireJobs(_ context.Context, ids []int64) ([]int64,
 		return nil, err
 	}
 	c := b.wire("acquire")
+	c.requestIDs = slices.Clone(ids)
 	ctx, cancel := context.WithTimeout(b.ctx, operationTimeout)
 	got, callErr := b.session.AcquireJobs(c.context(ctx), slices.Clone(ids))
 	cancel()
 	_, _, accepted, status := c.facts()
 	r.Accepted = accepted
 	r.HTTPStatus = status
-	known := c.observed() && status == 200 && accepted != nil && accepted.Count != nil && *accepted.Count == 1 && len(accepted.IDs) == 1 && accepted.IDs[0] == ids[0] && ((callErr == nil && slices.Equal(got, ids)) || (callErr != nil && b.ctx.Err() != nil)) && b.session.Session().SessionID.String() == b.sessionID
+	known := c.observed() && status == 200 && accepted.matches(ids) && ((callErr == nil && slices.Equal(got, ids)) || (callErr != nil && b.ctx.Err() != nil)) && b.session.Session().SessionID.String() == b.sessionID
 	resultRef, err := b.finish(r, known)
 	if err != nil {
 		return nil, err

experiments/g01-scaleset/livecanary/baseline_listener_test.go

@@ -175,7 +175,7 @@ func newBaselineFixtureWithApproval(t *testing.T, change func(string, any) any,
 	t.Cleanup(transport.CloseIdleConnections)
 	outer := transport.Clone()
 	outer.RegisterProtocol("https", baselineRoundTrip(func(r *http.Request) (*http.Response, error) {
-		response, err := (responseBudgetTransport{inner: transport}).RoundTrip(r)
+		response, err := (responseBudgetTransport{inner: baselineRequestCaptureTransport{inner: transport}}).RoundTrip(r)
 		if err == nil && f.afterResponse != nil {
 			f.afterResponse(r, response)
 		}

experiments/g01-scaleset/livecanary/baseline_message.go

@@ -11,8 +11,9 @@ import (
 	"github.com/actions/scaleset"
 )
 
-// Pointer facts retain missing/null separately from explicit zero/false. No
-// queue/acquisition URL, token, arbitrary display text or raw response is kept.
+// Pointer facts retain missing/null separately from explicit zero/false.
+// Persistent facts omit queue/acquisition URLs, tokens, arbitrary display text
+// and raw responses.
 type baselineStatistics struct {
 	Available  *int `json:"total_available_jobs"`
 	Acquired   *int `json:"total_acquired_jobs"`
@@ -55,11 +56,66 @@ type baselineSessionFacts struct {
 	SetID            int                 `json:"set_id"`
 	SetName          string              `json:"set_name"`
 	GroupID          int                 `json:"group_id"`
+	queueURL         string              `json:"-"` // ephemeral drain validation only; never serialized
+	authorization    string              `json:"-"` // ephemeral drain validation only; never serialized
 }
+
+// baselineRunnerFacts retains only the bounded fields needed to compare the
+// pinned SDK's runner lookup result with the exact wire response. The list
+// count is kept separately because the SDK collapses count==0 to nil and
+// otherwise returns only the first value.
+type baselineRunnerFacts struct {
+	Count      int
+	ID         int
+	Name       string
+	ScaleSetID int
+}
+
+func decodeBaselineRunner(data []byte) (*baselineRunnerFacts, error) {
+	var w struct {
+		Count *int `json:"count"`
+		Value *[]struct {
+			ID         *int    `json:"id"`
+			Name       *string `json:"name"`
+			ScaleSetID *int    `json:"runnerScaleSetId"`
+		} `json:"value"`
+	}
+	if !uniqueKeys(json.NewDecoder(strings.NewReader(string(data)))) || json.Unmarshal(data, &w) != nil || w.Count == nil || w.Value == nil || *w.Count < 0 || *w.Count > 1 || *w.Count != len(*w.Value) {
+		return nil, ErrRemote
+	}
+	facts := &baselineRunnerFacts{Count: *w.Count}
+	if facts.Count == 0 {
+		return facts, nil
+	}
+	candidate := (*w.Value)[0]
+	if candidate.ID == nil || *candidate.ID <= 0 || candidate.Name == nil || !baselineText(*candidate.Name, 256) || candidate.ScaleSetID == nil || *candidate.ScaleSetID <= 0 {
+		return nil, ErrRemote
+	}
+	facts.ID = *candidate.ID
+	facts.Name = *candidate.Name
+	facts.ScaleSetID = *candidate.ScaleSetID
+	return facts, nil
+}
+
+func (r *baselineRunnerFacts) matches(v *scaleset.RunnerReference) bool {
+	if r == nil {
+		return false
+	}
+	if r.Count == 0 {
+		return v == nil
+	}
+	return r.Count == 1 && v != nil && r.ID == v.ID && r.Name == v.Name && r.ScaleSetID == v.RunnerScaleSetID
+}
+
 type baselineAccepted struct {
 	Count *int    `json:"count"`
 	IDs   []int64 `json:"ids"`
 }
+
+func (a *baselineAccepted) matches(ids []int64) bool {
+	return a != nil && a.Count != nil && *a.Count == len(ids) && slices.Equal(a.IDs, ids)
+}
+
 type baselineSetFacts struct {
 	ID            int                 `json:"id"`
 	Name          string              `json:"name"`
@@ -95,7 +151,27 @@ func decodeBaselineSet(data []byte) (*baselineSetFacts, error) {
 	return &baselineSetFacts{w.ID, w.Name, w.Group, w.Labels, w.Setting.Disabled, s}, nil
 }
 func (s *baselineSetFacts) eligible(a Approval, id int) bool {
-	if s == nil || s.ID != id || s.Name != a.setName() || s.GroupID != a.RunnerGroupID || s.DisableUpdate == nil || !*s.DisableUpdate || !s.Statistics.eligible(false) {
+	return s.eligibleWithStats(a, id, false)
+}
+
+func (s *baselineSetFacts) eligibleForDrain(a Approval, id int) bool {
+	if !s.matchesOwner(a, id) || s.Statistics == nil {
+		return false
+	}
+	for _, p := range []*int{s.Statistics.Available, s.Statistics.Acquired, s.Statistics.Assigned, s.Statistics.Running, s.Statistics.Registered, s.Statistics.Busy, s.Statistics.Idle} {
+		if p == nil || *p < 0 {
+			return false
+		}
+	}
+	return true
+}
+
+func (s *baselineSetFacts) eligibleWithStats(a Approval, id int, acquired bool) bool {
+	return s.matchesOwner(a, id) && s.Statistics.eligible(acquired)
+}
+
+func (s *baselineSetFacts) matchesOwner(a Approval, id int) bool {
+	if s == nil || s.ID != id || s.Name != a.setName() || s.GroupID != a.RunnerGroupID || s.DisableUpdate == nil || !*s.DisableUpdate {
 		return false
 	}
 	for _, l := range s.Labels {
@@ -105,6 +181,17 @@ func (s *baselineSetFacts) eligible(a Approval, id int) bool {
 	}
 	return false
 }
+
+// matches compares the bounded facts captured by the strict wire reader with
+// the SDK value returned from the same request. A caller may use eligible to
+// establish approved identity, but must also prove that the lossy SDK object
+// did not disagree with those wire facts.
+func (s *baselineSetFacts) matches(v *scaleset.RunnerScaleSet) bool {
+	if s == nil || v == nil || s.ID != v.ID || s.Name != v.Name || s.GroupID != v.RunnerGroupID || s.DisableUpdate == nil || *s.DisableUpdate != v.RunnerSetting.DisableUpdate || !slices.Equal(s.Labels, v.Labels) {
+		return false
+	}
+	return s.Statistics.matches(v.Statistics)
+}
 func baselineText(s string, limit int) bool {
 	if len(s) > limit || !utf8.ValidString(s) {
 		return false
@@ -116,6 +203,14 @@ func baselineText(s string, limit int) bool {
 	}
 	return true
 }
+
+// validDrainAuthorizationToken accepts only a bounded, header-safe opaque
+// token. The value is retained solely in memory to corroborate the pinned SDK
+// session and bind later marked runtime requests.
+func validDrainAuthorizationToken(value string) bool {
+	return value != "" && baselineText(value, 4096) && strings.TrimSpace(value) == value && !strings.ContainsAny(value, "\r\n\x00")
+}
+
 func baselineStats(data json.RawMessage) (*baselineStatistics, error) {
 	if len(data) == 0 || string(data) == "null" {
 		return nil, nil
@@ -157,6 +252,19 @@ func (s *baselineStatistics) matches(sdk *scaleset.RunnerScaleSetStatistic) bool
 	}
 	return true
 }
+
+func (s *baselineStatistics) completeDrain() bool {
+	if s == nil {
+		return false
+	}
+	values := []*int{s.Available, s.Acquired, s.Assigned, s.Running, s.Registered, s.Busy, s.Idle}
+	for _, value := range values {
+		if value == nil || *value < 0 {
+			return false
+		}
+	}
+	return validKnownDrainStatistics(drainStatistics{Available: *s.Available, Acquired: *s.Acquired, Assigned: *s.Assigned, Running: *s.Running, Registered: *s.Registered, Busy: *s.Busy, Idle: *s.Idle})
+}
 func decodeBaselineItem(data []byte, index int) (baselineItem, error) {
 	var w struct {
 		Kind        string          `json:"messageType"`
@@ -264,7 +372,7 @@ func decodeBaselineSession(data []byte) (*baselineSessionFacts, error) {
 	if err != nil {
 		return nil, err
 	}
-	x := &baselineSessionFacts{SessionID: w.ID, Owner: w.Owner, Statistics: s}
+	x := &baselineSessionFacts{SessionID: w.ID, Owner: w.Owner, Statistics: s, queueURL: w.URL, authorization: w.Token}
 	if len(w.Set) > 0 && string(w.Set) != "null" {
 		// Set contains SDK-defined fields not used here; ambiguity is rejected,
 		// but unrelated forward-compatible set metadata is not copied to history.