diff --git a/lib/puppet/parser/compiler.rb b/lib/puppet/parser/compiler.rb
index 9b3b5d45e..703a7799a 100644
--- a/lib/puppet/parser/compiler.rb
+++ b/lib/puppet/parser/compiler.rb
@@ -1,487 +1,489 @@
 require 'puppet/node'
 require 'puppet/resource/catalog'
 require 'puppet/util/errors'
 
 require 'puppet/resource/type_collection_helper'
 
 # Maintain a graph of scopes, along with a bunch of data
 # about the individual catalog we're compiling.
 class Puppet::Parser::Compiler
   include Puppet::Util
   include Puppet::Util::Errors
   include Puppet::Resource::TypeCollectionHelper
 
   def self.compile(node)
     # We get these from the environment and only cache them in a thread
     # variable for the duration of the compilation.  If nothing else is using
     # the thread, though, we can leave 'em hanging round with no ill effects,
     # and this is safer than cleaning them at the end and assuming that will
     # stick until the next entry to this function.
     Thread.current[:known_resource_types] = nil
     Thread.current[:env_module_directories] = nil
 
     # ...and we actually do the compile now we have caching ready.
     new(node).compile.to_resource
   rescue => detail
     message = "#{detail} on node #{node.name}"
     Puppet.log_exception(detail, message)
     raise Puppet::Error, message
  end
 
   attr_reader :node, :facts, :collections, :catalog, :node_scope, :resources, :relationships
 
   # Add a collection to the global list.
   def add_collection(coll)
     @collections << coll
   end
 
   def add_relationship(dep)
     @relationships << dep
   end
 
   # Store a resource override.
   def add_override(override)
     # If possible, merge the override in immediately.
     if resource = @catalog.resource(override.ref)
       resource.merge(override)
     else
       # Otherwise, store the override for later; these
       # get evaluated in Resource#finish.
       @resource_overrides[override.ref] << override
     end
   end
 
   # Store a resource in our resource table.
   def add_resource(scope, resource)
     @resources << resource
 
     # Note that this will fail if the resource is not unique.
     @catalog.add_resource(resource)
 
     if resource.type.to_s.downcase != "class" && resource[:stage]
       raise ArgumentError, "Only classes can set 'stage'; normal resources like #{resource} cannot change run stage"
     end
 
     # Stages should not be inside of classes.  They are always a
     # top-level container, regardless of where they appear in the
     # manifest.
     return if resource.type.to_s.downcase == "stage"
 
     # This adds a resource to the class it lexically appears in in the
     # manifest.
     if resource.type.to_s.downcase != "class"
       return @catalog.add_edge(scope.resource, resource)
     end
   end
 
   # Do we use nodes found in the code, vs. the external node sources?
   def ast_nodes?
     known_resource_types.nodes?
   end
 
   # Store the fact that we've evaluated a class
   def add_class(name)
     @catalog.add_class(name) unless name == ""
   end
 
 
   # Return a list of all of the defined classes.
   def classlist
     @catalog.classes
   end
 
   # Compiler our catalog.  This mostly revolves around finding and evaluating classes.
   # This is the main entry into our catalog.
   def compile
     # Set the client's parameters into the top scope.
     set_node_parameters
     create_settings_scope
 
     evaluate_main
 
     evaluate_ast_node
 
     evaluate_node_classes
 
     evaluate_generators
 
     finish
 
     fail_on_unevaluated
 
     @catalog
   end
 
   # LAK:FIXME There are no tests for this.
   def delete_collection(coll)
     @collections.delete(coll) if @collections.include?(coll)
   end
 
   # Return the node's environment.
   def environment
     unless defined?(@environment)
       @environment = (node.environment and node.environment != "") ? node.environment : nil
     end
     Puppet::Node::Environment.current = @environment
     @environment
   end
 
   # Evaluate all of the classes specified by the node.
   def evaluate_node_classes
     evaluate_classes(@node.classes, topscope)
   end
 
   # Evaluate each specified class in turn.  If there are any classes we can't
   # find, raise an error.  This method really just creates resource objects
   # that point back to the classes, and then the resources are themselves
   # evaluated later in the process.
   def evaluate_classes(classes, scope, lazy_evaluate = true)
     raise Puppet::DevError, "No source for scope passed to evaluate_classes" unless scope.source
     class_parameters = nil
     # if we are a param class, save the classes hash
     # and transform classes to be the keys
     if classes.class == Hash
       class_parameters = classes
       classes = classes.keys
     end
     classes.each do |name|
       # If we can find the class, then make a resource that will evaluate it.
       if klass = scope.find_hostclass(name)
 
         # If parameters are passed, then attempt to create a duplicate resource
         # so the appropriate error is thrown.
         if class_parameters
           resource = klass.ensure_in_catalog(scope, class_parameters[name] || {})
         else
           next if scope.class_scope(klass)
           resource = klass.ensure_in_catalog(scope)
         end
 
         # If they've disabled lazy evaluation (which the :include function does),
         # then evaluate our resource immediately.
         resource.evaluate unless lazy_evaluate
       else
         raise Puppet::Error, "Could not find class #{name} for #{node.name}"
       end
     end
   end
 
   def evaluate_relationships
     @relationships.each { |rel| rel.evaluate(catalog) }
   end
 
   # Return a resource by either its ref or its type and title.
   def findresource(*args)
     @catalog.resource(*args)
   end
 
   def initialize(node, options = {})
     @node = node
 
     options.each do |param, value|
       begin
         send(param.to_s + "=", value)
       rescue NoMethodError
         raise ArgumentError, "Compiler objects do not accept #{param}"
       end
     end
 
     initvars
   end
 
   # Create a new scope, with either a specified parent scope or
   # using the top scope.
   def newscope(parent, options = {})
     parent ||= topscope
     options[:compiler] = self
     scope = Puppet::Parser::Scope.new(options)
     scope.parent = parent
     scope
   end
 
   # Return any overrides for the given resource.
   def resource_overrides(resource)
     @resource_overrides[resource.ref]
   end
 
   # The top scope is usually the top-level scope, but if we're using AST nodes,
   # then it is instead the node's scope.
   def topscope
     node_scope || @topscope
   end
 
   private
 
   # If ast nodes are enabled, then see if we can find and evaluate one.
   def evaluate_ast_node
     return unless ast_nodes?
 
     # Now see if we can find the node.
     astnode = nil
     @node.names.each do |name|
       break if astnode = known_resource_types.node(name.to_s.downcase)
     end
 
     unless (astnode ||= known_resource_types.node("default"))
       raise Puppet::ParseError, "Could not find default node or by name with '#{node.names.join(", ")}'"
     end
 
     # Create a resource to model this node, and then add it to the list
     # of resources.
     resource = astnode.ensure_in_catalog(topscope)
 
     resource.evaluate
 
     # Now set the node scope appropriately, so that :topscope can
     # behave differently.
     @node_scope = topscope.class_scope(astnode)
   end
 
   # Evaluate our collections and return true if anything returned an object.
   # The 'true' is used to continue a loop, so it's important.
   def evaluate_collections
     return false if @collections.empty?
 
     found_something = false
     exceptwrap do
       # We have to iterate over a dup of the array because
       # collections can delete themselves from the list, which
       # changes its length and causes some collections to get missed.
       @collections.dup.each do |collection|
         found_something = true if collection.evaluate
       end
     end
 
     found_something
   end
 
   # Make sure all of our resources have been evaluated into native resources.
   # We return true if any resources have, so that we know to continue the
   # evaluate_generators loop.
   def evaluate_definitions
     exceptwrap do
       !unevaluated_resources.each { |resource| resource.evaluate }.empty?
     end
   end
 
   # Iterate over collections and resources until we're sure that the whole
   # compile is evaluated.  This is necessary because both collections
   # and defined resources can generate new resources, which themselves could
   # be defined resources.
   def evaluate_generators
     count = 0
     loop do
       done = true
 
       # Call collections first, then definitions.
       done = false if evaluate_collections
       done = false if evaluate_definitions
       break if done
 
       count += 1
 
       if count > 1000
         raise Puppet::ParseError, "Somehow looped more than 1000 times while evaluating host catalog"
       end
     end
   end
 
   # Find and evaluate our main object, if possible.
   def evaluate_main
     @main = known_resource_types.find_hostclass([""], "") || known_resource_types.add(Puppet::Resource::Type.new(:hostclass, ""))
     @topscope.source = @main
     @main_resource = Puppet::Parser::Resource.new("class", :main, :scope => @topscope, :source => @main)
     @topscope.resource = @main_resource
 
     add_resource(@topscope, @main_resource)
 
     @main_resource.evaluate
   end
 
   # Make sure the entire catalog is evaluated.
   def fail_on_unevaluated
     fail_on_unevaluated_overrides
     fail_on_unevaluated_resource_collections
   end
 
   # If there are any resource overrides remaining, then we could
   # not find the resource they were supposed to override, so we
   # want to throw an exception.
   def fail_on_unevaluated_overrides
     remaining = []
     @resource_overrides.each do |name, overrides|
       remaining.concat overrides
     end
 
     unless remaining.empty?
       fail Puppet::ParseError,
         "Could not find resource(s) %s for overriding" % remaining.collect { |o|
           o.ref
         }.join(", ")
     end
   end
 
   # Make sure we don't have any remaining collections that specifically
   # look for resources, because we want to consider those to be
   # parse errors.
   def fail_on_unevaluated_resource_collections
     remaining = []
     @collections.each do |coll|
       # We're only interested in the 'resource' collections,
       # which result from direct calls of 'realize'.  Anything
       # else is allowed not to return resources.
       # Collect all of them, so we have a useful error.
       if r = coll.resources
         if r.is_a?(Array)
           remaining += r
         else
           remaining << r
         end
       end
     end
 
     raise Puppet::ParseError, "Failed to realize virtual resources #{remaining.join(', ')}" unless remaining.empty?
   end
 
   # Make sure all of our resources and such have done any last work
   # necessary.
   def finish
     evaluate_relationships
 
     resources.each do |resource|
       # Add in any resource overrides.
       if overrides = resource_overrides(resource)
         overrides.each do |over|
           resource.merge(over)
         end
 
         # Remove the overrides, so that the configuration knows there
         # are none left.
         overrides.clear
       end
 
       resource.finish if resource.respond_to?(:finish)
     end
 
     add_resource_metaparams
   end
 
   def add_resource_metaparams
     unless main = catalog.resource(:class, :main)
       raise "Couldn't find main"
     end
 
     names = []
     Puppet::Type.eachmetaparam do |name|
       next if Puppet::Parser::Resource.relationship_parameter?(name)
       names << name
     end
 
     data = {}
     catalog.walk(main, :out) do |source, target|
       if source_data = data[source] || metaparams_as_data(source, names)
         # only store anything in the data hash if we've actually got
         # data
         data[source] ||= source_data
         source_data.each do |param, value|
           target[param] = value if target[param].nil?
         end
         data[target] = source_data.merge(metaparams_as_data(target, names))
       end
 
       target.tag(*(source.tags))
     end
   end
 
   def metaparams_as_data(resource, params)
     data = nil
     params.each do |param|
       unless resource[param].nil?
         # Because we could be creating a hash for every resource,
         # and we actually probably don't often have any data here at all,
         # we're optimizing a bit by only creating a hash if there's
         # any data to put in it.
         data ||= {}
         data[param] = resource[param]
       end
     end
     data
   end
 
   # Set up all of our internal variables.
   def initvars
     # The list of objects that will available for export.
     @exported_resources = {}
 
     # The list of overrides.  This is used to cache overrides on objects
     # that don't exist yet.  We store an array of each override.
     @resource_overrides = Hash.new do |overs, ref|
       overs[ref] = []
     end
 
     # The list of collections that have been created.  This is a global list,
     # but they each refer back to the scope that created them.
     @collections = []
 
     # The list of relationships to evaluate.
     @relationships = []
 
     # For maintaining the relationship between scopes and their resources.
     @catalog = Puppet::Resource::Catalog.new(@node.name)
     @catalog.version = known_resource_types.version
 
+    @catalog.environment = @node.environment.to_s
+
     # Create our initial scope and a resource that will evaluate main.
     @topscope = Puppet::Parser::Scope.new(:compiler => self)
 
     @main_stage_resource = Puppet::Parser::Resource.new("stage", :main, :scope => @topscope)
     @catalog.add_resource(@main_stage_resource)
 
     # local resource array to maintain resource ordering
     @resources = []
 
     # Make sure any external node classes are in our class list
     if @node.classes.class == Hash
       @catalog.add_class(*@node.classes.keys)
     else
       @catalog.add_class(*@node.classes)
     end
   end
 
   # Set the node's parameters into the top-scope as variables.
   def set_node_parameters
     node.parameters.each do |param, value|
       @topscope[param] = value
     end
 
     # These might be nil.
     catalog.client_version = node.parameters["clientversion"]
     catalog.server_version = node.parameters["serverversion"]
   end
 
   def create_settings_scope
     unless settings_type = environment.known_resource_types.hostclass("settings")
       settings_type = Puppet::Resource::Type.new :hostclass, "settings"
       environment.known_resource_types.add(settings_type)
     end
 
     settings_resource = Puppet::Parser::Resource.new("class", "settings", :scope => @topscope)
 
     @catalog.add_resource(settings_resource)
 
     settings_type.evaluate_code(settings_resource)
 
     scope = @topscope.class_scope(settings_type)
 
     Puppet.settings.each do |name, setting|
       next if name.to_s == "name"
       scope[name.to_s] = environment[name]
     end
   end
 
   # Return an array of all of the unevaluated resources.  These will be definitions,
   # which need to get evaluated into native resources.
   def unevaluated_resources
     # The order of these is significant for speed due to short-circuting
     resources.reject { |resource| resource.evaluated? or resource.virtual? or resource.builtin_type? }
   end
 end
diff --git a/lib/puppet/resource/catalog.rb b/lib/puppet/resource/catalog.rb
index 2c7cdd5bc..0a2c8dda3 100644
--- a/lib/puppet/resource/catalog.rb
+++ b/lib/puppet/resource/catalog.rb
@@ -1,602 +1,611 @@
 require 'puppet/node'
 require 'puppet/indirector'
 require 'puppet/simple_graph'
 require 'puppet/transaction'
 
 require 'puppet/util/pson'
 
 require 'puppet/util/tagging'
 
 # This class models a node catalog.  It is the thing
 # meant to be passed from server to client, and it contains all
 # of the information in the catalog, including the resources
 # and the relationships between them.
 class Puppet::Resource::Catalog < Puppet::SimpleGraph
   class DuplicateResourceError < Puppet::Error; end
 
   extend Puppet::Indirector
   indirects :catalog, :terminus_setting => :catalog_terminus
 
   include Puppet::Util::Tagging
   extend Puppet::Util::Pson
 
   # The host name this is a catalog for.
   attr_accessor :name
 
   # The catalog version.  Used for testing whether a catalog
   # is up to date.
   attr_accessor :version
 
   # How long this catalog took to retrieve.  Used for reporting stats.
   attr_accessor :retrieval_duration
 
   # Whether this is a host catalog, which behaves very differently.
   # In particular, reports are sent, graphs are made, and state is
   # stored in the state database.  If this is set incorrectly, then you often
   # end up in infinite loops, because catalogs are used to make things
   # that the host catalog needs.
   attr_accessor :host_config
 
   # Whether this catalog was retrieved from the cache, which affects
   # whether it is written back out again.
   attr_accessor :from_cache
 
   # Some metadata to help us compile and generally respond to the current state.
   attr_accessor :client_version, :server_version
 
+  # The environment for this catalog
+  attr_accessor :environment
+
   # Add classes to our class list.
   def add_class(*classes)
     classes.each do |klass|
       @classes << klass
     end
 
     # Add the class names as tags, too.
     tag(*classes)
   end
 
   def title_key_for_ref( ref )
     ref =~ /^([-\w:]+)\[(.*)\]$/m
     [$1, $2]
   end
 
   # Add a resource to our graph and to our resource table.
   # This is actually a relatively complicated method, because it handles multiple
   # aspects of Catalog behaviour:
   # * Add the resource to the resource table
   # * Add the resource to the resource graph
   # * Add the resource to the relationship graph
   # * Add any aliases that make sense for the resource (e.g., name != title)
   def add_resource(*resource)
     add_resource(*resource[0..-2]) if resource.length > 1
     resource = resource.pop
     raise ArgumentError, "Can only add objects that respond to :ref, not instances of #{resource.class}" unless resource.respond_to?(:ref)
     fail_on_duplicate_type_and_title(resource)
     title_key = title_key_for_ref(resource.ref)
 
     @transient_resources << resource if applying?
     @resource_table[title_key] = resource
 
     # If the name and title differ, set up an alias
 
     if resource.respond_to?(:name) and resource.respond_to?(:title) and resource.respond_to?(:isomorphic?) and resource.name != resource.title
       self.alias(resource, resource.uniqueness_key) if resource.isomorphic?
     end
 
     resource.catalog = self if resource.respond_to?(:catalog=)
     add_vertex(resource)
     @relationship_graph.add_vertex(resource) if @relationship_graph
   end
 
   # Create an alias for a resource.
   def alias(resource, key)
     resource.ref =~ /^(.+)\[/
     class_name = $1 || resource.class.name
 
     newref = [class_name, key].flatten
 
     if key.is_a? String
       ref_string = "#{class_name}[#{key}]"
       return if ref_string == resource.ref
     end
 
     # LAK:NOTE It's important that we directly compare the references,
     # because sometimes an alias is created before the resource is
     # added to the catalog, so comparing inside the below if block
     # isn't sufficient.
     if existing = @resource_table[newref]
       return if existing == resource
       resource_declaration = " at #{resource.file}:#{resource.line}" if resource.file and resource.line
       existing_declaration = " at #{existing.file}:#{existing.line}" if existing.file and existing.line
       msg = "Cannot alias #{resource.ref} to #{key.inspect}#{resource_declaration}; resource #{newref.inspect} already declared#{existing_declaration}"
       raise ArgumentError, msg
     end
     @resource_table[newref] = resource
     @aliases[resource.ref] ||= []
     @aliases[resource.ref] << newref
   end
 
   # Apply our catalog to the local host.  Valid options
   # are:
   #   :tags - set the tags that restrict what resources run
   #       during the transaction
   #   :ignoreschedules - tell the transaction to ignore schedules
   #       when determining the resources to run
   def apply(options = {})
     @applying = true
 
     Puppet::Util::Storage.load if host_config?
 
     transaction = Puppet::Transaction.new(self, options[:report])
     register_report = options[:report].nil?
 
     transaction.tags = options[:tags] if options[:tags]
     transaction.ignoreschedules = true if options[:ignoreschedules]
     transaction.for_network_device = options[:network_device]
 
     transaction.add_times :config_retrieval => self.retrieval_duration || 0
 
     begin
       Puppet::Util::Log.newdestination(transaction.report) if register_report
       begin
         transaction.evaluate
       ensure
         Puppet::Util::Log.close(transaction.report) if register_report
       end
     rescue Puppet::Error => detail
       Puppet.log_exception(detail, "Could not apply complete catalog: #{detail}")
     rescue => detail
       Puppet.log_exception(detail, "Got an uncaught exception of type #{detail.class}: #{detail}")
     ensure
       # Don't try to store state unless we're a host config
       # too recursive.
       Puppet::Util::Storage.store if host_config?
     end
 
     yield transaction if block_given?
 
     return transaction
   ensure
     @applying = false
   end
 
   # Are we in the middle of applying the catalog?
   def applying?
     @applying
   end
 
   def clear(remove_resources = true)
     super()
     # We have to do this so that the resources clean themselves up.
     @resource_table.values.each { |resource| resource.remove } if remove_resources
     @resource_table.clear
 
     if @relationship_graph
       @relationship_graph.clear
       @relationship_graph = nil
     end
   end
 
   def classes
     @classes.dup
   end
 
   # Create a new resource and register it in the catalog.
   def create_resource(type, options)
     unless klass = Puppet::Type.type(type)
       raise ArgumentError, "Unknown resource type #{type}"
     end
     return unless resource = klass.new(options)
 
     add_resource(resource)
     resource
   end
 
   # Make sure all of our resources are "finished".
   def finalize
     make_default_resources
 
     @resource_table.values.each { |resource| resource.finish }
 
     write_graph(:resources)
   end
 
   def host_config?
     host_config
   end
 
   def initialize(name = nil)
     super()
     @name = name if name
     @classes = []
     @resource_table = {}
     @transient_resources = []
     @applying = false
     @relationship_graph = nil
 
     @host_config = true
 
     @aliases = {}
 
     if block_given?
       yield(self)
       finalize
     end
   end
 
   # Make the default objects necessary for function.
   def make_default_resources
     # We have to add the resources to the catalog, or else they won't get cleaned up after
     # the transaction.
 
     # First create the default scheduling objects
     Puppet::Type.type(:schedule).mkdefaultschedules.each { |res| add_resource(res) unless resource(res.ref) }
 
     # And filebuckets
     if bucket = Puppet::Type.type(:filebucket).mkdefaultbucket
       add_resource(bucket) unless resource(bucket.ref)
     end
   end
 
   # Create a graph of all of the relationships in our catalog.
   def relationship_graph
     unless @relationship_graph
       # It's important that we assign the graph immediately, because
       # the debug messages below use the relationships in the
       # relationship graph to determine the path to the resources
       # spitting out the messages.  If this is not set,
       # then we get into an infinite loop.
       @relationship_graph = Puppet::SimpleGraph.new
 
       # First create the dependency graph
       self.vertices.each do |vertex|
         @relationship_graph.add_vertex vertex
         vertex.builddepends.each do |edge|
           @relationship_graph.add_edge(edge)
         end
       end
 
       # Lastly, add in any autorequires
       @relationship_graph.vertices.each do |vertex|
         vertex.autorequire(self).each do |edge|
           unless @relationship_graph.edge?(edge.source, edge.target) # don't let automatic relationships conflict with manual ones.
             unless @relationship_graph.edge?(edge.target, edge.source)
               vertex.debug "Autorequiring #{edge.source}"
               @relationship_graph.add_edge(edge)
             else
               vertex.debug "Skipping automatic relationship with #{(edge.source == vertex ? edge.target : edge.source)}"
             end
           end
         end
       end
       @relationship_graph.write_graph(:relationships) if host_config?
 
       # Then splice in the container information
       splice!(@relationship_graph)
 
       @relationship_graph.write_graph(:expanded_relationships) if host_config?
     end
     @relationship_graph
   end
 
   # Impose our container information on another graph by using it
   # to replace any container vertices X with a pair of verticies
   # { admissible_X and completed_X } such that that
   #
   #    0) completed_X depends on admissible_X
   #    1) contents of X each depend on admissible_X
   #    2) completed_X depends on each on the contents of X
   #    3) everything which depended on X depens on completed_X
   #    4) admissible_X depends on everything X depended on
   #    5) the containers and their edges must be removed
   #
   # Note that this requires attention to the possible case of containers
   # which contain or depend on other containers, but has the advantage
   # that the number of new edges created scales linearly with the number
   # of contained verticies regardless of how containers are related;
   # alternatives such as replacing container-edges with content-edges
   # scale as the product of the number of external dependences, which is
   # to say geometrically in the case of nested / chained containers.
   #
   Default_label = { :callback => :refresh, :event => :ALL_EVENTS }
   def splice!(other)
     stage_class      = Puppet::Type.type(:stage)
     whit_class       = Puppet::Type.type(:whit)
     component_class  = Puppet::Type.type(:component)
     containers = vertices.find_all { |v| (v.is_a?(component_class) or v.is_a?(stage_class)) and vertex?(v) }
     #
     # These two hashes comprise the aforementioned attention to the possible
     #   case of containers that contain / depend on other containers; they map
     #   containers to their sentinels but pass other verticies through.  Thus we
     #   can "do the right thing" for references to other verticies that may or
     #   may not be containers.
     #
     admissible = Hash.new { |h,k| k }
     completed  = Hash.new { |h,k| k }
     containers.each { |x|
       admissible[x] = whit_class.new(:name => "admissible_#{x.ref}", :catalog => self)
       completed[x]  = whit_class.new(:name => "completed_#{x.ref}",  :catalog => self)
     }
     #
     # Implement the six requierments listed above
     #
     containers.each { |x|
       contents = adjacent(x, :direction => :out)
       other.add_edge(admissible[x],completed[x]) if contents.empty? # (0)
       contents.each { |v|
         other.add_edge(admissible[x],admissible[v],Default_label) # (1)
         other.add_edge(completed[v], completed[x], Default_label) # (2)
       }
       # (3) & (5)
       other.adjacent(x,:direction => :in,:type => :edges).each { |e|
         other.add_edge(completed[e.source],admissible[x],e.label)
         other.remove_edge! e
       }
       # (4) & (5)
       other.adjacent(x,:direction => :out,:type => :edges).each { |e|
         other.add_edge(completed[x],admissible[e.target],e.label)
         other.remove_edge! e
       }
     }
     containers.each { |x| other.remove_vertex! x } # (5)
   end
 
   # Remove the resource from our catalog.  Notice that we also call
   # 'remove' on the resource, at least until resource classes no longer maintain
   # references to the resource instances.
   def remove_resource(*resources)
     resources.each do |resource|
       @resource_table.delete(resource.ref)
       if aliases = @aliases[resource.ref]
         aliases.each { |res_alias| @resource_table.delete(res_alias) }
         @aliases.delete(resource.ref)
       end
       remove_vertex!(resource) if vertex?(resource)
       @relationship_graph.remove_vertex!(resource) if @relationship_graph and @relationship_graph.vertex?(resource)
       resource.remove
     end
   end
 
   # Look a resource up by its reference (e.g., File[/etc/passwd]).
   def resource(type, title = nil)
     # Always create a resource reference, so that it always canonizes how we
     # are referring to them.
     if title
       res = Puppet::Resource.new(type, title)
     else
       # If they didn't provide a title, then we expect the first
       # argument to be of the form 'Class[name]', which our
       # Reference class canonizes for us.
       res = Puppet::Resource.new(nil, type)
     end
     title_key      = [res.type, res.title.to_s]
     uniqueness_key = [res.type, res.uniqueness_key].flatten
     @resource_table[title_key] || @resource_table[uniqueness_key]
   end
 
   def resource_refs
     resource_keys.collect{ |type, name| name.is_a?( String ) ? "#{type}[#{name}]" : nil}.compact
   end
 
   def resource_keys
     @resource_table.keys
   end
 
   def resources
     @resource_table.values.uniq
   end
 
   def self.from_pson(data)
     result = new(data['name'])
 
     if tags = data['tags']
       result.tag(*tags)
     end
 
     if version = data['version']
       result.version = version
     end
 
+    if environment = data['environment']
+      result.environment = environment
+    end
+
     if resources = data['resources']
       resources = PSON.parse(resources) if resources.is_a?(String)
       resources.each do |res|
         resource_from_pson(result, res)
       end
     end
 
     if edges = data['edges']
       edges = PSON.parse(edges) if edges.is_a?(String)
       edges.each do |edge|
         edge_from_pson(result, edge)
       end
     end
 
     if classes = data['classes']
       result.add_class(*classes)
     end
 
     result
   end
 
   def self.edge_from_pson(result, edge)
     # If no type information was presented, we manually find
     # the class.
     edge = Puppet::Relationship.from_pson(edge) if edge.is_a?(Hash)
     unless source = result.resource(edge.source)
       raise ArgumentError, "Could not convert from pson: Could not find relationship source #{edge.source.inspect}"
     end
     edge.source = source
 
     unless target = result.resource(edge.target)
       raise ArgumentError, "Could not convert from pson: Could not find relationship target #{edge.target.inspect}"
     end
     edge.target = target
 
     result.add_edge(edge)
   end
 
   def self.resource_from_pson(result, res)
     res = Puppet::Resource.from_pson(res) if res.is_a? Hash
     result.add_resource(res)
   end
 
   PSON.register_document_type('Catalog',self)
   def to_pson_data_hash
     {
       'document_type' => 'Catalog',
       'data'       => {
         'tags'      => tags,
         'name'      => name,
         'version'   => version,
+        'environment' => environment,
         'resources' => vertices.collect { |v| v.to_pson_data_hash },
         'edges'     => edges.   collect { |e| e.to_pson_data_hash },
         'classes'   => classes
         },
       'metadata' => {
         'api_version' => 1
         }
     }
   end
 
   def to_pson(*args)
     to_pson_data_hash.to_pson(*args)
   end
 
   # Convert our catalog into a RAL catalog.
   def to_ral
     to_catalog :to_ral
   end
 
   # Convert our catalog into a catalog of Puppet::Resource instances.
   def to_resource
     to_catalog :to_resource
   end
 
   # filter out the catalog, applying +block+ to each resource.
   # If the block result is false, the resource will
   # be kept otherwise it will be skipped
   def filter(&block)
     to_catalog :to_resource, &block
   end
 
   # Store the classes in the classfile.
   def write_class_file
     ::File.open(Puppet[:classfile], "w") do |f|
       f.puts classes.join("\n")
     end
   rescue => detail
     Puppet.err "Could not create class file #{Puppet[:classfile]}: #{detail}"
   end
 
   # Store the list of resources we manage
   def write_resource_file
     ::File.open(Puppet[:resourcefile], "w") do |f|
       to_print = resources.map do |resource|
         next unless resource.managed?
         if resource.name_var
           "#{resource.type}[#{resource[resource.name_var]}]"
         else
           "#{resource.ref.downcase}"
         end
       end.compact
       f.puts to_print.join("\n")
     end
   rescue => detail
     Puppet.err "Could not create resource file #{Puppet[:resourcefile]}: #{detail}"
   end
 
   # Produce the graph files if requested.
   def write_graph(name)
     # We only want to graph the main host catalog.
     return unless host_config?
 
     super
   end
 
   private
 
   # Verify that the given resource isn't declared elsewhere.
   def fail_on_duplicate_type_and_title(resource)
     # Short-curcuit the common case,
     return unless existing_resource = @resource_table[title_key_for_ref(resource.ref)]
 
     # If we've gotten this far, it's a real conflict
     msg = "Duplicate declaration: #{resource.ref} is already declared"
 
     msg << " in file #{existing_resource.file} at line #{existing_resource.line}" if existing_resource.file and existing_resource.line
 
     msg << "; cannot redeclare" if resource.line or resource.file
 
     raise DuplicateResourceError.new(msg)
   end
 
   # An abstracted method for converting one catalog into another type of catalog.
   # This pretty much just converts all of the resources from one class to another, using
   # a conversion method.
   def to_catalog(convert)
     result = self.class.new(self.name)
 
     result.version = self.version
+    result.environment = self.environment
 
     map = {}
     vertices.each do |resource|
       next if virtual_not_exported?(resource)
       next if block_given? and yield resource
 
       #This is hackity hack for 1094
       #Aliases aren't working in the ral catalog because the current instance of the resource
       #has a reference to the catalog being converted. . . So, give it a reference to the new one
       #problem solved. . .
       if resource.class == Puppet::Resource
         resource = resource.dup
         resource.catalog = result
       elsif resource.is_a?(Puppet::Parser::Resource)
         resource = resource.to_resource
         resource.catalog = result
       end
 
       if resource.is_a?(Puppet::Resource) and convert.to_s == "to_resource"
         newres = resource
       else
         newres = resource.send(convert)
       end
 
       # We can't guarantee that resources don't munge their names
       # (like files do with trailing slashes), so we have to keep track
       # of what a resource got converted to.
       map[resource.ref] = newres
 
       result.add_resource newres
     end
 
     message = convert.to_s.gsub "_", " "
     edges.each do |edge|
       # Skip edges between virtual resources.
       next if virtual_not_exported?(edge.source)
       next if block_given? and yield edge.source
 
       next if virtual_not_exported?(edge.target)
       next if block_given? and yield edge.target
 
       unless source = map[edge.source.ref]
         raise Puppet::DevError, "Could not find resource #{edge.source.ref} when converting #{message} resources"
       end
 
       unless target = map[edge.target.ref]
         raise Puppet::DevError, "Could not find resource #{edge.target.ref} when converting #{message} resources"
       end
 
       result.add_edge(source, target, edge.label)
     end
 
     map.clear
 
     result.add_class(*self.classes)
     result.tag(*self.tags)
 
     result
   end
 
   def virtual_not_exported?(resource)
     resource.respond_to?(:virtual?) and resource.virtual? and (resource.respond_to?(:exported?) and not resource.exported?)
   end
 end