diff --git a/go/e2e/agent_ops.go b/go/e2e/agent_ops.go index 193cc35e..cdf2e310 100644 --- a/go/e2e/agent_ops.go +++ b/go/e2e/agent_ops.go @@ -5,6 +5,8 @@ package e2e import ( "context" "fmt" + "strings" + "time" "connectrpc.com/connect" @@ -127,3 +129,101 @@ func (f *Fixture) RemoveWorkspace(ctx context.Context, containerName, clientRequ } return nil } + +// waitRunnerEnrolled blocks until the embedded compass-runner has enrolled with +// the server, or the budget elapses. It is the enrollment counterpart to the +// stack's own waitReady/waitPostgres poll (stack.go): stack.Up returns as soon +// as the runner CHILD is spawned, but the runner enrolls ASYNCHRONOUSLY over the +// TLS door AFTER Up returns, so a leg that Provisions immediately races that +// enrollment and fails `unavailable: no runner enrolled to serve session`. This +// gate closes that race so every Provisioning leg starts against an enrolled +// runner. +// +// The observable enrollment signal available to the cross-process fixture is a +// lightweight enrollment-gated probe. The client GetAgentStatus is served off +// the Server's board projection (server/service.go), NOT a Runner relay, so it +// answers even with no Runner and cannot observe enrollment. StopAgentSession, +// by contrast, relays through the hub's routerFor exactly as Provision does, so +// it returns the CodeUnavailable `no runner enrolled` error until a Runner has +// enrolled — and once one has, a Stop of a synthetic never-started session id is +// an idempotent Runner-side no-op (host.Stop returns success for an unknown +// session; the session-end transcript flush is skipped since the id has no +// entries), so the probe has NO container or session side effect. ONLY that +// specific unavailable-no-runner condition is treated as not-yet-ready; any +// other error is a real failure and is returned immediately. Enrollment is a +// MONOTONIC one-time transition, so this is an event-gated readiness poll on a +// real cross-process signal, not a retry-as-sync: it returns the instant the +// probe stops reporting no-runner. The poll respects ctx cancellation; a budget +// timeout is a legible error. +func (f *Fixture) waitRunnerEnrolled(ctx context.Context) error { + deadline := f.now().Add(enrollPollBudget) + ticker := time.NewTicker(enrollPollInterval) + defer ticker.Stop() + for { + if !f.now().Before(deadline) { + return fmt.Errorf("runner did not enroll within %s", enrollPollBudget) + } + if ready, err := f.runnerEnrolledProbe(ctx, deadline); err != nil { + return err + } else if ready { + return nil + } + select { + case <-ctx.Done(): + return ctx.Err() + case <-ticker.C: + } + } +} + +// enrollProbeSessionID is the synthetic, never-started session id the enrollment +// probe Stops. It is namespaced so it can never collide with a real +// Server-minted session id; a Stop of it is an idempotent Runner-side no-op. +const enrollProbeSessionID = "e2e-enroll-probe-nonexistent-session" + +// runnerEnrolledProbe runs one lightweight enrollment-gated probe. It reports +// ready=true once the Runner is enrolled (the Stop relay no longer returns the +// no-runner error), ready=false while enrollment is still pending (the specific +// CodeUnavailable `no runner enrolled` condition), and a non-nil error for any +// other failure — which waitRunnerEnrolled surfaces immediately rather than +// polling through. +func (f *Fixture) runnerEnrolledProbe(ctx context.Context, deadline time.Time) (ready bool, err error) { + perProbe := rpcTimeout + if !deadline.IsZero() { + if remaining := deadline.Sub(f.now()); remaining < perProbe { + perProbe = remaining + } + } + if perProbe <= 0 { + perProbe = time.Millisecond + } + rctx, cancel := context.WithTimeout(ctx, perProbe) + defer cancel() + _, err = f.Compass().StopAgentSession(rctx, connect.NewRequest(&compassv1.StopAgentSessionRequest{ + SessionId: enrollProbeSessionID, + })) + ready, retry, cerr := classifyEnrollProbe(err) + if retry { + return false, nil + } + return ready, cerr +} + +// classifyEnrollProbe classifies a StopAgentSession probe result into the +// enrollment-readiness signal, as a pure function so its branches are unit +// testable without a live client. A nil error means the Runner is enrolled. The +// substring "no runner enrolled" is a load-bearing cross-package coupling to the +// production error raised by routerFor at go/internal/runnerhub/hub.go; only that specific +// CodeUnavailable condition is treated as not-yet-ready (retry). A CodeUnavailable +// that does NOT carry that message — a transient transport flap — is intentionally +// surfaced as fatal rather than retried, which is acceptable for a deterministic +// e2e readiness gate. Any other error is a real failure surfaced immediately. +func classifyEnrollProbe(err error) (ready bool, retry bool, cerr error) { + if err == nil { + return true, false, nil + } + if connect.CodeOf(err) == connect.CodeUnavailable && strings.Contains(err.Error(), "no runner enrolled") { + return false, true, nil + } + return false, false, fmt.Errorf("runner enrollment probe (StopAgentSession): %w", err) +} diff --git a/go/e2e/agent_ops_enroll_test.go b/go/e2e/agent_ops_enroll_test.go new file mode 100644 index 00000000..914ebcd3 --- /dev/null +++ b/go/e2e/agent_ops_enroll_test.go @@ -0,0 +1,111 @@ +//go:build podman + +package e2e + +import ( + "context" + "errors" + "strings" + "testing" + "time" + + "connectrpc.com/connect" +) + +// TestClassifyEnrollProbe pins the pure classification of a StopAgentSession +// enrollment probe result into (ready, retry, cerr) without a live client. +func TestClassifyEnrollProbe(t *testing.T) { + tests := []struct { + name string + err error + wantReady bool + wantRetry bool + wantErr bool + }{ + { + name: "nil error means enrolled", + err: nil, + wantReady: true, + wantRetry: false, + wantErr: false, + }, + { + name: "unavailable no runner enrolled retries", + err: connect.NewError(connect.CodeUnavailable, errors.New("unavailable: no runner enrolled to serve session")), + wantReady: false, + wantRetry: true, + wantErr: false, + }, + { + name: "internal error is surfaced", + err: connect.NewError(connect.CodeInternal, errors.New("boom")), + wantReady: false, + wantRetry: false, + wantErr: true, + }, + { + name: "unavailable with other message is surfaced not retried", + err: connect.NewError(connect.CodeUnavailable, errors.New("some other unavailable")), + wantReady: false, + wantRetry: false, + wantErr: true, + }, + { + name: "deadline exceeded is surfaced not retried", + err: context.DeadlineExceeded, + wantReady: false, + wantRetry: false, + wantErr: true, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + ready, retry, cerr := classifyEnrollProbe(tt.err) + if ready != tt.wantReady { + t.Errorf("ready = %v, want %v", ready, tt.wantReady) + } + if retry != tt.wantRetry { + t.Errorf("retry = %v, want %v", retry, tt.wantRetry) + } + if tt.wantErr { + if cerr == nil { + t.Fatalf("cerr = nil, want non-nil") + } + if !errors.Is(cerr, tt.err) { + t.Errorf("cerr = %v, want to wrap %v", cerr, tt.err) + } + if !strings.Contains(cerr.Error(), "runner enrollment probe (StopAgentSession)") { + t.Errorf("cerr = %q, want probe prefix", cerr.Error()) + } + } else if cerr != nil { + t.Errorf("cerr = %v, want nil", cerr) + } + }) + } +} + +// TestWaitRunnerEnrolledBudgetTimeout drives the budget-timeout branch of +// waitRunnerEnrolled through the injectable clock seam: the fake clock jumps +// past the deadline on the loop-top check, so the poll returns the clean +// budget-exhausted error without ever firing a live probe (a bare Fixture with +// only now set — Compass() is never reached). +func TestWaitRunnerEnrolledBudgetTimeout(t *testing.T) { + base := time.Now() + calls := 0 + f := &Fixture{ + now: func() time.Time { + calls++ + if calls == 1 { + return base // deadline = base + enrollPollBudget + } + return base.Add(enrollPollBudget + time.Second) // past the deadline + }, + } + err := f.waitRunnerEnrolled(context.Background()) + if err == nil { + t.Fatal("waitRunnerEnrolled() = nil, want budget-timeout error") + } + if !strings.Contains(err.Error(), "did not enroll within") { + t.Errorf("err = %q, want budget-timeout message", err.Error()) + } +} diff --git a/go/e2e/fixture.go b/go/e2e/fixture.go index 474223ce..a78de487 100644 --- a/go/e2e/fixture.go +++ b/go/e2e/fixture.go @@ -49,6 +49,11 @@ type Fixture struct { // WithCannedModel, else nil. Its lifecycle rides a t.Cleanup registered at // startup, so a consumer never closes it directly. stub *cannedModelServer + // now is the injectable wall-clock for the enrollment-readiness poll; + // defaults to time.Now. A test overrides it to drive the budget-timeout + // branch of waitRunnerEnrolled — the enrollment counterpart to the stack's + // s.deps.now() seam. + now func() time.Time } // fixtureConfig holds the optional knobs a caller flips through fixtureOption @@ -262,7 +267,7 @@ func NewFixture(ctx context.Context, t *testing.T, opts ...fixtureOption) *Fixtu t.Fatalf("build authed clients: %v", err) } - return &Fixture{ + f := &Fixture{ compass: compass, comms: comms, stack: st, @@ -271,7 +276,23 @@ func NewFixture(ctx context.Context, t *testing.T, opts ...fixtureOption) *Fixtu serverURL: serverURL, runtimeDir: runtimeDir, stub: stub, + now: time.Now, } + + // stack.Up returns as soon as the compass-runner CHILD is spawned, but the + // runner enrolls with the server ASYNCHRONOUSLY over the TLS door AFTER Up + // returns. A leg that Provisions immediately would otherwise race that + // enrollment and fail `unavailable: no runner enrolled to serve session`. + // Gate the fixture's post-Up readiness on the runner being enrolled — the + // enrollment counterpart to the stack's own waitReady/waitPostgres — so every + // leg starts against an enrolled runner. Event-gated on a real cross-process + // signal (an enrollment-gated probe), never a sleep. On the WithSite re-attach + // path the runner is already enrolled, so the first probe passes immediately. + if err := f.waitRunnerEnrolled(ctx); err != nil { + t.Fatalf("wait for runner enrollment: %v", err) + } + + return f } // cannedAgentDir is the in-container path the canned models.yml is delivered diff --git a/go/e2e/timeouts.go b/go/e2e/timeouts.go index d4544922..07fc08d2 100644 --- a/go/e2e/timeouts.go +++ b/go/e2e/timeouts.go @@ -23,3 +23,20 @@ const settleTimeout = 2 * time.Minute // carries the matching MessagePosted fails visibly here instead of blocking to // the go-test timeout. A deterministic deadline, never a retry loop. const deliverTimeout = 1 * time.Minute + +// enrollPollInterval and enrollPollBudget bound the runner-enrollment readiness +// poll NewFixture runs after stack.Up returns. Up returns as soon as the +// compass-runner CHILD is spawned (stack spawnChain step 7), but the runner +// enrolls ASYNCHRONOUSLY — it dials the server over the TLS door and enrolls +// after Up has already returned — so a leg that Provisions immediately races +// that enrollment. These bound the enrollment counterpart to the stack's own +// waitReady/waitPostgres poll: enrollment is a fast one-time transition (a +// single server dial once the server is already answering), so the budget is +// far smaller than readyPollBudget while still failing a genuinely wedged +// enrollment legibly rather than hanging to the go-test timeout; the interval +// matches readyPollInterval's magnitude. A deterministic deadline, never a +// retry loop. +const ( + enrollPollInterval = 100 * time.Millisecond + enrollPollBudget = 15 * time.Second +)