diff --git a/go/internal/board/projection.go b/go/internal/board/projection.go index 5cc67f9a2..ca33df362 100644 --- a/go/internal/board/projection.go +++ b/go/internal/board/projection.go @@ -50,12 +50,16 @@ type Projection struct { sessions map[string]sessionEntry } -// sessionEntry is the board's per-session record: the latest state and the -// agent account it was attributed to (empty when the hub could not resolve the -// binding — the stated DL-167 residual gap). +// sessionEntry is the board's per-session record: the latest state, the agent +// account it was attributed to (empty when the hub could not resolve the +// binding — the stated DL-167 residual gap), and the owning Runner's runtime +// tier and egress posture (stamped from enrollment, so the snapshot path carries +// them too). type sessionEntry struct { - state compassv1.AgentSessionState - account string + state compassv1.AgentSessionState + account string + tier compassv1.RuntimeTier + egressPosture compassv1.EgressPosture } // NewProjection constructs an empty board over the SubscribeEvents bus it fans @@ -95,7 +99,12 @@ func (p *Projection) PublishSessionStatus(status *compassv1.AgentSessionStatus) } p.mu.Lock() defer p.mu.Unlock() - p.sessions[status.GetSessionId()] = sessionEntry{state: status.GetState(), account: status.GetAgentAccountId()} + p.sessions[status.GetSessionId()] = sessionEntry{ + state: status.GetState(), + account: status.GetAgentAccountId(), + tier: status.GetRuntimeTier(), + egressPosture: status.GetEgressPosture(), + } p.bus.Publish(&compassv1.SubscribeEventsResponse{ Payload: &compassv1.SubscribeEventsResponse_AgentSessionStatus{ @@ -151,7 +160,14 @@ func isTerminal(state compassv1.AgentSessionState) bool { } // statusOf builds one board entry from a retained session record, carrying the -// DL-167 agent_account_id alongside the state. +// DL-167 agent_account_id and the owning Runner's runtime tier and egress +// posture alongside the state. func statusOf(sessionID string, entry sessionEntry) *compassv1.AgentSessionStatus { - return &compassv1.AgentSessionStatus{SessionId: sessionID, State: entry.state, AgentAccountId: entry.account} + return &compassv1.AgentSessionStatus{ + SessionId: sessionID, + State: entry.state, + AgentAccountId: entry.account, + RuntimeTier: entry.tier, + EgressPosture: entry.egressPosture, + } } diff --git a/go/internal/gen/compass/v1/runner.pb.go b/go/internal/gen/compass/v1/runner.pb.go index 70430c697..4b4c68969 100644 --- a/go/internal/gen/compass/v1/runner.pb.go +++ b/go/internal/gen/compass/v1/runner.pb.go @@ -129,9 +129,19 @@ func (RunnerErrorCode) EnumDescriptor() ([]byte, []int) { // CodeUnauthenticated, so the field is a defense-in-depth cross-check, not a // trusted input. The credential itself rides the transport as a bearer token, // never a field here (mirroring IssueTokenRequest, compass.proto:237-242). +// +// runtime_tier and egress_posture are declared once here, not per session: +// they are Runner-wide facts of the one backend this Runner drives, so the hub +// stamps them onto every session it owns rather than have each lifecycle frame +// repeat them. Unlike runner_id they are trusted as declared, with nothing to +// cross-check them against: only the Runner can observe which namespace an +// agent got, and it is the component that enforces egress in the first place, +// so a Runner that lies here has already lost containment either way. type EnrollRequest struct { state protoimpl.MessageState `protogen:"open.v1"` RunnerId string `protobuf:"bytes,1,opt,name=runner_id,json=runnerId,proto3" json:"runner_id,omitempty"` + RuntimeTier v1.RuntimeTier `protobuf:"varint,2,opt,name=runtime_tier,json=runtimeTier,proto3,enum=compass.v1.RuntimeTier" json:"runtime_tier,omitempty"` + EgressPosture v1.EgressPosture `protobuf:"varint,3,opt,name=egress_posture,json=egressPosture,proto3,enum=compass.v1.EgressPosture" json:"egress_posture,omitempty"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } @@ -173,6 +183,20 @@ func (x *EnrollRequest) GetRunnerId() string { return "" } +func (x *EnrollRequest) GetRuntimeTier() v1.RuntimeTier { + if x != nil { + return x.RuntimeTier + } + return v1.RuntimeTier(0) +} + +func (x *EnrollRequest) GetEgressPosture() v1.EgressPosture { + if x != nil { + return x.EgressPosture + } + return v1.EgressPosture(0) +} + // Enroll response: the handshake ack. `reattached` distinguishes a fresh // enrollment from a re-attach of an already-registered Runner (OQ6 duplicate // enrollment) — the wire-level evidence a @@ -2016,9 +2040,11 @@ var File_compass_v1_runner_proto protoreflect.FileDescriptor const file_compass_v1_runner_proto_rawDesc = "" + "\n" + "\x17compass/v1/runner.proto\x12\n" + - "compass.v1\x1a\x1ecompass/v1/agent_gateway.proto\x1a\x16compass/v1/agent.proto\x1a\x18compass/v1/compass.proto\x1a\x16compass/v1/forge.proto\",\n" + + "compass.v1\x1a\x1ecompass/v1/agent_gateway.proto\x1a\x16compass/v1/agent.proto\x1a\x18compass/v1/compass.proto\x1a\x16compass/v1/forge.proto\"\xaa\x01\n" + "\rEnrollRequest\x12\x1b\n" + - "\trunner_id\x18\x01 \x01(\tR\brunnerId\"0\n" + + "\trunner_id\x18\x01 \x01(\tR\brunnerId\x12:\n" + + "\fruntime_tier\x18\x02 \x01(\x0e2\x17.compass.v1.RuntimeTierR\vruntimeTier\x12@\n" + + "\x0eegress_posture\x18\x03 \x01(\x0e2\x19.compass.v1.EgressPostureR\regressPosture\"0\n" + "\x0eEnrollResponse\x12\x1e\n" + "\n" + "reattached\x18\x01 \x01(\bR\n" + @@ -2191,91 +2217,95 @@ var file_compass_v1_runner_proto_goTypes = []any{ (*RelayBoardCallResponse)(nil), // 24: compass.v1.RelayBoardCallResponse (*CommitConversationFrameRequest)(nil), // 25: compass.v1.CommitConversationFrameRequest (*CommitConversationFrameResponse)(nil), // 26: compass.v1.CommitConversationFrameResponse - (*v1.StartAgentSessionResponse)(nil), // 27: compass.v1.StartAgentSessionResponse - (*v1.StopAgentSessionResponse)(nil), // 28: compass.v1.StopAgentSessionResponse - (*v1.ReloadAgentSessionResponse)(nil), // 29: compass.v1.ReloadAgentSessionResponse - (*v1.GetAgentStatusResponse)(nil), // 30: compass.v1.GetAgentStatusResponse - (*v1.ProvisionAgentWorkspaceResponse)(nil), // 31: compass.v1.ProvisionAgentWorkspaceResponse - (*v1.RemoveAgentWorkspaceResponse)(nil), // 32: compass.v1.RemoveAgentWorkspaceResponse - (*v1.StartAgentSessionRequest)(nil), // 33: compass.v1.StartAgentSessionRequest - (*v1.StopAgentSessionRequest)(nil), // 34: compass.v1.StopAgentSessionRequest - (*v1.ReloadAgentSessionRequest)(nil), // 35: compass.v1.ReloadAgentSessionRequest - (*v1.GetAgentStatusRequest)(nil), // 36: compass.v1.GetAgentStatusRequest - (*v1.ProvisionAgentWorkspaceRequest)(nil), // 37: compass.v1.ProvisionAgentWorkspaceRequest - (*ForgeNotification)(nil), // 38: compass.v1.ForgeNotification - (*v1.RemoveAgentWorkspaceRequest)(nil), // 39: compass.v1.RemoveAgentWorkspaceRequest - (*AgentControl)(nil), // 40: compass.v1.AgentControl - (v1.SecretDelivery)(0), // 41: compass.v1.SecretDelivery - (v1.SecretKind)(0), // 42: compass.v1.SecretKind - (*AgentFrame)(nil), // 43: compass.v1.AgentFrame - (*CommsCallRequest)(nil), // 44: compass.v1.CommsCallRequest - (*CommsCallResult)(nil), // 45: compass.v1.CommsCallResult - (*LifecycleCallRequest)(nil), // 46: compass.v1.LifecycleCallRequest - (*LifecycleCallResult)(nil), // 47: compass.v1.LifecycleCallResult - (*ForgeCallRequest)(nil), // 48: compass.v1.ForgeCallRequest - (*ForgeCallResult)(nil), // 49: compass.v1.ForgeCallResult - (*BoardCallRequest)(nil), // 50: compass.v1.BoardCallRequest - (*BoardCallResult)(nil), // 51: compass.v1.BoardCallResult + (v1.RuntimeTier)(0), // 27: compass.v1.RuntimeTier + (v1.EgressPosture)(0), // 28: compass.v1.EgressPosture + (*v1.StartAgentSessionResponse)(nil), // 29: compass.v1.StartAgentSessionResponse + (*v1.StopAgentSessionResponse)(nil), // 30: compass.v1.StopAgentSessionResponse + (*v1.ReloadAgentSessionResponse)(nil), // 31: compass.v1.ReloadAgentSessionResponse + (*v1.GetAgentStatusResponse)(nil), // 32: compass.v1.GetAgentStatusResponse + (*v1.ProvisionAgentWorkspaceResponse)(nil), // 33: compass.v1.ProvisionAgentWorkspaceResponse + (*v1.RemoveAgentWorkspaceResponse)(nil), // 34: compass.v1.RemoveAgentWorkspaceResponse + (*v1.StartAgentSessionRequest)(nil), // 35: compass.v1.StartAgentSessionRequest + (*v1.StopAgentSessionRequest)(nil), // 36: compass.v1.StopAgentSessionRequest + (*v1.ReloadAgentSessionRequest)(nil), // 37: compass.v1.ReloadAgentSessionRequest + (*v1.GetAgentStatusRequest)(nil), // 38: compass.v1.GetAgentStatusRequest + (*v1.ProvisionAgentWorkspaceRequest)(nil), // 39: compass.v1.ProvisionAgentWorkspaceRequest + (*ForgeNotification)(nil), // 40: compass.v1.ForgeNotification + (*v1.RemoveAgentWorkspaceRequest)(nil), // 41: compass.v1.RemoveAgentWorkspaceRequest + (*AgentControl)(nil), // 42: compass.v1.AgentControl + (v1.SecretDelivery)(0), // 43: compass.v1.SecretDelivery + (v1.SecretKind)(0), // 44: compass.v1.SecretKind + (*AgentFrame)(nil), // 45: compass.v1.AgentFrame + (*CommsCallRequest)(nil), // 46: compass.v1.CommsCallRequest + (*CommsCallResult)(nil), // 47: compass.v1.CommsCallResult + (*LifecycleCallRequest)(nil), // 48: compass.v1.LifecycleCallRequest + (*LifecycleCallResult)(nil), // 49: compass.v1.LifecycleCallResult + (*ForgeCallRequest)(nil), // 50: compass.v1.ForgeCallRequest + (*ForgeCallResult)(nil), // 51: compass.v1.ForgeCallResult + (*BoardCallRequest)(nil), // 52: compass.v1.BoardCallRequest + (*BoardCallResult)(nil), // 53: compass.v1.BoardCallResult } var file_compass_v1_runner_proto_depIdxs = []int32{ - 27, // 0: compass.v1.SessionsRequest.start:type_name -> compass.v1.StartAgentSessionResponse - 28, // 1: compass.v1.SessionsRequest.stop:type_name -> compass.v1.StopAgentSessionResponse - 29, // 2: compass.v1.SessionsRequest.reload:type_name -> compass.v1.ReloadAgentSessionResponse - 30, // 3: compass.v1.SessionsRequest.status:type_name -> compass.v1.GetAgentStatusResponse - 31, // 4: compass.v1.SessionsRequest.provision:type_name -> compass.v1.ProvisionAgentWorkspaceResponse - 14, // 5: compass.v1.SessionsRequest.error:type_name -> compass.v1.RunnerError - 32, // 6: compass.v1.SessionsRequest.remove:type_name -> compass.v1.RemoveAgentWorkspaceResponse - 33, // 7: compass.v1.SessionsResponse.start:type_name -> compass.v1.StartAgentSessionRequest - 34, // 8: compass.v1.SessionsResponse.stop:type_name -> compass.v1.StopAgentSessionRequest - 35, // 9: compass.v1.SessionsResponse.reload:type_name -> compass.v1.ReloadAgentSessionRequest - 36, // 10: compass.v1.SessionsResponse.status:type_name -> compass.v1.GetAgentStatusRequest - 37, // 11: compass.v1.SessionsResponse.provision:type_name -> compass.v1.ProvisionAgentWorkspaceRequest - 38, // 12: compass.v1.SessionsResponse.forge_notification:type_name -> compass.v1.ForgeNotification - 10, // 13: compass.v1.SessionsResponse.secrets_version:type_name -> compass.v1.SecretsVersion - 13, // 14: compass.v1.SessionsResponse.config_version:type_name -> compass.v1.ConfigVersion - 6, // 15: compass.v1.SessionsResponse.deliver_control:type_name -> compass.v1.DispatchControl - 39, // 16: compass.v1.SessionsResponse.remove:type_name -> compass.v1.RemoveAgentWorkspaceRequest - 5, // 17: compass.v1.SessionsResponse.resume_body:type_name -> compass.v1.ResumeBody - 40, // 18: compass.v1.DispatchControl.op:type_name -> compass.v1.AgentControl - 9, // 19: compass.v1.FetchSecretsResponse.secrets:type_name -> compass.v1.ResolvedSecret - 41, // 20: compass.v1.ResolvedSecret.delivery:type_name -> compass.v1.SecretDelivery - 42, // 21: compass.v1.ResolvedSecret.kind:type_name -> compass.v1.SecretKind - 0, // 22: compass.v1.RunnerError.code:type_name -> compass.v1.RunnerErrorCode - 43, // 23: compass.v1.PublishEventsRequest.frame:type_name -> compass.v1.AgentFrame - 44, // 24: compass.v1.RelayCommsCallRequest.call:type_name -> compass.v1.CommsCallRequest - 45, // 25: compass.v1.RelayCommsCallResponse.result:type_name -> compass.v1.CommsCallResult - 46, // 26: compass.v1.RelayLifecycleCallRequest.call:type_name -> compass.v1.LifecycleCallRequest - 47, // 27: compass.v1.RelayLifecycleCallResponse.result:type_name -> compass.v1.LifecycleCallResult - 48, // 28: compass.v1.RelayForgeCallRequest.call:type_name -> compass.v1.ForgeCallRequest - 49, // 29: compass.v1.RelayForgeCallResponse.result:type_name -> compass.v1.ForgeCallResult - 50, // 30: compass.v1.RelayBoardCallRequest.call:type_name -> compass.v1.BoardCallRequest - 51, // 31: compass.v1.RelayBoardCallResponse.result:type_name -> compass.v1.BoardCallResult - 43, // 32: compass.v1.CommitConversationFrameRequest.frame:type_name -> compass.v1.AgentFrame - 1, // 33: compass.v1.RunnerService.Enroll:input_type -> compass.v1.EnrollRequest - 3, // 34: compass.v1.RunnerService.Sessions:input_type -> compass.v1.SessionsRequest - 15, // 35: compass.v1.RunnerService.PublishEvents:input_type -> compass.v1.PublishEventsRequest - 17, // 36: compass.v1.RunnerService.RelayCommsCall:input_type -> compass.v1.RelayCommsCallRequest - 19, // 37: compass.v1.RunnerService.RelayLifecycleCall:input_type -> compass.v1.RelayLifecycleCallRequest - 21, // 38: compass.v1.RunnerService.RelayForgeCall:input_type -> compass.v1.RelayForgeCallRequest - 23, // 39: compass.v1.RunnerService.RelayBoardCall:input_type -> compass.v1.RelayBoardCallRequest - 25, // 40: compass.v1.RunnerService.CommitConversationFrame:input_type -> compass.v1.CommitConversationFrameRequest - 7, // 41: compass.v1.RunnerService.FetchSecrets:input_type -> compass.v1.FetchSecretsRequest - 11, // 42: compass.v1.RunnerService.FetchAgentConfig:input_type -> compass.v1.FetchAgentConfigRequest - 2, // 43: compass.v1.RunnerService.Enroll:output_type -> compass.v1.EnrollResponse - 4, // 44: compass.v1.RunnerService.Sessions:output_type -> compass.v1.SessionsResponse - 16, // 45: compass.v1.RunnerService.PublishEvents:output_type -> compass.v1.PublishEventsResponse - 18, // 46: compass.v1.RunnerService.RelayCommsCall:output_type -> compass.v1.RelayCommsCallResponse - 20, // 47: compass.v1.RunnerService.RelayLifecycleCall:output_type -> compass.v1.RelayLifecycleCallResponse - 22, // 48: compass.v1.RunnerService.RelayForgeCall:output_type -> compass.v1.RelayForgeCallResponse - 24, // 49: compass.v1.RunnerService.RelayBoardCall:output_type -> compass.v1.RelayBoardCallResponse - 26, // 50: compass.v1.RunnerService.CommitConversationFrame:output_type -> compass.v1.CommitConversationFrameResponse - 8, // 51: compass.v1.RunnerService.FetchSecrets:output_type -> compass.v1.FetchSecretsResponse - 12, // 52: compass.v1.RunnerService.FetchAgentConfig:output_type -> compass.v1.FetchAgentConfigResponse - 43, // [43:53] is the sub-list for method output_type - 33, // [33:43] is the sub-list for method input_type - 33, // [33:33] is the sub-list for extension type_name - 33, // [33:33] is the sub-list for extension extendee - 0, // [0:33] is the sub-list for field type_name + 27, // 0: compass.v1.EnrollRequest.runtime_tier:type_name -> compass.v1.RuntimeTier + 28, // 1: compass.v1.EnrollRequest.egress_posture:type_name -> compass.v1.EgressPosture + 29, // 2: compass.v1.SessionsRequest.start:type_name -> compass.v1.StartAgentSessionResponse + 30, // 3: compass.v1.SessionsRequest.stop:type_name -> compass.v1.StopAgentSessionResponse + 31, // 4: compass.v1.SessionsRequest.reload:type_name -> compass.v1.ReloadAgentSessionResponse + 32, // 5: compass.v1.SessionsRequest.status:type_name -> compass.v1.GetAgentStatusResponse + 33, // 6: compass.v1.SessionsRequest.provision:type_name -> compass.v1.ProvisionAgentWorkspaceResponse + 14, // 7: compass.v1.SessionsRequest.error:type_name -> compass.v1.RunnerError + 34, // 8: compass.v1.SessionsRequest.remove:type_name -> compass.v1.RemoveAgentWorkspaceResponse + 35, // 9: compass.v1.SessionsResponse.start:type_name -> compass.v1.StartAgentSessionRequest + 36, // 10: compass.v1.SessionsResponse.stop:type_name -> compass.v1.StopAgentSessionRequest + 37, // 11: compass.v1.SessionsResponse.reload:type_name -> compass.v1.ReloadAgentSessionRequest + 38, // 12: compass.v1.SessionsResponse.status:type_name -> compass.v1.GetAgentStatusRequest + 39, // 13: compass.v1.SessionsResponse.provision:type_name -> compass.v1.ProvisionAgentWorkspaceRequest + 40, // 14: compass.v1.SessionsResponse.forge_notification:type_name -> compass.v1.ForgeNotification + 10, // 15: compass.v1.SessionsResponse.secrets_version:type_name -> compass.v1.SecretsVersion + 13, // 16: compass.v1.SessionsResponse.config_version:type_name -> compass.v1.ConfigVersion + 6, // 17: compass.v1.SessionsResponse.deliver_control:type_name -> compass.v1.DispatchControl + 41, // 18: compass.v1.SessionsResponse.remove:type_name -> compass.v1.RemoveAgentWorkspaceRequest + 5, // 19: compass.v1.SessionsResponse.resume_body:type_name -> compass.v1.ResumeBody + 42, // 20: compass.v1.DispatchControl.op:type_name -> compass.v1.AgentControl + 9, // 21: compass.v1.FetchSecretsResponse.secrets:type_name -> compass.v1.ResolvedSecret + 43, // 22: compass.v1.ResolvedSecret.delivery:type_name -> compass.v1.SecretDelivery + 44, // 23: compass.v1.ResolvedSecret.kind:type_name -> compass.v1.SecretKind + 0, // 24: compass.v1.RunnerError.code:type_name -> compass.v1.RunnerErrorCode + 45, // 25: compass.v1.PublishEventsRequest.frame:type_name -> compass.v1.AgentFrame + 46, // 26: compass.v1.RelayCommsCallRequest.call:type_name -> compass.v1.CommsCallRequest + 47, // 27: compass.v1.RelayCommsCallResponse.result:type_name -> compass.v1.CommsCallResult + 48, // 28: compass.v1.RelayLifecycleCallRequest.call:type_name -> compass.v1.LifecycleCallRequest + 49, // 29: compass.v1.RelayLifecycleCallResponse.result:type_name -> compass.v1.LifecycleCallResult + 50, // 30: compass.v1.RelayForgeCallRequest.call:type_name -> compass.v1.ForgeCallRequest + 51, // 31: compass.v1.RelayForgeCallResponse.result:type_name -> compass.v1.ForgeCallResult + 52, // 32: compass.v1.RelayBoardCallRequest.call:type_name -> compass.v1.BoardCallRequest + 53, // 33: compass.v1.RelayBoardCallResponse.result:type_name -> compass.v1.BoardCallResult + 45, // 34: compass.v1.CommitConversationFrameRequest.frame:type_name -> compass.v1.AgentFrame + 1, // 35: compass.v1.RunnerService.Enroll:input_type -> compass.v1.EnrollRequest + 3, // 36: compass.v1.RunnerService.Sessions:input_type -> compass.v1.SessionsRequest + 15, // 37: compass.v1.RunnerService.PublishEvents:input_type -> compass.v1.PublishEventsRequest + 17, // 38: compass.v1.RunnerService.RelayCommsCall:input_type -> compass.v1.RelayCommsCallRequest + 19, // 39: compass.v1.RunnerService.RelayLifecycleCall:input_type -> compass.v1.RelayLifecycleCallRequest + 21, // 40: compass.v1.RunnerService.RelayForgeCall:input_type -> compass.v1.RelayForgeCallRequest + 23, // 41: compass.v1.RunnerService.RelayBoardCall:input_type -> compass.v1.RelayBoardCallRequest + 25, // 42: compass.v1.RunnerService.CommitConversationFrame:input_type -> compass.v1.CommitConversationFrameRequest + 7, // 43: compass.v1.RunnerService.FetchSecrets:input_type -> compass.v1.FetchSecretsRequest + 11, // 44: compass.v1.RunnerService.FetchAgentConfig:input_type -> compass.v1.FetchAgentConfigRequest + 2, // 45: compass.v1.RunnerService.Enroll:output_type -> compass.v1.EnrollResponse + 4, // 46: compass.v1.RunnerService.Sessions:output_type -> compass.v1.SessionsResponse + 16, // 47: compass.v1.RunnerService.PublishEvents:output_type -> compass.v1.PublishEventsResponse + 18, // 48: compass.v1.RunnerService.RelayCommsCall:output_type -> compass.v1.RelayCommsCallResponse + 20, // 49: compass.v1.RunnerService.RelayLifecycleCall:output_type -> compass.v1.RelayLifecycleCallResponse + 22, // 50: compass.v1.RunnerService.RelayForgeCall:output_type -> compass.v1.RelayForgeCallResponse + 24, // 51: compass.v1.RunnerService.RelayBoardCall:output_type -> compass.v1.RelayBoardCallResponse + 26, // 52: compass.v1.RunnerService.CommitConversationFrame:output_type -> compass.v1.CommitConversationFrameResponse + 8, // 53: compass.v1.RunnerService.FetchSecrets:output_type -> compass.v1.FetchSecretsResponse + 12, // 54: compass.v1.RunnerService.FetchAgentConfig:output_type -> compass.v1.FetchAgentConfigResponse + 45, // [45:55] is the sub-list for method output_type + 35, // [35:45] is the sub-list for method input_type + 35, // [35:35] is the sub-list for extension type_name + 35, // [35:35] is the sub-list for extension extendee + 0, // [0:35] is the sub-list for field type_name } func init() { file_compass_v1_runner_proto_init() } diff --git a/go/internal/runner/enroll_identity_test.go b/go/internal/runner/enroll_identity_test.go new file mode 100644 index 000000000..8ab31b4c8 --- /dev/null +++ b/go/internal/runner/enroll_identity_test.go @@ -0,0 +1,92 @@ +//go:build unix + +package runner + +// Dial declares the Runner's runtime tier and egress posture ONCE at enrollment, +// derived from the engine it drives. This drives the real Dial (interceptor- +// wrapped client) against a recording RunnerService handler and asserts the +// EnrollRequest carried the right wire enums for a HOST engine (tier HOST, egress +// UNENFORCED) — the two facts the hub stamps onto every session status. + +import ( + "context" + "net/http" + "net/http/httptest" + "sync" + "testing" + + "connectrpc.com/connect" + + compassv1 "github.com/RigelBuild/compass/go/gen/compass/v1" + compassv1internal "github.com/RigelBuild/compass/go/internal/gen/compass/v1" + "github.com/RigelBuild/compass/go/internal/gen/compass/v1/compassv1internalconnect" + "github.com/RigelBuild/compass/go/internal/runtime" +) + +// recordingEnroll is a RunnerService handler that captures the EnrollRequest it +// received, so a Dial test can assert on the tier + posture the client declared. +type recordingEnroll struct { + compassv1internalconnect.UnimplementedRunnerServiceHandler + mu sync.Mutex + req *compassv1internal.EnrollRequest +} + +func (r *recordingEnroll) Enroll(_ context.Context, req *connect.Request[compassv1internal.EnrollRequest]) (*connect.Response[compassv1internal.EnrollResponse], error) { + r.mu.Lock() + defer r.mu.Unlock() + r.req = req.Msg + return connect.NewResponse(&compassv1internal.EnrollResponse{Reattached: false}), nil +} + +func (r *recordingEnroll) enrolled() *compassv1internal.EnrollRequest { + r.mu.Lock() + defer r.mu.Unlock() + return r.req +} + +// recordingEnrollServer stands up an h2c httptest RunnerService serving rec and +// returns its base URL, torn down via t.Cleanup. +func recordingEnrollServer(t *testing.T, rec *recordingEnroll) string { + t.Helper() + path, handler := compassv1internalconnect.NewRunnerServiceHandler(rec) + mux := http.NewServeMux() + mux.Handle(path, handler) + srv := httptest.NewUnstartedServer(mux) + srv.Config.Protocols = cleartextHTTP2() + srv.Start() + t.Cleanup(srv.Close) + return srv.URL +} + +// TestDialDeclaresEngineTierAndPosture pins the enrollment declaration: dialing +// with a HOST engine sends runtime_tier=HOST and egress_posture=UNENFORCED on the +// EnrollRequest, derived from the engine via runtime.TierOf / runtime.PostureOf. +// +// Negative control: dropping the two fields from Dial's EnrollRequest (the +// pre-fix shape, runner_id only) reddens both assertions — observed "runtime_tier +// = RUNTIME_TIER_UNSPECIFIED, want RUNTIME_TIER_HOST". +func TestDialDeclaresEngineTierAndPosture(t *testing.T) { + rec := &recordingEnroll{} + url := recordingEnrollServer(t, rec) + + // context.Background() is the test root context. + if _, err := Dial(context.Background(), RunnerConfig{ + RunnerID: "r-1", + ServerAddr: url, + Token: "tok", + Engine: runtime.NewHostRuntime(t.TempDir()), + }); err != nil { + t.Fatalf("Dial = %v, want success", err) + } + + got := rec.enrolled() + if got == nil { + t.Fatal("handler recorded no EnrollRequest") + } + if got.GetRuntimeTier() != compassv1.RuntimeTier_RUNTIME_TIER_HOST { + t.Errorf("EnrollRequest runtime_tier = %v, want RUNTIME_TIER_HOST", got.GetRuntimeTier()) + } + if got.GetEgressPosture() != compassv1.EgressPosture_EGRESS_POSTURE_UNENFORCED { + t.Errorf("EnrollRequest egress_posture = %v, want EGRESS_POSTURE_UNENFORCED", got.GetEgressPosture()) + } +} diff --git a/go/internal/runner/runner.go b/go/internal/runner/runner.go index 3e011a408..6e6f2bc0c 100644 --- a/go/internal/runner/runner.go +++ b/go/internal/runner/runner.go @@ -115,7 +115,9 @@ func Dial(ctx context.Context, cfg RunnerConfig) (*ServerLink, error) { connect.WithInterceptors(otelInterceptor, &bearerToken{token: cfg.Token}), ) resp, err := client.Enroll(ctx, connect.NewRequest(&compassv1internal.EnrollRequest{ - RunnerId: cfg.RunnerID, + RunnerId: cfg.RunnerID, + RuntimeTier: runtimeTierProto(runtime.TierOf(cfg.Engine)), + EgressPosture: egressPostureProto(runtime.PostureOf(cfg.Engine)), })) if err != nil { return nil, fmt.Errorf("enrolling runner %q: %w", cfg.RunnerID, err) diff --git a/go/internal/runnerhub/binding_cache_test.go b/go/internal/runnerhub/binding_cache_test.go index 40279e077..58d851e2c 100644 --- a/go/internal/runnerhub/binding_cache_test.go +++ b/go/internal/runnerhub/binding_cache_test.go @@ -31,6 +31,7 @@ import ( "sync" "testing" + compassv1 "github.com/RigelBuild/compass/go/gen/compass/v1" "github.com/RigelBuild/compass/go/internal/fabric" "github.com/RigelBuild/compass/go/internal/store" ) @@ -228,7 +229,7 @@ func TestRestartResolvesPreRestartBindingBothDirections(t *testing.T) { // A FIRST enroll on a fresh hub (the restart case): reattached is false, so // enroll does NOT reap the durable rows — they are still valid. - if reattached := hub.enroll(context.Background(), "runner-1", runnerSubject()); reattached { + if reattached := hub.enroll(context.Background(), "runner-1", runnerSubject(), compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED); reattached { t.Fatal("first enroll on a fresh hub reported reattached=true, want false (no durable reap)") } @@ -255,7 +256,7 @@ func TestFailClosedStoppedNeverSeenAndPostReconnect(t *testing.T) { hub := newHubOnly() bindings := newFakeBindingStore() hub.SetSessionBindingStore(bindings) - hub.enroll(context.Background(), "runner-1", runnerSubject()) + hub.enroll(context.Background(), "runner-1", runnerSubject(), compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED) // Never-seen: no binding anywhere. if acct, ok := hub.accountForSession(context.Background(), "ghost"); ok { @@ -283,7 +284,7 @@ func TestFailClosedStoppedNeverSeenAndPostReconnect(t *testing.T) { t.Fatalf("accountForSession(sess-pre) before reconnect = (%q, %v), want (%s, true)", acct, ok, testAgentAccount) } // Runner reconnects: a re-enroll (reattached=true) durably reaps. - if reattached := hub.enroll(context.Background(), "runner-1", runnerSubject()); !reattached { + if reattached := hub.enroll(context.Background(), "runner-1", runnerSubject(), compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED); !reattached { t.Fatal("second enroll reported reattached=false, want true (a Runner reconnect)") } if acct, ok := hub.accountForSession(context.Background(), "sess-pre"); ok { @@ -320,7 +321,7 @@ func TestPeerBindingChangeEvictsOtherInstanceCache(t *testing.T) { bindingsB.seed("sess-old") hubB.SetSessionBindingStore(bindingsB) hubB.SetRoutingFabric(routing) - hubB.enroll(context.Background(), "runner-1", runnerSubject()) + hubB.enroll(context.Background(), "runner-1", runnerSubject(), compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED) hubB.bindContainer("cont-new", testAgentAccount) // B re-points the account onto sess-new: displaces sess-old, publishes the @@ -360,7 +361,7 @@ func TestDisplacedSessionResolvesNowhere(t *testing.T) { hub := newHubOnly() bindings := newFakeBindingStore() hub.SetSessionBindingStore(bindings) - hub.enroll(context.Background(), "runner-1", runnerSubject()) + hub.enroll(context.Background(), "runner-1", runnerSubject(), compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED) // Bind the account to sess-old. hub.bindContainer("cont-old", testAgentAccount) @@ -406,7 +407,7 @@ func TestAckPathBindingReadNeverRunsUnderSystemRole(t *testing.T) { hub.SetDeliveryStore(del) // Enrolled Runner + empty maps (no bindSession) => the ack's account // resolve is a cache MISS that falls through to the read-through table. - hub.enroll(context.Background(), "runner-1", runnerSubject()) + hub.enroll(context.Background(), "runner-1", runnerSubject(), compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED) if err := hub.Deliver(context.Background(), RunnerEvent{ RunnerSeq: 1, SessionID: "sess-1", Frame: deliveryAckFrame("m1"), @@ -434,7 +435,7 @@ func TestAckPathBindingReadNeverRunsUnderSystemRole(t *testing.T) { hub.SetSessionBindingStore(bindings) del := newFakeDeliveryStore() hub.SetDeliveryStore(del) - hub.enroll(context.Background(), "runner-1", runnerSubject()) + hub.enroll(context.Background(), "runner-1", runnerSubject(), compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED) if err := hub.Deliver(context.Background(), RunnerEvent{ RunnerSeq: 1, SessionID: "sess-1", Frame: forgeAckFrame("sub-1"), @@ -470,7 +471,7 @@ func TestStoreFaultsFallBackWithoutLosingFailClosed(t *testing.T) { routing := &fakeRoutingFabric{} hub.SetSessionBindingStore(bindings) hub.SetRoutingFabric(routing) - hub.enroll(context.Background(), "runner-1", runnerSubject()) + hub.enroll(context.Background(), "runner-1", runnerSubject(), compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED) hub.bindContainer("cont-1", testAgentAccount) hub.promoteSession(context.Background(), "cont-1", "sess-1") @@ -488,13 +489,13 @@ func TestStoreFaultsFallBackWithoutLosingFailClosed(t *testing.T) { hub := newHubOnly() bindings := newFakeBindingStore() hub.SetSessionBindingStore(bindings) - hub.enroll(context.Background(), "runner-1", runnerSubject()) + hub.enroll(context.Background(), "runner-1", runnerSubject(), compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED) hub.bindContainer("cont-1", testAgentAccount) hub.promoteSession(context.Background(), "cont-1", "sess-1") // The reconnect sweep now faults; the in-RAM snapshot must still drive it. bindings.deleteForRunnerErr = faultErr - hub.enroll(context.Background(), "runner-1", runnerSubject()) + hub.enroll(context.Background(), "runner-1", runnerSubject(), compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED) if acct, ok := hub.accountForSession(context.Background(), "sess-1"); ok { t.Fatalf("accountForSession(sess-1) = (%q, true) after a reconnect, want fail-closed: a reap fault must not leave a dead session resolvable", acct) @@ -507,7 +508,7 @@ func TestStoreFaultsFallBackWithoutLosingFailClosed(t *testing.T) { bindings.seed("sess-1") bindings.resolveErr = faultErr hub.SetSessionBindingStore(bindings) - hub.enroll(context.Background(), "runner-1", runnerSubject()) + hub.enroll(context.Background(), "runner-1", runnerSubject(), compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED) if acct, ok := hub.accountForSession(context.Background(), "sess-1"); ok { t.Fatalf("accountForSession(sess-1) = (%q, true), want fail-closed: a store fault must never resolve", acct) @@ -520,7 +521,7 @@ func TestStoreFaultsFallBackWithoutLosingFailClosed(t *testing.T) { bindings.seed("sess-1") bindings.reverseErr = faultErr hub.SetSessionBindingStore(bindings) - hub.enroll(context.Background(), "runner-1", runnerSubject()) + hub.enroll(context.Background(), "runner-1", runnerSubject(), compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED) if sess, ok := hub.SessionForAccount(context.Background(), testAgentAccount); ok { t.Fatalf("SessionForAccount = (%q, true), want fail-closed: a store fault must never resolve", sess) @@ -541,7 +542,7 @@ func TestReusedSessionIDConflictIsSwallowed(t *testing.T) { hub := newHubOnly() bindings := newFakeBindingStore() hub.SetSessionBindingStore(bindings) - hub.enroll(context.Background(), "runner-1", runnerSubject()) + hub.enroll(context.Background(), "runner-1", runnerSubject(), compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED) // A row from before the joint restart, under a DIFFERENT account. bindings.mu.Lock() @@ -580,7 +581,7 @@ func TestConcurrentResolveDuringAFaultingReapCannotResurrect(t *testing.T) { release: make(chan struct{}), } hub.SetSessionBindingStore(bindings) - hub.enroll(context.Background(), "runner-1", runnerSubject()) + hub.enroll(context.Background(), "runner-1", runnerSubject(), compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED) hub.bindContainer("cont-1", testAgentAccount) hub.promoteSession(context.Background(), "cont-1", "sess-1") @@ -590,7 +591,7 @@ func TestConcurrentResolveDuringAFaultingReapCannotResurrect(t *testing.T) { done := make(chan struct{}) go func() { defer close(done) - hub.enroll(context.Background(), "runner-1", runnerSubject()) + hub.enroll(context.Background(), "runner-1", runnerSubject(), compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED) }() <-bindings.entered // maps are cleared; the reap is in flight and will fail diff --git a/go/internal/runnerhub/commands_test.go b/go/internal/runnerhub/commands_test.go index 5be44ff5b..fc90ae9ad 100644 --- a/go/internal/runnerhub/commands_test.go +++ b/go/internal/runnerhub/commands_test.go @@ -99,7 +99,7 @@ func TestStartRelaySurfacesAlreadyRunningAsAlreadyExists(t *testing.T) { hub := newHubOnly() // Enroll a Runner and bind a send that answers every command with an // ALREADY_RUNNING error result correlated by the pushed request id. - hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}) + hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}, compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED) router, _, err := hub.routerFor("any") if err != nil { t.Fatalf("routerFor after enroll = %v, want a router", err) @@ -130,7 +130,7 @@ func TestStartRelaySurfacesAlreadyRunningAsAlreadyExists(t *testing.T) { // exercised. func TestStartRelayReturnsSessionIdOnSuccess(t *testing.T) { hub := newHubOnly() - hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}) + hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}, compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED) router, _, _ := hub.routerFor("any") router.attach(func(cmd *compassv1internal.SessionsResponse) error { go router.complete(&compassv1internal.SessionsRequest{ @@ -156,7 +156,7 @@ func TestStartRelayReturnsSessionIdOnSuccess(t *testing.T) { // from. A non-zero count means the initial-signal path was re-introduced. func TestStartEmitsNoInitialSignal(t *testing.T) { hub := newHubOnly() - hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}) + hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}, compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED) hub.bindContainer("c1", testAgentAccount) router, _, _ := hub.routerFor("any") rec := newRecordingSend() @@ -187,7 +187,7 @@ func TestStartEmitsNoInitialSignal(t *testing.T) { // variant, and the typed result flows back. func TestRemoveRelayReturnsResponseOnSuccess(t *testing.T) { hub := newHubOnly() - hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}) + hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}, compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED) router, _, _ := hub.routerFor("any") var sawRemove bool router.attach(func(cmd *compassv1internal.SessionsResponse) error { @@ -223,7 +223,7 @@ func TestRemoveRelayReturnsResponseOnSuccess(t *testing.T) { // true after teardown and reddens this. func TestRemoveClearsContainerBinding(t *testing.T) { hub := newHubOnly() - hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}) + hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}, compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED) hub.bindContainer("c1", testAgentAccount) if !hub.HasContainerBinding("c1") { t.Fatal("precondition: container c1 should be bound after bindContainer") @@ -250,7 +250,7 @@ func TestRemoveClearsContainerBinding(t *testing.T) { // the pushed request id — the seam the SessionState fallback tests drive. func attachStatusResponder(t *testing.T, hub *Hub, statuses []*compassv1.AgentSessionStatus) { t.Helper() - hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}) + hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}, compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED) router, _, err := hub.routerFor("any") if err != nil { t.Fatalf("routerFor after enroll = %v, want a router", err) diff --git a/go/internal/runnerhub/config_fetch_test.go b/go/internal/runnerhub/config_fetch_test.go index 191aae20f..7c1e39955 100644 --- a/go/internal/runnerhub/config_fetch_test.go +++ b/go/internal/runnerhub/config_fetch_test.go @@ -25,6 +25,7 @@ import ( "connectrpc.com/connect" + compassv1 "github.com/RigelBuild/compass/go/gen/compass/v1" compassv1internal "github.com/RigelBuild/compass/go/internal/gen/compass/v1" "github.com/RigelBuild/compass/go/internal/store" ) @@ -83,7 +84,7 @@ func drainConfigStream(t *testing.T, stream *connect.ServerStreamForClient[compa // from the CodeUnavailable of a transient transport fault. func TestFetchAgentConfigNoConfigStoreFailsPrecondition(t *testing.T) { hub := newHubOnly() - hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}) + hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}, compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED) url := newMountedH2CServerWithConfig(t, hub, runnerResolverForFetch().resolve, nil) client := newRawRunnerClient(t, url, "runner-tok") @@ -108,7 +109,7 @@ func TestFetchAgentConfigNoConfigStoreFailsPrecondition(t *testing.T) { // loop is exercised (a bug that sent one frame, or dropped the tail, reds). func TestFetchAgentConfigStreamsVersionThenChunks(t *testing.T) { hub := newHubOnly() - hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}) + hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}, compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED) // A bundle spanning multiple chunk frames. want := bytes.Repeat([]byte("compass-config-"), configChunkBytes/10) cfg := &fakeConfigStore{version: "v-1", bundle: want} @@ -137,7 +138,7 @@ func TestFetchAgentConfigStreamsVersionThenChunks(t *testing.T) { // state (the Runner materializes an empty dir), never an error. func TestFetchAgentConfigUnconfiguredEmptyVersion(t *testing.T) { hub := newHubOnly() - hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}) + hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}, compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED) cfg := &fakeConfigStore{err: store.ErrNotFound} url := newMountedH2CServerWithConfig(t, hub, runnerResolverForFetch().resolve, cfg) client := newRawRunnerClient(t, url, "runner-tok") @@ -161,7 +162,7 @@ func TestFetchAgentConfigUnconfiguredEmptyVersion(t *testing.T) { // not read as "no config". func TestFetchAgentConfigStoreErrorMapsToInternal(t *testing.T) { hub := newHubOnly() - hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}) + hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}, compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED) cfg := &fakeConfigStore{err: errors.New("db boom")} url := newMountedH2CServerWithConfig(t, hub, runnerResolverForFetch().resolve, cfg) client := newRawRunnerClient(t, url, "runner-tok") @@ -186,7 +187,7 @@ func TestFetchAgentConfigStoreErrorMapsToInternal(t *testing.T) { // that streamed the bundle anyway (wasting the reconnect) reds. func TestFetchAgentConfigIfVersionMatchStreamsVersionOnly(t *testing.T) { hub := newHubOnly() - hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}) + hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}, compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED) cfg := &fakeConfigStore{version: "v-held", bundle: []byte("should-not-be-sent")} url := newMountedH2CServerWithConfig(t, hub, runnerResolverForFetch().resolve, cfg) client := newRawRunnerClient(t, url, "runner-tok") @@ -210,7 +211,7 @@ func TestFetchAgentConfigIfVersionMatchStreamsVersionOnly(t *testing.T) { // a stale Runner reconnecting with an old version still gets the new bytes. func TestFetchAgentConfigIfVersionMismatchStreamsBundle(t *testing.T) { hub := newHubOnly() - hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}) + hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}, compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED) cfg := &fakeConfigStore{version: "v-new", bundle: []byte("new-bytes")} url := newMountedH2CServerWithConfig(t, hub, runnerResolverForFetch().resolve, cfg) client := newRawRunnerClient(t, url, "runner-tok") diff --git a/go/internal/runnerhub/config_signal_test.go b/go/internal/runnerhub/config_signal_test.go index 5e5e860b8..1a432ddd9 100644 --- a/go/internal/runnerhub/config_signal_test.go +++ b/go/internal/runnerhub/config_signal_test.go @@ -14,6 +14,7 @@ import ( "context" "testing" + compassv1 "github.com/RigelBuild/compass/go/gen/compass/v1" compassv1internal "github.com/RigelBuild/compass/go/internal/gen/compass/v1" "github.com/RigelBuild/compass/go/internal/store" ) @@ -25,7 +26,7 @@ import ( // version), never a minted token. func TestSignalConfigVersionPushesStoreVersion(t *testing.T) { hub := newHubOnly() - hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}) + hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}, compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED) bindSession(hub, "sess-a") bindSession(hub, "sess-b") router, _, err := hub.routerFor("any") @@ -60,7 +61,7 @@ func TestSignalConfigVersionPushesStoreVersion(t *testing.T) { // fleet-cleared marker the Runner reads as "materialize an empty dir". func TestSignalConfigVersionEmptyVersionIsTheClearedMarker(t *testing.T) { hub := newHubOnly() - hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}) + hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}, compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED) bindSession(hub, "sess-a") router, _, err := hub.routerFor("any") if err != nil { @@ -87,7 +88,7 @@ func TestSignalConfigVersionEmptyVersionIsTheClearedMarker(t *testing.T) { // (nothing bound) pushes nothing and is a clean success. func TestSignalConfigVersionNoLiveSessionsIsNoop(t *testing.T) { hub := newHubOnly() - hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}) + hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}, compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED) router, _, _ := hub.routerFor("any") rec := newRecordingSend() router.attach(rec.send) diff --git a/go/internal/runnerhub/deliveryarm_test.go b/go/internal/runnerhub/deliveryarm_test.go index 66cb1be08..4b57ac0bf 100644 --- a/go/internal/runnerhub/deliveryarm_test.go +++ b/go/internal/runnerhub/deliveryarm_test.go @@ -349,7 +349,7 @@ func TestDeliverSessionNilPresenceSinkIsSafe(t *testing.T) { // DispatchControl returns without blocking. func TestDispatchControlSendOnlyDoesNotBlock(t *testing.T) { hub := newHubOnly() - hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}) + hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}, compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED) router, _, err := hub.routerFor("sess-1") if err != nil { t.Fatalf("routerFor: %v", err) @@ -400,7 +400,7 @@ func TestDispatchControlSendOnlyDoesNotBlock(t *testing.T) { // is observed (counted), not dropped as unknown. func TestDispatchControlNoLiveStreamRefuses(t *testing.T) { hub := newHubOnly() - hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}) + hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}, compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED) // No stream attached: send is nil. op := &compassv1internal.AgentControl{ Control: &compassv1internal.AgentControl_Deliver{ diff --git a/go/internal/runnerhub/enroll_reap_test.go b/go/internal/runnerhub/enroll_reap_test.go index 18fd6c6dd..87617bb87 100644 --- a/go/internal/runnerhub/enroll_reap_test.go +++ b/go/internal/runnerhub/enroll_reap_test.go @@ -17,6 +17,7 @@ import ( "sync" "testing" + compassv1 "github.com/RigelBuild/compass/go/gen/compass/v1" "github.com/RigelBuild/compass/go/internal/store" ) @@ -53,7 +54,7 @@ func TestEnrollFiresReapSinkWithClearedSessionIDs(t *testing.T) { hub.SetSessionReapSink(fake) // A first enroll binds the Runner, then two live sessions promote onto it. - hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}) + hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}, compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED) hub.bindContainer("c1", "acct-a") hub.promoteSession(context.Background(), "c1", "sess-a") hub.bindContainer("c2", "acct-b") @@ -66,7 +67,7 @@ func TestEnrollFiresReapSinkWithClearedSessionIDs(t *testing.T) { } // The Runner reconnects: enroll clears both bindings and reaps both ids. - hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}) + hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}, compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED) calls := fake.snapshot() if len(calls) != 2 { @@ -85,12 +86,12 @@ func TestEnrollFiresReapSinkWithClearedSessionIDs(t *testing.T) { func TestEnrollNilReapSinkStillClears(t *testing.T) { hub := newHubOnly() // no SetSessionReapSink - hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}) + hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}, compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED) hub.bindContainer("c1", "acct-a") hub.promoteSession(context.Background(), "c1", "sess-a") // A re-enroll with no reap sink clears the binding without panicking. - hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}) + hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}, compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED) if sess, ok := hub.SessionForAccount(context.Background(), "acct-a"); ok { t.Fatalf("SessionForAccount(acct-a) = %q ok=true after re-enroll, want ok=false (binding cleared)", sess) diff --git a/go/internal/runnerhub/handler.go b/go/internal/runnerhub/handler.go index 943ec071a..2c26ed405 100644 --- a/go/internal/runnerhub/handler.go +++ b/go/internal/runnerhub/handler.go @@ -82,7 +82,7 @@ func (h *Handler) Enroll(ctx context.Context, req *connect.Request[compassv1inte // enroll under an identity other than its token's. return nil, errUnauthenticated } - reattached := h.hub.enroll(ctx, subj.ID, subj) + reattached := h.hub.enroll(ctx, subj.ID, subj, req.Msg.GetRuntimeTier(), req.Msg.GetEgressPosture()) return connect.NewResponse(&compassv1internal.EnrollResponse{Reattached: reattached}), nil } diff --git a/go/internal/runnerhub/hub.go b/go/internal/runnerhub/hub.go index 2432cdd4b..362b7be00 100644 --- a/go/internal/runnerhub/hub.go +++ b/go/internal/runnerhub/hub.go @@ -486,9 +486,11 @@ type Hub struct { // attachedRunner is one enrolled Runner: its id, its authenticated token // subject, and the command router that reaches its live Sessions stream. type attachedRunner struct { - id string - subject store.Subject - router *commandRouter + id string + subject store.Subject + router *commandRouter + tier compassv1.RuntimeTier + egressPosture compassv1.EgressPosture } // NewHub constructs a hub over the two write-through sinks and the agent-comms @@ -801,8 +803,14 @@ func (h *Hub) deliverSession(ctx context.Context, sessionID string, sf *compassv // unbindSession has not yet dropped it) carries its account; one published // after a Runner reconnect cleared the maps carries none (the stated residual // gap). accountForSession takes h.mu; deliverSession holds no lock here. + // runnerRuntimeIdentity reads the enrolled Runner's tier/posture under the + // same lock, so a status published after a reattach reflects the newly + // enrolled Runner's values, never the previous Runner's. It is a separate + // acquisition from the account read above, so the two can straddle a + // re-enroll; harmless while one Runner enrolls at a time. account, hasAccount := h.accountForSession(ctx, sessionID) - status := &compassv1.AgentSessionStatus{SessionId: sessionID, State: state} + tier, egressPosture := h.runnerRuntimeIdentity() + status := &compassv1.AgentSessionStatus{SessionId: sessionID, State: state, RuntimeTier: tier, EgressPosture: egressPosture} if hasAccount { status.AgentAccountId = string(account) } @@ -1045,12 +1053,12 @@ type promotedPair struct { // // A hub with no binding store wired keeps the original in-RAM snapshot behaviour // (every existing enroll test), driving offline/reapedSessions from the maps. -func (h *Hub) enroll(ctx context.Context, id string, subject store.Subject) (reattached bool) { +func (h *Hub) enroll(ctx context.Context, id string, subject store.Subject, tier compassv1.RuntimeTier, egressPosture compassv1.EgressPosture) (reattached bool) { h.mu.Lock() reattached = h.runner != nil router := newCommandRouter() router.log = h.log - h.runner = &attachedRunner{id: id, subject: subject, router: router} + h.runner = &attachedRunner{id: id, subject: subject, router: router, tier: tier, egressPosture: egressPosture} // Snapshot the live (account -> session) bindings BEFORE clearing them, for // the no-store path: each previously-bound account loses its live session on // this re-enroll and must be driven to presence OFFLINE (RIG-1569 T8). enroll @@ -1155,3 +1163,19 @@ func (h *Hub) routerFor(sessionID string) (*commandRouter, string, error) { } return h.runner.router, h.runner.id, nil } + +// runnerRuntimeIdentity returns the enrolled Runner's declared runtime tier and +// egress posture under h.mu, so a session status stamps the Runner that owns it +// today rather than racing a re-enroll. No Runner enrolled yields UNSPECIFIED on +// both — the wire's "we do not know", never a plausible default. +func (h *Hub) runnerRuntimeIdentity() (compassv1.RuntimeTier, compassv1.EgressPosture) { + h.mu.Lock() + defer h.mu.Unlock() + if h.runner == nil { + // Reachable only before the first enroll: no disconnect path nils + // h.runner, and that is load-bearing — the board replaces its whole + // entry per publish, so a later UNSPECIFIED would regress a known tier. + return compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED + } + return h.runner.tier, h.runner.egressPosture +} diff --git a/go/internal/runnerhub/hub_test.go b/go/internal/runnerhub/hub_test.go index 48894092b..9683c3c39 100644 --- a/go/internal/runnerhub/hub_test.go +++ b/go/internal/runnerhub/hub_test.go @@ -167,10 +167,10 @@ func TestEnrollDuplicateReattaches(t *testing.T) { hub := newHubOnly() subj := store.Subject{Kind: store.SubjectRunner, ID: "runner-1"} - if reattached := hub.enroll(context.Background(), "runner-1", subj); reattached { + if reattached := hub.enroll(context.Background(), "runner-1", subj, compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED); reattached { t.Fatal("first enroll reattached = true, want false (fresh registration)") } - if reattached := hub.enroll(context.Background(), "runner-1", subj); !reattached { + if reattached := hub.enroll(context.Background(), "runner-1", subj, compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED); !reattached { t.Fatal("second enroll reattached = false, want true (single-Runner MVP re-attaches)") } // A router is resolvable after enrollment (a session command has a Runner to diff --git a/go/internal/runnerhub/provision_dedup_test.go b/go/internal/runnerhub/provision_dedup_test.go index 46f6b90cd..819cf2e53 100644 --- a/go/internal/runnerhub/provision_dedup_test.go +++ b/go/internal/runnerhub/provision_dedup_test.go @@ -52,7 +52,7 @@ type provisionOutcome struct { // Hub.Provision through the real router. func enrollAttached(t *testing.T, hub *Hub, send *recordingSend) *commandRouter { t.Helper() - hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}) + hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}, compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED) router, _, err := hub.routerFor("any") if err != nil { t.Fatalf("routerFor after enroll = %v, want the live router", err) diff --git a/go/internal/runnerhub/relay_comms_test.go b/go/internal/runnerhub/relay_comms_test.go index 222163700..540c2062d 100644 --- a/go/internal/runnerhub/relay_comms_test.go +++ b/go/internal/runnerhub/relay_comms_test.go @@ -270,7 +270,7 @@ func TestRelayCommsCallDropsBindingOnRunnerReconnect(t *testing.T) { } // The Runner reconnects (re-enroll), which drops ALL agent-comms bindings. - hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}) + hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}, compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED) // The SAME session_id now fails closed — the binding is gone, so no stale // account is reachable. @@ -330,7 +330,7 @@ func TestRelayCommsCallStoppedSessionFailsClosedNotFound(t *testing.T) { // helper. func TestProvisionThenStartBindsSessionToProvisionedAccount(t *testing.T) { hub := newHubOnly() - hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}) + hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}, compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED) router, _, err := hub.routerFor("any") if err != nil { t.Fatalf("routerFor after enroll = %v, want a router", err) @@ -377,7 +377,7 @@ func TestProvisionThenStartBindsSessionToProvisionedAccount(t *testing.T) { // RelayCommsCall fails closed CodeNotFound — never an empty-account attribution. func TestProvisionWithEmptyAccountLeavesNoBindingAndFailsClosed(t *testing.T) { hub, comms := newHubWithComms() - hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}) + hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}, compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED) router, _, err := hub.routerFor("any") if err != nil { t.Fatalf("routerFor after enroll = %v, want a router", err) @@ -439,7 +439,7 @@ func TestProvisionWithEmptyAccountLeavesNoBindingAndFailsClosed(t *testing.T) { // re-enrolls, and asserts the reverse map is empty. func TestEnrollClearsReverseAccountSessions(t *testing.T) { hub := newHubOnly() - hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}) + hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}, compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED) bindLiveSession(hub) // acct-agent -> sess-1, via the real Provision->Start path // Sanity: the reverse map is populated before the re-enroll. @@ -449,7 +449,7 @@ func TestEnrollClearsReverseAccountSessions(t *testing.T) { // A Runner reconnect: enroll re-attaches and MUST drop every stale binding, // forward AND reverse. - hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}) + hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}, compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED) if sess, ok := hub.SessionForAccount(context.Background(), "acct-agent"); ok { t.Fatalf("SessionForAccount(acct-agent) = %q, ok=true after re-enroll; want ok=false — enroll left a stale reverse entry, so a dead session resolves as live", sess) @@ -523,7 +523,7 @@ func TestEnrollFiresTerminalPresenceEdgePerBoundAccountAndClears(t *testing.T) { hub := newHubOnly() pres := &fakePresenceSink{} hub.SetPresenceSink(pres) - hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}) + hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}, compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED) hub.bindContainer("c1", "acct-a") hub.promoteSession(context.Background(), "c1", "sess-a") hub.bindContainer("c2", "acct-b") @@ -531,7 +531,7 @@ func TestEnrollFiresTerminalPresenceEdgePerBoundAccountAndClears(t *testing.T) { // A Runner reconnect: enroll drops every binding and drives each previously- // bound account OFFLINE. - hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}) + hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}, compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED) life := pres.lifecycleSnapshot() if len(life) != 2 { @@ -563,7 +563,7 @@ func TestFirstEnrollFiresNoTerminalPresenceEdge(t *testing.T) { pres := &fakePresenceSink{} hub.SetPresenceSink(pres) - hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}) + hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}, compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED) if life := pres.lifecycleSnapshot(); len(life) != 0 { t.Fatalf("lifecycle edges after first enroll = %d, want 0 (nothing was bound): %+v", len(life), life) diff --git a/go/internal/runnerhub/relay_operator_fault_test.go b/go/internal/runnerhub/relay_operator_fault_test.go index 733793c4f..51d661bfa 100644 --- a/go/internal/runnerhub/relay_operator_fault_test.go +++ b/go/internal/runnerhub/relay_operator_fault_test.go @@ -24,7 +24,7 @@ import ( // See docs/designs/infra/runtime/compass-runner-gateway-error-sentinels/design.md. func TestProvisionRelaySurfacesOperatorFaultAsFailedPrecondition(t *testing.T) { hub := newHubOnly() - hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}) + hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}, compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED) router, _, err := hub.routerFor("any") if err != nil { t.Fatalf("routerFor after enroll = %v, want a router", err) @@ -34,7 +34,7 @@ func TestProvisionRelaySurfacesOperatorFaultAsFailedPrecondition(t *testing.T) { // the socket diagnostic followed by the appended gateway.ErrOperatorConfig // sentinel text. The Runner is simulated here, so the wire carries only the // string — the gateway package is deliberately not imported. - const diag = "serving agent socket for container \"cont-op\": agent socket path \"/run/compass/containers/cont-op/agent.sock\" is 120 bytes, over the 108-byte AF_UNIX limit: shorten the socket's parent directory or the agent account id: operator-fault runner configuration" + const diag = "serving agent socket for container \"cont-op\": agent socket path \"/run/compass/containers/cont-op/agent.sock\" is 120 bytes, over the 108-byte AF_UNIX limit: shorten the Runner's --runtime-dir or the agent account id: operator-fault runner configuration" router.attach(func(cmd *compassv1internal.SessionsResponse) error { go router.complete(&compassv1internal.SessionsRequest{ diff --git a/go/internal/runnerhub/runtime_identity_test.go b/go/internal/runnerhub/runtime_identity_test.go new file mode 100644 index 000000000..9d0d70005 --- /dev/null +++ b/go/internal/runnerhub/runtime_identity_test.go @@ -0,0 +1,182 @@ +//go:build unix + +package runnerhub + +// The enrollment-carried runtime tier + egress posture reach BOTH production +// render paths. A Runner declares its tier and posture ONCE at enrollment; the +// hub stamps them onto every session status it publishes. These tests drive a +// session lifecycle frame through the REAL hub wired to the REAL Bridge board +// (board.Projection as the LifecycleSink), so one delivery exercises the two +// surfaces off one source of truth: +// - the UI's SubscribeEvents fan-out (bus.Subscribe -> Live), and +// - the CLI's GetAgentStatus snapshot (board.Snapshot). +// White-box (package runnerhub) so the test drives the unexported enroll/bind +// lifecycle directly. Sleep-free: the delivery records+fans synchronously under +// the projection's lock, so every assertion reads a settled fact. + +import ( + "context" + "testing" + "time" + + "github.com/RigelBuild/compass/go/events" + compassv1 "github.com/RigelBuild/compass/go/gen/compass/v1" + "github.com/RigelBuild/compass/go/internal/board" + "github.com/RigelBuild/compass/go/internal/store" +) + +// newHubWithBoard wires the real Bridge board as the hub's LifecycleSink over a +// fresh bus, so a delivered lifecycle frame both fans onto SubscribeEvents and +// records into the snapshot. The bus is closed at test end. +func newHubOverBoard(t *testing.T) (*Hub, *board.Projection, *events.Bus[*compassv1.SubscribeEventsResponse]) { + t.Helper() + bus := events.NewBus[*compassv1.SubscribeEventsResponse]() + t.Cleanup(bus.Close) + brd := board.NewProjection(bus) + return NewHub(brd, &fakeTailSink{}, nil, discardLogger()), brd, bus +} + +// recvSessionStatus reads one live bus event and returns its AgentSessionStatus, +// failing fast on an early close or a stall. +func recvSessionStatus(t *testing.T, ch <-chan events.Stamped[*compassv1.SubscribeEventsResponse]) *compassv1.AgentSessionStatus { + t.Helper() + select { + case e, ok := <-ch: + if !ok { + t.Fatal("live channel closed before an event arrived") + } + got := e.Payload.GetAgentSessionStatus() + if got == nil { + t.Fatalf("live event carried a non-AgentSessionStatus payload: %v", e.Payload) + } + return got + case <-time.After(testTimeout): + t.Fatal("timed out waiting for a live event") + return nil + } +} + +// identitySessionID is the one session these tests drive; the tier/posture +// stamp is Runner-wide, so a second session would assert nothing new. +const identitySessionID = "sess-1" + +// deliverWorking pushes a WORKING lifecycle frame for the test session through +// the hub. +func deliverWorking(t *testing.T, hub *Hub, seq uint64) { + t.Helper() + if err := hub.Deliver(context.Background(), RunnerEvent{ + RunnerSeq: seq, SessionID: identitySessionID, + Frame: sessionStateFrame(compassv1.AgentSessionState_AGENT_SESSION_STATE_WORKING), + }); err != nil { + t.Fatalf("Deliver(WORKING) = %v, want nil", err) + } +} + +// TestEnrolledTierAndPostureReachBothRenderPaths pins the central fix: a Runner +// enrolled as HOST/UNENFORCED has both facts stamped onto a session's status on +// BOTH surfaces — the SubscribeEvents fan-out (the UI path) and the board +// snapshot (the CLI's GetAgentStatus path) — off one delivery. +// +// Negative control: reverting deliverSession to publish {SessionId, State, +// AgentAccountId} with no tier/posture (or reverting runnerRuntimeIdentity to +// return UNSPECIFIED) reddens every assertion below — observed +// "bus event runtime_tier = RUNTIME_TIER_UNSPECIFIED, want RUNTIME_TIER_HOST". +func TestEnrolledTierAndPostureReachBothRenderPaths(t *testing.T) { + hub, brd, bus := newHubOverBoard(t) + + sub, err := bus.Subscribe(0, bus.InstanceEpoch()) + if err != nil { + t.Fatalf("Subscribe: %v", err) + } + t.Cleanup(sub.Cancel) + + hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}, + compassv1.RuntimeTier_RUNTIME_TIER_HOST, compassv1.EgressPosture_EGRESS_POSTURE_UNENFORCED) + bindSession(hub, identitySessionID) // bind AFTER enroll (enroll clears bindings) + + deliverWorking(t, hub, 1) + + // UI path: the SubscribeEvents fan-out carries the enrolled tier + posture. + got := recvSessionStatus(t, sub.Live) + if got.GetRuntimeTier() != compassv1.RuntimeTier_RUNTIME_TIER_HOST { + t.Errorf("bus event runtime_tier = %v, want RUNTIME_TIER_HOST", got.GetRuntimeTier()) + } + if got.GetEgressPosture() != compassv1.EgressPosture_EGRESS_POSTURE_UNENFORCED { + t.Errorf("bus event egress_posture = %v, want EGRESS_POSTURE_UNENFORCED", got.GetEgressPosture()) + } + + // CLI path: the board snapshot (GetAgentStatus) carries them too. + snap := brd.Snapshot(identitySessionID) + if len(snap) != 1 { + t.Fatalf("Snapshot(sess-1) = %d entries, want 1", len(snap)) + } + if snap[0].GetRuntimeTier() != compassv1.RuntimeTier_RUNTIME_TIER_HOST { + t.Errorf("snapshot runtime_tier = %v, want RUNTIME_TIER_HOST", snap[0].GetRuntimeTier()) + } + if snap[0].GetEgressPosture() != compassv1.EgressPosture_EGRESS_POSTURE_UNENFORCED { + t.Errorf("snapshot egress_posture = %v, want EGRESS_POSTURE_UNENFORCED", snap[0].GetEgressPosture()) + } +} + +// TestReEnrollUpdatesStampedTierAndPosture pins the reattach invariant: a status +// published after a Runner re-enrolls with DIFFERENT values must reflect the +// NEWLY enrolled Runner's tier + posture, never the previous Runner's. A +// re-enroll clears bindings, so the session is re-bound before the second frame. +// +// Negative control: making runnerRuntimeIdentity read a cached first-enroll value +// (or dropping the h.mu read so it races the reattach) reddens the assertion — +// observed "runtime_tier = RUNTIME_TIER_HOST, want RUNTIME_TIER_PODMAN". +func TestReEnrollUpdatesStampedTierAndPosture(t *testing.T) { + hub, brd, _ := newHubOverBoard(t) + + hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}, + compassv1.RuntimeTier_RUNTIME_TIER_HOST, compassv1.EgressPosture_EGRESS_POSTURE_UNENFORCED) + bindSession(hub, identitySessionID) + deliverWorking(t, hub, 1) + + // The Runner reconnects declaring a different tier + posture. + hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}, + compassv1.RuntimeTier_RUNTIME_TIER_PODMAN, compassv1.EgressPosture_EGRESS_POSTURE_ARMED) + bindSession(hub, identitySessionID) + deliverWorking(t, hub, 2) + + snap := brd.Snapshot(identitySessionID) + if len(snap) != 1 { + t.Fatalf("Snapshot(sess-1) = %d entries, want 1", len(snap)) + } + if snap[0].GetRuntimeTier() != compassv1.RuntimeTier_RUNTIME_TIER_PODMAN { + t.Errorf("runtime_tier = %v, want RUNTIME_TIER_PODMAN (the re-enrolled Runner's value, not the stale HOST)", snap[0].GetRuntimeTier()) + } + if snap[0].GetEgressPosture() != compassv1.EgressPosture_EGRESS_POSTURE_ARMED { + t.Errorf("egress_posture = %v, want EGRESS_POSTURE_ARMED (the re-enrolled Runner's value, not the stale UNENFORCED)", snap[0].GetEgressPosture()) + } +} + +// TestEnrollWithNoRuntimeIdentityYieldsUnspecified pins runnerRuntimeIdentity's +// fail-honest default only: a Runner that declares nothing stays UNSPECIFIED +// rather than a plausible PODMAN/ARMED, because a security surface that guesses +// is worse than one that says it does not know. Its expectation is the zero +// value, so propagation is pinned by the two tests above, not here. +// +// Negative control: defaulting runnerRuntimeIdentity to PODMAN/ARMED when the +// enrolled values are zero reddens both assertions — observed "runtime_tier = +// RUNTIME_TIER_PODMAN, want RUNTIME_TIER_UNSPECIFIED". +func TestEnrollWithNoRuntimeIdentityYieldsUnspecified(t *testing.T) { + hub, brd, _ := newHubOverBoard(t) + + hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}, + compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED) + bindSession(hub, identitySessionID) + deliverWorking(t, hub, 1) + + snap := brd.Snapshot(identitySessionID) + if len(snap) != 1 { + t.Fatalf("Snapshot(sess-1) = %d entries, want 1", len(snap)) + } + if snap[0].GetRuntimeTier() != compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED { + t.Errorf("runtime_tier = %v, want RUNTIME_TIER_UNSPECIFIED (never a guessed default)", snap[0].GetRuntimeTier()) + } + if snap[0].GetEgressPosture() != compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED { + t.Errorf("egress_posture = %v, want EGRESS_POSTURE_UNSPECIFIED (never a guessed default)", snap[0].GetEgressPosture()) + } +} diff --git a/go/internal/runnerhub/secrets_test.go b/go/internal/runnerhub/secrets_test.go index 1df0fd830..35e96512b 100644 --- a/go/internal/runnerhub/secrets_test.go +++ b/go/internal/runnerhub/secrets_test.go @@ -20,6 +20,7 @@ import ( "connectrpc.com/connect" + compassv1 "github.com/RigelBuild/compass/go/gen/compass/v1" compassv1internal "github.com/RigelBuild/compass/go/internal/gen/compass/v1" "github.com/RigelBuild/compass/go/internal/secrets" "github.com/RigelBuild/compass/go/internal/store" @@ -68,7 +69,7 @@ func runnerResolverForFetch() *fakeResolver { // can never pull the secret set. func TestFetchSecretsUnboundSessionPermissionDenied(t *testing.T) { hub := newHubOnly() - hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}) + hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}, compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED) // No session bound: the hub has no live session for "sess-unbound". resolver := &fakeResolverSecrets{set: []secrets.ResolvedSecret{{Name: "A", Value: "v"}}} url := newMountedH2CServerWithResolver(t, hub, runnerResolverForFetch().resolve, resolver) @@ -92,7 +93,7 @@ func TestFetchSecretsUnboundSessionPermissionDenied(t *testing.T) { // delivery/kind enums translated at the edge. func TestFetchSecretsBoundSessionReturnsResolvedSet(t *testing.T) { hub := newHubOnly() - hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}) + hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}, compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED) bindSession(hub, "sess-1") resolver := &fakeResolverSecrets{set: []secrets.ResolvedSecret{ {Name: "DB_URL", Value: "postgres://secret", Version: "v1", Delivery: secrets.DeliveryEnv, Kind: secrets.SecretGeneric}, @@ -126,7 +127,7 @@ func TestFetchSecretsBoundSessionReturnsResolvedSet(t *testing.T) { // swallowed as an empty set. func TestFetchSecretsResolveErrorInternal(t *testing.T) { hub := newHubOnly() - hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}) + hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}, compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED) bindSession(hub, "sess-1") resolver := &fakeResolverSecrets{resolveErr: errors.New("resolve boom")} url := newMountedH2CServerWithResolver(t, hub, runnerResolverForFetch().resolve, resolver) @@ -146,7 +147,7 @@ func TestFetchSecretsResolveErrorInternal(t *testing.T) { // window, before any session) resolves the set via the container_name selector. func TestFetchSecretsByBoundContainerReturnsResolvedSet(t *testing.T) { hub := newHubOnly() - hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}) + hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}, compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED) hub.bindContainer("cont-1", testAgentAccount) resolver := &fakeResolverSecrets{set: []secrets.ResolvedSecret{{Name: "A", Value: "v", Version: "v1"}}} url := newMountedH2CServerWithResolver(t, hub, runnerResolverForFetch().resolve, resolver) @@ -166,7 +167,7 @@ func TestFetchSecretsByBoundContainerReturnsResolvedSet(t *testing.T) { // reached — the pre-exec analogue of the unbound-session rejection. func TestFetchSecretsUnboundContainerPermissionDenied(t *testing.T) { hub := newHubOnly() - hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}) + hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}, compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED) resolver := &fakeResolverSecrets{set: []secrets.ResolvedSecret{{Name: "A", Value: "v"}}} url := newMountedH2CServerWithResolver(t, hub, runnerResolverForFetch().resolve, resolver) client := newRawRunnerClient(t, url, "runner-tok") @@ -184,7 +185,7 @@ func TestFetchSecretsUnboundContainerPermissionDenied(t *testing.T) { // is CodeInvalidArgument — a contract skew, never a silent empty set. func TestFetchSecretsMissingSelectorInvalidArgument(t *testing.T) { hub := newHubOnly() - hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}) + hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}, compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED) resolver := &fakeResolverSecrets{set: []secrets.ResolvedSecret{{Name: "A", Value: "v"}}} url := newMountedH2CServerWithResolver(t, hub, runnerResolverForFetch().resolve, resolver) client := newRawRunnerClient(t, url, "runner-tok") @@ -203,7 +204,7 @@ func TestFetchSecretsMissingSelectorInvalidArgument(t *testing.T) { // ambiguous request (CodeInvalidArgument) rather than silently preferring one. func TestFetchSecretsBothSelectorsInvalidArgument(t *testing.T) { hub := newHubOnly() - hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}) + hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}, compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED) hub.bindContainer("cont-1", testAgentAccount) bindSession(hub, "sess-1") resolver := &fakeResolverSecrets{set: []secrets.ResolvedSecret{{Name: "A", Value: "v"}}} @@ -227,7 +228,7 @@ func TestFetchSecretsBothSelectorsInvalidArgument(t *testing.T) { // without also tolerating a transient outage as "no secrets". func TestFetchSecretsNoResolverFailedPrecondition(t *testing.T) { hub := newHubOnly() - hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}) + hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}, compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED) bindSession(hub, "sess-1") url := newMountedH2CServerWithResolver(t, hub, runnerResolverForFetch().resolve, nil) client := newRawRunnerClient(t, url, "runner-tok") @@ -262,7 +263,7 @@ func TestResolvedSecretMappingRedactsValue(t *testing.T) { // content hash — an opaque counter. func TestSignalSecretsVersionPushesMonotonicToken(t *testing.T) { hub := newHubOnly() - hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}) + hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}, compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED) bindSession(hub, "sess-a") bindSession(hub, "sess-b") router, _, err := hub.routerFor("any") @@ -319,7 +320,7 @@ func TestSignalSecretsVersionPushesMonotonicToken(t *testing.T) { // notify is not an error. func TestSignalSecretsVersionNoLiveSessionsIsNoop(t *testing.T) { hub := newHubOnly() - hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}) + hub.enroll(context.Background(), "runner-1", store.Subject{Kind: store.SubjectRunner, ID: "runner-1"}, compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED) router, _, _ := hub.routerFor("any") rec := newRecordingSend() router.attach(rec.send) diff --git a/go/internal/runtime/agent.go b/go/internal/runtime/agent.go index a6fdc1550..5536adf92 100644 --- a/go/internal/runtime/agent.go +++ b/go/internal/runtime/agent.go @@ -261,10 +261,7 @@ func (r *AgentRuntime) WriteAgentFile(ctx context.Context, id WorkloadID, uid ui // can surface it per session instead of inferring containment from a green // launch. func (r *AgentRuntime) EgressPosture() EgressPosture { - if r.egressUnenforced() { - return EgressPostureUnenforced - } - return EgressPostureArmed + return PostureOf(r.runtime) } // createAndStart creates then starts the container, cleaning up a created but @@ -376,8 +373,7 @@ func (r *AgentRuntime) provision(ctx context.Context, id WorkloadID, spec AgentS } func (r *AgentRuntime) egressUnenforced() bool { - unenforcer, ok := r.runtime.(egressUnenforcer) - return ok && unenforcer.EgressUnenforced() + return PostureOf(r.runtime) == EgressPostureUnenforced } func (r *AgentRuntime) selfArmsEgress() bool { diff --git a/go/internal/runtime/tier.go b/go/internal/runtime/tier.go index 37c367012..665a68413 100644 --- a/go/internal/runtime/tier.go +++ b/go/internal/runtime/tier.go @@ -36,6 +36,18 @@ func TierOf(engine WorkloadRuntime) WorkloadTier { return "" } +// PostureOf reports how a backend constrains agent egress, probing the same +// egressUnenforcer marker AgentRuntime.EgressPosture does so the value a Runner +// declares at enrollment matches the one a live workload would report. A backend +// with no isolation boundary to firewall is unenforced; every other backend is +// armed. +func PostureOf(engine WorkloadRuntime) EgressPosture { + if unenforcer, ok := engine.(egressUnenforcer); ok && unenforcer.EgressUnenforced() { + return EgressPostureUnenforced + } + return EgressPostureArmed +} + // Tier reports the podman tier. func (p *PodmanCLI) Tier() WorkloadTier { return WorkloadTierPodman } diff --git a/proto/compass/v1/runner.proto b/proto/compass/v1/runner.proto index 2d73f9bb3..ae23664c1 100644 --- a/proto/compass/v1/runner.proto +++ b/proto/compass/v1/runner.proto @@ -190,8 +190,18 @@ service RunnerService { // CodeUnauthenticated, so the field is a defense-in-depth cross-check, not a // trusted input. The credential itself rides the transport as a bearer token, // never a field here (mirroring IssueTokenRequest, compass.proto:237-242). +// +// runtime_tier and egress_posture are declared once here, not per session: +// they are Runner-wide facts of the one backend this Runner drives, so the hub +// stamps them onto every session it owns rather than have each lifecycle frame +// repeat them. Unlike runner_id they are trusted as declared, with nothing to +// cross-check them against: only the Runner can observe which namespace an +// agent got, and it is the component that enforces egress in the first place, +// so a Runner that lies here has already lost containment either way. message EnrollRequest { string runner_id = 1; + RuntimeTier runtime_tier = 2; + EgressPosture egress_posture = 3; } // Enroll response: the handshake ack. `reattached` distinguishes a fresh