class GolixCore(metaclass=API):
    ''' Wrapper around Golix library that automates much of the state
    management, holds the Agent's identity, etc etc.
    '''
    _librarian = weak_property('__librarian')
    DEFAULT_LEGROOM = 7
    
    # Async stuff
    _executor = readonly_property('__executor')
    _loop = readonly_property('__loop')
    
    @public_api
    def __init__(self, executor, loop, *args, **kwargs):
        ''' Create a new agent. Persister should subclass _PersisterBase
        (eventually this requirement may be changed).
        
        persister isinstance _PersisterBase
        dispatcher isinstance DispatcherBase
        _identity isinstance golix.FirstParty
        '''
        super().__init__(*args, **kwargs)
        self._mutex_request = threading.Lock()
        self._mutex_container = threading.Lock()
        self._mutex_sbinding = threading.Lock()
        self._mutex_dbinding = threading.Lock()
        self._mutex_xbinding = threading.Lock()
        
        # Added during bootstrap
        self.__identity = None
        
        # Async-specific stuff
        setattr(self, '__executor', executor)
        setattr(self, '__loop', loop)
        
    @__init__.fixture
    def __init__(self, test_agent, *args, librarian=None, **kwargs):
        ''' Just, yknow, throw in the towel. Err, agent. Whatever.
        '''
        super(GolixCore.__fixture__, self).__init__(
            executor = None,
            loop = None,
            *args,
            **kwargs
        )
        self._identity = test_agent
        
        if librarian is not None:
            self._librarian = librarian
        
    def assemble(self, librarian):
        # Chicken, meet egg.
        self._librarian = librarian
        
    def bootstrap(self, account):
        # This must be done ASAGDFP. Must be absolute first thing to bootstrap.
        self._identity = account._identity
        
    @property
    def _identity(self):
        ''' Wrap our identity in such a way that, if it's not defined,
        we can fall back on the FirstParty class.
        '''
        if self.__identity is None:
            return FirstParty
        else:
            return self.__identity
            
    @_identity.setter
    def _identity(self, identity):
        ''' Set the identity.
        '''
        self.__identity = identity
        
    @property
    @public_api
    def whoami(self):
        ''' Return the Agent's Ghid.
        '''
        return self._identity.ghid
    
    @public_api
    async def unpack_request(self, request):
        ''' Just like it says on the label...
        Note that the request is PACKED, not unpacked.
        '''
        # Run the actual function in the executor
        return (await self._loop.run_in_executor(
            self._executor,
            self._unpack_request,
            request
        ))
        
    @unpack_request.fixture
    async def unpack_request(self, request):
        ''' Bypass executor for the fixture.
        '''
        return self._unpack_request(request)
        
    def _unpack_request(self, request):
        ''' Just like it says on the label...
        Note that the request is PACKED, not unpacked.
        '''
        with self._mutex_request:
            unpacked = self._identity.unpack_request(request)
        return unpacked
        
    @public_api
    async def open_request(self, unpacked):
        ''' Just like it says on the label...
        Note that the request is UNPACKED, not packed.
        '''
        try:
            requestor = SecondParty.from_packed(
                await self._librarian.retrieve(unpacked.author)
            )
            
        except KeyError as exc:
            raise UnknownParty(
                'Request author unknown: ' + str(unpacked.author)
            ) from exc
        
        # Run the actual function in the executor
        return (await self._loop.run_in_executor(
            self._executor,
            self._open_request,
            unpacked,
            requestor
        ))
        
    @open_request.fixture
    async def open_request(self, unpacked):
        ''' Also bypass executor here.
        '''
        requestor = SecondParty.from_packed(
            await self._librarian.retrieve(unpacked.author)
        )
        return self._open_request(unpacked, requestor)
        
    def _open_request(self, unpacked, requestor):
        ''' Just like it says on the label...
        Note that the request is UNPACKED, not packed.
        '''
        return self._identity.receive_request(requestor, unpacked)
    
    @public_api
    async def make_request(self, recipient, payload):
        # Just like it says on the label...
        try:
            recipient = SecondParty.from_packed(
                await self._librarian.retrieve(recipient)
            )
        except KeyError as exc:
            raise UnknownParty(
                'Request author unknown: ' + str(recipient)
            ) from exc
        
        # Run the actual function in the executor
        return (await self._loop.run_in_executor(
            self._executor,
            self._make_request,
            recipient,
            payload
        ))
        
    @make_request.fixture
    async def make_request(self, recipient, payload):
        ''' Bypass the goddamn executor. Ffs.
        '''
        recipient = SecondParty.from_packed(
            await self._librarian.retrieve(recipient)
        )
        return self._make_request(recipient, payload)
        
    def _make_request(self, recipient, payload):
        # Just like it says on the label...
        with self._mutex_request:
            return self._identity.make_request(
                recipient = recipient,
                request = payload,
            )
    
    @public_api
    async def open_container(self, container, secret):
        author = SecondParty.from_packed(
            await self._librarian.retrieve(container.author)
        )
        
        # Wrapper around golix.FirstParty.receive_container.
        # Run the actual function in the executor
        return (await self._loop.run_in_executor(
            self._executor,
            self._open_container,
            container,
            secret,
            author
        ))
        
    @open_container.fixture
    async def open_container(self, container, secret):
        ''' Bypass executor for fixture.
        '''
        author = SecondParty.from_packed(
            await self._librarian.retrieve(container.author)
        )
        
        return self._open_container(container, secret, author)
        
    def _open_container(self, container, secret, author):
        # Wrapper around golix.FirstParty.receive_container.
        with self._mutex_container:
            return self._identity.receive_container(
                author = author,
                secret = secret,
                container = container
            )
    
    @public_api
    async def make_container(self, data, secret):
        # Simple wrapper around golix.FirstParty.make_container
        # Run the actual function in the executor
        return (await self._loop.run_in_executor(
            self._executor,
            self._make_container,
            data,
            secret
        ))
        
    @make_container.fixture
    async def make_container(self, data, secret):
        ''' Bypass executor.
        '''
        return self._make_container(data, secret)
        
    def _make_container(self, data, secret):
        # Simple wrapper around golix.FirstParty.make_container
        with self._mutex_container:
            return self._identity.make_container(
                secret = secret,
                plaintext = data
            )

    @public_api
    async def make_binding_stat(self, target):
        # Note that this requires no open() method, as bindings are verified by
        # the local persister.
        # Run the actual function in the executor
        return (await self._loop.run_in_executor(
            self._executor,
            self._make_binding_stat,
            target
        ))
        
    @make_binding_stat.fixture
    async def make_binding_stat(self, target):
        ''' Skip the executor for the fixture.
        '''
        return self._make_binding_stat(target)

    def _make_binding_stat(self, target):
        # Note that this requires no open() method, as bindings are verified by
        # the local persister.
        with self._mutex_sbinding:
            return self._identity.make_bind_static(target)
    
    @public_api
    async def make_binding_dyn(self, target, ghid=None, history=None):
        ''' Make a new dynamic binding frame.
        If supplied, ghid is the dynamic address, and history is an
        ordered iterable of the previous frame ghids.
        '''
        # Either ghid AND history must be defined, XOR ghid AND history must be
        # undefined.
        if bool(ghid) ^ bool(history):
            raise ValueError('Mixed def of ghid/history while dyn binding.')
        
        if ghid:
            current_binding = await self._librarian.summarize(ghid)
            counter = current_binding.counter + 1
        else:
            counter = 0
            
        # Run the actual function in the executor
        return (await self._loop.run_in_executor(
            self._executor,
            self._make_binding_dyn,
            ghid,
            counter,
            target,
            history
        ))
        
    @make_binding_dyn.fixture
    async def make_binding_dyn(self, target, ghid=None, history=None):
        ''' Yep, just bypass the executor
        '''
        if ghid:
            current_binding = await self._librarian.summarize(ghid)
            counter = current_binding.counter + 1
        else:
            counter = 0
        
        return self._make_binding_dyn(ghid, counter, target, history)
        
    def _make_binding_dyn(self, ghid, counter, target, history):
        ''' Make a new dynamic binding frame.
        If supplied, ghid is the dynamic address, and history is an
        ordered iterable of the previous frame ghids.
        '''
        if history is not None:
            target_vector = [target, *history]
        else:
            target_vector = [target]
        
        with self._mutex_dbinding:
            return self._identity.make_bind_dynamic(
                counter = counter,
                target_vector = target_vector,
                ghid_dynamic = ghid
            )
    
    @public_api
    async def make_debinding(self, target):
        # Simple wrapper around golix.FirstParty.make_debind
        # Run the actual function in the executor
        return (await self._loop.run_in_executor(
            self._executor,
            self._make_debinding,
            target
        ))
        
    @make_debinding.fixture
    async def make_debinding(self, target):
        ''' Bypass executor for the fixture.
        '''
        return self._make_debinding(target)
        
    def _make_debinding(self, target):
        # Simple wrapper around golix.FirstParty.make_debind
        with self._mutex_xbinding:
            return self._identity.make_debind(target) class Oracle(metaclass=API):
    ''' Source for total internal truth and state tracking of objects.
    
    Maintains <ghid>: <obj> lookup. Used by dispatchers to track obj
    state. Might eventually be used by AgentBase. Just a quick way to
    store and retrieve any objects based on an associated ghid.
    '''
    # These are actually used by the oracle itself
    _salmonator = weak_property('__salmonator')
    
    # These are only here to pass along to GAOs
    _golcore = weak_property('__golcore')
    _ghidproxy = weak_property('__ghidproxy')
    _privateer = weak_property('__privateer')
    _percore = weak_property('__percore')
    _librarian = weak_property('__librarian')
    
    def __init__(self, *args, **kwargs):
        ''' Sets up internal tracking.
        '''
        super().__init__(*args, **kwargs)
        self._lookup = {}
        
    @fixture_api
    def RESET(self):
        ''' Simply re-call init.
        '''
        self._lookup.clear()
        
    def assemble(self, golcore, ghidproxy, privateer, percore, librarian,
                 salmonator):
        # Chicken, meet egg.
        self._golcore = golcore
        self._ghidproxy = ghidproxy
        self._privateer = privateer
        self._percore = percore
        self._librarian = librarian
        self._salmonator = salmonator
        
    @fixture_api
    def add_object(self, ghid, obj):
        ''' Add an object to the fixture.
        '''
        self._lookup[ghid] = obj
            
    @public_api
    async def get_object(self, gaoclass, ghid, *args, **kwargs):
        ''' Get an object.
        '''
        if ghid in self._lookup:
            # TODO: this is bad, because we're suppresing any potential argsig
            # problems. We should fix the abstraction so that it doesn't break
            # assumptions like that.
            obj = self._lookup[ghid]
            logger.debug(
                'GAO ' + str(ghid) + ' already exists in Oracle memory as: ' +
                str(type(obj))
            )
            
            if not isinstance(obj, gaoclass):
                raise TypeError(
                    'Object has already been resolved, and is not the '
                    'correct GAO class.'
                )
            else:
                logger.debug(
                    'GAO ' + str(ghid) + ' acceptable for ' + str(gaoclass)
                )
                
            await obj._ctx.wait()
            if ghid not in self._lookup:
                raise RuntimeError('Contentious delete while getting object.')
        
        else:
            logger.info(
                'GAO ' + str(ghid) +
                ' not currently in Oracle memory. Attempting load as: ' +
                str(gaoclass)
            )
            
            # First create the actual GAO. We do not need to have the ghid
            # downloaded to do this -- object creation is just making a Python
            # object locally.
            obj = gaoclass(
                ghid,
                None,   # dynamic
                None,   # author
                7,      # legroom (will be overwritten by pull)
                *args,
                golcore = self._golcore,
                ghidproxy = self._ghidproxy,
                privateer = self._privateer,
                percore = self._percore,
                librarian = self._librarian,
                **kwargs
            )
            
            # TODO: fix leaky abstraction
            obj._ctx = asyncio.Event()
            
            # Always do this first to make sure we have the most recent version
            # in all subsequent calls, without a race condition.
            self._lookup[ghid] = obj
            
            try:
                # Now immediately subscribe to the object upstream, so that there
                # is no race condition getting updates
                await self._salmonator.register(ghid)
                # Add deregister as a finalizer, but don't call it atexit. TODO:
                # fix leaky abstraction
                finalizer = weakref.finalize(obj, self._salmonator._deregister,
                                             ghid)
                finalizer.atexit = False
                # Explicitly pull the object from salmonator to ensure we have
                # the newest version, and that it is available locally in
                # librarian if also available anywhere else. Note that
                # salmonator handles modal switching for dynamic/static. It
                # will also (by default, with skip_refresh=False) pull in any
                # updates that have accumulated in the meantime.
                await self._salmonator.attempt_pull(ghid, quiet=True)
                
                # Now actually fetch the object. This may KeyError if the
                # ghid is still unknown.
                await obj._pull()
            
            # Got an exception? Revert the lookup and reraise
            except Exception:
                del self._lookup[ghid]
                raise
                
            # We have to release other waiters regardless
            finally:
                # Finally, let future callers unblock
                obj._ctx.set()
            
        return obj
        
    @get_object.fixture
    async def get_object(self, gaoclass, ghid, *args, **kwargs):
        ''' Do the easy thing and just pull it out of lookup.
        '''
        # We should also check argsig while we're at it
        signature = inspect.Signature.from_callable(gaoclass)
        signature.bind(
            ghid,
            None,   # dynamic
            None,   # author
            7,      # legroom (will be overwritten by pull)
            *args,
            golcore = self,     # It's not going to be used and may not exist
            ghidproxy = self,   # It's not going to be used and may not exist
            privateer = self,   # It's not going to be used and may not exist
            percore = self,     # It's not going to be used and may not exist
            librarian = self,   # It's not going to be used and may not exist
            **kwargs
        )
            
        return self._lookup[ghid]
        
    @public_api
    async def new_object(self, gaoclass, dynamic, legroom, *args, **kwargs):
        ''' Creates a new object and returns it. Passes all *kwargs to
        the declared gao_class. Requires a zeroth state, and calls push
        internally.
        '''
        obj = gaoclass(
            None,                   # ghid
            dynamic,
            self._golcore.whoami,   # author
            legroom,
            *args,
            golcore = self._golcore,
            ghidproxy = self._ghidproxy,
            privateer = self._privateer,
            percore = self._percore,
            librarian = self._librarian,
            **kwargs
        )
        await obj._push()
        
        # This is used strictly by oracle. TODO: fix leaky abstraction
        obj._ctx = asyncio.Event()
        obj._ctx.set()
        # Do this before registering with salmonator, in case the latter errors
        self._lookup[obj.ghid] = obj
        
        # Finally, register to receive any concurrent updates from other
        # simultaneous sessions, and then return the object
        await self._salmonator.register(obj.ghid)
        # Add deregister as a finalizer, but don't call it atexit. TODO: fix
        # leaky abstraction
        finalizer = weakref.finalize(obj, self._salmonator._deregister,
                                     obj.ghid)
        finalizer.atexit = False
        
        return obj
            
    @new_object.fixture
    async def new_object(self, gaoclass, dynamic, legroom, *args, **kwargs):
        ''' Relies upon add_object, but otherwise just pops something
        from the lookup.
        '''
        # We should also check argsig while we're at it
        signature = inspect.Signature.from_callable(gaoclass)
        signature.bind(
            None,               # ghid
            dynamic,
            None,               # author (normally assigned through whoami)
            legroom,
            *args,
            golcore = self,     # It's not going to be used and may not exist
            ghidproxy = self,   # It's not going to be used and may not exist
            privateer = self,   # It's not going to be used and may not exist
            percore = self,     # It's not going to be used and may not exist
            librarian = self,   # It's not going to be used and may not exist
            **kwargs
        )
        
        ghid, obj = self._lookup.popitem()
        self._lookup[ghid] = obj
        return obj
        
    def forget(self, ghid):
        ''' Removes the object from the cache. Next time an application
        wants it, it will need to be acquired from persisters.
        
        Indempotent; will not raise KeyError if called more than once.
        '''
        try:
            del self._lookup[ghid]
        except KeyError:
            logger.debug(str(ghid) + ' unknown to oracle.')
            
    def __contains__(self, ghid):
        ''' Checks for the ghid in cache (but does not check for global
        availability; that would require checking the persister for its
        existence and the privateer for access).
        '''
        return ghid in self._lookup class GhidProxier(metaclass=API):
    ''' Resolve the base container GHID from any associated ghid. Uses
    all weak references, so should not interfere with GCing objects.
    '''
    # Note that we can't really cache aliases, because their proxies will
    # not update when we change things unless the proxy is also removed
    # from the cache. Since the objects may (or may not) exist locally in
    # memory anyways, we should just take advantage of that, and allow our
    # inquisitor to more easily manage memory consumption as well.
    _librarian = weak_property('__librarian')
    
    @fixture_api
    def __init__(self, *args, **kwargs):
        ''' Add in a dict to store resolutions.
        '''
        super(GhidProxier.__fixture__, self).__init__(*args, **kwargs)
        self.lookup = {}
        
    def assemble(self, librarian):
        # Chicken, meet egg.
        self._librarian = librarian
        
    def __mklink(self, proxy, target):
        ''' Set, or update, a ghid proxy.
        
        Ghids must only ever have a single proxy. Calling chain on an 
        existing proxy will update the target.
        '''
        raise NotImplementedError('Explicit link creation has been removed.')
        
        if not isinstance(proxy, Ghid):
            raise TypeError('Proxy must be Ghid.')
            
        if not isinstance(target, Ghid):
            raise TypeError('Target must be ghid.')
        
        with self._modlock:
            self._refs[proxy] = target
    
    @public_api
    async def resolve(self, ghid):
        ''' Protect the entry point with a global lock, but don't leave
        the recursive bit alone.
        
        TODO: make this guarantee, through using the persister's
        librarian, that the resolved ghid IS, in fact, a container.
        
        TODO: consider adding a depth limit to resolution.
        '''
        if not isinstance(ghid, Ghid):
            raise TypeError('Can only resolve a ghid.')
            
        return (await self._resolve(ghid))
        
    async def _resolve(self, ghid):
        ''' Recursively resolves the container ghid for a proxy (or a
        container).
        '''
        try:
            obj = await self._librarian.summarize(ghid)
        
        # TODO: make this an error?
        except KeyError:
            logger.warning(''.join((
                'GAO ',
                str(ghid),
                ' address resolver failed to verify: missing at librarian.\n',
                traceback.format_exc()
            )))
            
            result = ghid
        
        else:
            if isinstance(obj, _GeocLite):
                result = ghid
                
            else:
                result = await self._resolve(obj.target)
                
        return result
        
    @resolve.fixture
    async def resolve(self, ghid):
        ''' Ehhhh, okay. So we're going to fixture this, mostly for
        privateer, in a way that just returns the ghid immediately.
        '''
        if ghid in self.lookup:
            return self.lookup[ghid]
        else:
            return ghid class GAO(GAOCore):
    ''' A bare-bones GAO that performs trivial serialization (the
    serialization of the state is the state itself).
    '''
    
    def __init__(self, *args, state, **kwargs):
        super().__init__(*args, **kwargs)
        self.state = state
        
    async def pack_gao(self):
        ''' Packs self into a bytes object. May be overwritten in subs
        to pack more complex objects. Should always be a staticmethod or
        classmethod.
        
        May be used to implement, for example, packing self into a
        DispatchableState, etc etc.
        '''
        return self.state
        
    async def unpack_gao(self, packed):
        ''' Unpacks state from a bytes object and applies state to self.
        May be overwritten in subs to unpack more complex objects.
        
        May be used to implement, for example, dicts performing a
        clear() operation before an update() instead of just reassigning
        the object.
        '''
        self.state = packed
        
    def __eq__(self, other):
        ''' Ensure the other object is the same gao with the same state.
        '''
        equal = True
        
        try:
            equal &= (self.dynamic == other.dynamic)
            equal &= (self.ghid == other.ghid)
            equal &= (self.author == other.author)
            equal &= (self.state == other.state)
            
        except AttributeError:
            equal = False
        
        return equal class GAOSet(_GAOPickleBase, _SetMixin):
    ''' Combine GAO sets with pickle serialization.
    '''
    pass class GAODict(_GAOPickleBase, _DictMixin):
    ''' Combine GAO dicts with pickle serialization.
    '''
    pass class GAOSetMap(_GAOPickleBase, _SetMapMixin):
    ''' Combine GAO setmaps with pickle serialization.
    '''
    pass class PersistenceCore(metaclass=API):
    ''' Provides the core functions for storing Golix objects. Required
    for the hypergolix service to start.
    
    Can coordinate with both "upstream" and "downstream" persisters.
    Other persisters should pass through the "ingestive tract". Local
    objects can be published directly through calling the ingest_<type>
    methods.
    
    TODO: add librarian validation, so that attempting to update an
    object we already have an identical copy to silently exits.
    '''
    _doorman = weak_property('__doorman')
    _enforcer = weak_property('__enforcer')
    _lawyer = weak_property('__lawyer')
    _bookie = weak_property('__bookie')
    _postman = weak_property('__postman')
    _undertaker = weak_property('__undertaker')
    _librarian = weak_property('__librarian')
    _salmonator = weak_property('__salmonator')
    
    # This is a messy way of getting the suffix for validation and stuff but
    # it's getting the job done.
    _ATTR_LOOKUP = {
        _GidcLite: 'gidc',
        _GeocLite: 'geoc',
        _GobsLite: 'gobs',
        _GobdLite: 'gobd',
        _GdxxLite: 'gdxx',
        _GarqLite: 'garq'
    }
    
    @public_api
    def __init__(self, loop=None, *args, **kwargs):
        ''' Create a KeyedAsyncioLock for ingestion.
        '''
        self._ingestion_mutex = KeyedAsyncioLock(loop)
        
    @__init__.fixture
    def __init__(self, librarian=None, *args, **kwargs):
        ''' Right now, well, this is basically a noop. Anticipating
        substantial changes!
        '''
        super(PersistenceCore.__fixture__, self).__init__(
            *args,
            **kwargs
        )
        
        if librarian is not None:
            self._librarian = librarian
        
    def assemble(self, doorman, enforcer, lawyer, bookie, librarian, postman,
                 undertaker, salmonator):
        self._doorman = doorman
        self._enforcer = enforcer
        self._lawyer = lawyer
        self._bookie = bookie
        self._postman = postman
        self._undertaker = undertaker
        self._librarian = librarian
        self._salmonator = salmonator
    
    @public_api
    async def direct_ingest(self, obj, packed, remotable, skip_conn=None):
        ''' Standard ingestion flow for stuff. To be called from ingest
        above, or directly (for objects created "in-house").
        '''
        # Check for a redundant object, which will immediately short-circuit.
        if isinstance(obj, _GobdLite):
            check_ghid = obj.frame_ghid
            log_frame = True
            counter = obj.counter
            target = obj.target
        else:
            check_ghid = obj.ghid
            log_frame = False
            counter = 0
            target = None
        
        async with self._ingestion_mutex(check_ghid):
            if (await self._librarian.contains(check_ghid)):
                logger.debug(
                    str(check_ghid) + ' not ingested: already exists.'
                )
                return False
            
            else:
                logger.info(
                    'Ingesting ' + str(obj) +
                    (
                        ' frame ' + str(counter) + ': ' + str(check_ghid) +
                        ', target: ' + str(target)
                    ) * log_frame + '...'
                )
                # Calculate "gidc", etc
                suffix = self._ATTR_LOOKUP[type(obj)]
                validation_method = 'validate_' + suffix
                
                # Validate the object... (will raise for invalid)
                # ########################
                # Enforce target selection
                await getattr(self._enforcer, validation_method)(obj)
                # Now make sure authorship requirements are satisfied
                await getattr(self._lawyer, validation_method)(obj)
                # Finally make sure persistence rules are followed
                await getattr(self._bookie, validation_method)(obj)
                
                # Ingest the object
                # ########################
                # Alert the undertaker for any necessary GC of targets, etc. Do
                # that before storing at the librarian, so that the undertaker
                # has access to the old state.
                await getattr(self._undertaker, 'alert_' + suffix)(obj,
                                                                   skip_conn)
                # Finally, add it to the librarian.
                await self._librarian.store(obj, packed)
                
                if remotable:
                    await self._salmonator.push(obj.ghid)
            
                return True
    
    @direct_ingest.fixture
    async def direct_ingest(self, obj, packed, remotable, skip_conn=None):
        ''' Fixture direct ingest to just store at librarian directly
        with no verification (if librarian exists), otherwise simply
        return None.
        '''
        try:
            librarian = self._librarian
        except AttributeError:
            return None
        else:
            await librarian.store(obj, packed)
    
    @public_api
    async def attempt_load(self, packed, quiet=True):
        ''' Attempt to load a packed golix object. If quiet=False, it
        will raise if there is no loader.
        
        TODO: move into doorman.
        '''
        # This is kinda silly, but instead of spewing off a million different
        # threads that will immediately die, let's pre-select the loader
        magic = bytes(packed[:4])
        loaders = {
            b'GIDC': self._doorman.load_gidc,
            b'GEOC': self._doorman.load_geoc,
            b'GOBS': self._doorman.load_gobs,
            b'GOBD': self._doorman.load_gobd,
            b'GDXX': self._doorman.load_gdxx,
            b'GARQ': self._doorman.load_garq
        }
        
        try:
            loader = loaders[magic]
            
        # If no successful loader was found, return None, and allow the parent
        # to raise.
        except KeyError as exc:
            if quiet:
                return None
            else:
                raise MalformedGolixPrimitive('No loader found for magic: ' +
                                              str(magic)) from exc
        
        obj = await loader(packed)
            
        return obj
        
    @attempt_load.fixture
    async def attempt_load(self, packed):
        ''' Create an ad-hoc doorman fixture.
        '''
        # Note that, because of the weak ref, we need to actually hold this
        # here to prevent the doorman from being gc'd. So, just create a
        # disposable one.
        doorman = Doorman.__fixture__()
        self._doorman = doorman
        result = \
            await super(PersistenceCore.__fixture__, self).attempt_load(packed)
        del doorman
        return result
    
    @fixture_noop
    @public_api
    async def ingest(self, packed, remotable=True, skip_conn=None):
        ''' Called on an untrusted and unknown object. May be bypassed
        by locally-created, trusted objects (by calling the individual
        ingest methods directly). Parses, validates, and stores the
        object, and returns True; or, raises an error.
        '''
        # This may return None, but that will be caught by the KeyError below.
        obj = await self.attempt_load(packed)
        if obj is None:
            raise MalformedGolixPrimitive(
                'Packed bytes do not appear to be a Golix primitive.'
            )
        
        ingested = await self.direct_ingest(obj, packed, remotable)
        
        # Note that individual ingest methods are only called directly for
        # locally-built objects, which do not need a mail run.
        # If the object is identical to what we already have, the ingester
        # will return None.
        if ingested:
            await self._postman.schedule(obj, skip_conn=skip_conn)
            
        else:
            logger.debug(str(obj.ghid) + ' postal scheduling aborted on ' +
                         'unchanged object.')
        
        # Note: this is not the place for salmonator pushing! Locally
        # created/updated objects call the individual ingest methods
        # directly, so they have to be the ones that actually deal with it.
        
        return ingested class Doorman(metaclass=API):
    ''' Parses files and enforces crypto. Can be bypassed for trusted
    (aka locally-created) objects. Only called from within the typeless
    PersisterCore.ingest() method.
    '''
    _librarian = weak_property('__librarian')
    
    # Async stuff
    _executor = readonly_property('__executor')
    _loop = readonly_property('__loop')
    
    @public_api
    def __init__(self, executor, loop, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self._golix = ThirdParty()
        
        # These coordinate the threads in the executor to bolt-on thread safety
        # to the un-thread-safe smartyparse stuff.
        self._parselock_gidc = threading.Lock()
        self._parselock_geoc = threading.Lock()
        self._parselock_gobs = threading.Lock()
        self._parselock_gobd = threading.Lock()
        self._parselock_gdxx = threading.Lock()
        self._parselock_garq = threading.Lock()
        
        # Async-specific stuff
        setattr(self, '__executor', executor)
        setattr(self, '__loop', loop)
        
    @__init__.fixture
    def __init__(self, *args, **kwargs):
        super(Doorman.__fixture__, self).__init__(
            executor = None,
            loop = None
        )
        
    def assemble(self, librarian):
        # Called to link to the librarian.
        self._librarian = librarian
            
    def _verify_golix(self, obj, author):
        ''' Performs golix verification of the object. Meant to be
        called from within the executor.
        '''
        try:
            self._golix.verify_object(
                second_party = author.identity,
                obj = obj,
            )
        except SecurityError as exc:
            raise VerificationFailure(str(obj)) from exc
        
    @public_api
    async def load_gidc(self, packed):
        # Run the actual loader in the executor
        obj = await self._loop.run_in_executor(
            self._executor,
            self._load_gidc,
            packed
        )
            
        # No further verification required.
        return _GidcLite.from_golix(obj)
        
    @load_gidc.fixture
    async def load_gidc(self, packed):
        ''' Bypass the executor.
        '''
        return _GidcLite.from_golix(self._load_gidc(packed))
        
    def _load_gidc(self, packed):
        ''' Performs the actual loading.
        '''
        try:
            with self._parselock_gidc:
                return GIDC.unpack(packed)
        
        except Exception as exc:
            raise MalformedGolixPrimitive('Invalid formatting for GIDC ' +
                                          'object.') from exc
    
    @public_api
    async def load_geoc(self, packed):
        # Run the actual loader in the executor
        obj = await self._loop.run_in_executor(
            self._executor,
            self._load_geoc,
            packed
        )
            
        # Okay, now we need to verify the object
        try:
            author = await self._librarian.summarize(obj.author)
        
        except KeyError as exc:
            raise InvalidIdentity('Unknown author: ' +
                                  str(obj.author)) from exc
        
        # Finally, verify the Golix signature
        await self._loop.run_in_executor(
            self._executor,
            self._verify_golix,
            obj,
            author
        )
            
        return _GeocLite.from_golix(obj)
        
    @load_geoc.fixture
    async def load_geoc(self, packed):
        ''' Bypass the executor.
        '''
        return _GeocLite.from_golix(self._load_geoc(packed))
        
    def _load_geoc(self, packed):
        ''' Performs the actual loading.
        '''
        try:
            with self._parselock_geoc:
                return GEOC.unpack(packed)
        
        except Exception as exc:
            raise MalformedGolixPrimitive('Invalid formatting for GEOC ' +
                                          'object.') from exc
    
    @public_api
    async def load_gobs(self, packed):
        # Run the actual loader in the executor
        obj = await self._loop.run_in_executor(
            self._executor,
            self._load_gobs,
            packed
        )
            
        # Okay, now we need to verify the object
        try:
            author = await self._librarian.summarize(obj.binder)
        except KeyError as exc:
            raise InvalidIdentity('Unknown author: ' +
                                  str(obj.binder)) from exc
        
        # Finally, verify the Golix signature
        await self._loop.run_in_executor(
            self._executor,
            self._verify_golix,
            obj,
            author
        )
            
        return _GobsLite.from_golix(obj)
        
    @load_gobs.fixture
    async def load_gobs(self, packed):
        ''' Bypass the executor.
        '''
        return _GobsLite.from_golix(self._load_gobs(packed))
        
    def _load_gobs(self, packed):
        ''' Performs the actual loading.
        '''
        try:
            with self._parselock_gobs:
                return GOBS.unpack(packed)
        
        except Exception as exc:
            raise MalformedGolixPrimitive('Invalid formatting for GOBS ' +
                                          'object.') from exc
    
    @public_api
    async def load_gobd(self, packed):
        # Run the actual loader in the executor
        obj = await self._loop.run_in_executor(
            self._executor,
            self._load_gobd,
            packed
        )
            
        # Okay, now we need to verify the object
        try:
            author = await self._librarian.summarize(obj.binder)
        except KeyError as exc:
            raise InvalidIdentity('Unknown author: ' +
                                  str(obj.binder)) from exc
        
        # Finally, verify the Golix signature
        await self._loop.run_in_executor(
            self._executor,
            self._verify_golix,
            obj,
            author
        )
            
        return _GobdLite.from_golix(obj)
        
    @load_gobd.fixture
    async def load_gobd(self, packed):
        ''' Bypass the executor.
        '''
        return _GobdLite.from_golix(self._load_gobd(packed))
        
    def _load_gobd(self, packed):
        ''' Performs the actual loading.
        '''
        try:
            with self._parselock_gobd:
                return GOBD.unpack(packed)
        
        except Exception as exc:
            raise MalformedGolixPrimitive('Invalid formatting for GOBD '
                                          'object.') from exc
    
    @public_api
    async def load_gdxx(self, packed):
        # Run the actual loader in the executor
        obj = await self._loop.run_in_executor(
            self._executor,
            self._load_gdxx,
            packed
        )
            
        # Okay, now we need to verify the object
        try:
            author = await self._librarian.summarize(obj.debinder)
        except KeyError as exc:
            raise InvalidIdentity('Unknown author: ' +
                                  str(obj.debinder)) from exc
        
        # Finally, verify the Golix signature
        await self._loop.run_in_executor(
            self._executor,
            self._verify_golix,
            obj,
            author
        )
            
        return _GdxxLite.from_golix(obj)
        
    @load_gdxx.fixture
    async def load_gdxx(self, packed):
        ''' Bypass the executor.
        '''
        return _GdxxLite.from_golix(self._load_gdxx(packed))
        
    def _load_gdxx(self, packed):
        ''' Performs the actual loading.
        '''
        try:
            with self._parselock_gdxx:
                return GDXX.unpack(packed)
        
        except Exception as exc:
            raise MalformedGolixPrimitive('Invalid formatting for GDXX ' +
                                          'object.') from exc
    
    @public_api
    async def load_garq(self, packed):
        # Run the actual loader in the executor
        obj = await self._loop.run_in_executor(
            self._executor,
            self._load_garq,
            packed
        )
            
        # Persisters cannot further verify the object.
        return _GarqLite.from_golix(obj)
        
    @load_garq.fixture
    async def load_garq(self, packed):
        ''' Bypass the executor.
        '''
        return _GarqLite.from_golix(self._load_garq(packed))
        
    def _load_garq(self, packed):
        ''' Performs the actual loading.
        '''
        try:
            with self._parselock_garq:
                return GARQ.unpack(packed)
        
        except Exception as exc:
            raise MalformedGolixPrimitive('Invalid formatting for GARQ ' +
                                          'object.') from exc class Enforcer(metaclass=API):
    ''' Enforces valid target selections.
    '''
    _librarian = weak_property('__librarian')
    
    @fixture_api
    def __init__(self, librarian, *args, **kwargs):
        super(Enforcer.__fixture__, self).__init__(*args, **kwargs)
        self._librarian = librarian
        
    def assemble(self, librarian):
        # Call before using.
        self._librarian = librarian
        
    async def validate_gidc(self, obj):
        ''' GIDC need no target verification.
        '''
        return True
        
    async def validate_geoc(self, obj):
        ''' GEOC need no target validation.
        '''
        return True
        
    async def validate_gobs(self, obj):
        ''' Check if target is known, and if it is, validate it.
        '''
        try:
            target = await self._librarian.summarize(obj.target)
        # TODO: think more about this, and whether everything has been updated
        # appropriately to raise a DoesNotExist instead of a KeyError.
        # This could be more specific and say DoesNotExist
        except KeyError:
            logger.debug(str(obj) + ' target missing from librarian: ' +
                         str(obj.target))
        else:
            for forbidden in (_GidcLite, _GobsLite, _GdxxLite, _GarqLite):
                if isinstance(target, forbidden):
                    raise InvalidTarget(str(obj) + ' target invalid: ' +
                                        str(target))
        return True
        
    async def validate_gobd(self, obj):
        ''' Check if target is known, and if it is, validate it.
        
        Also do a state check on the dynamic binding.
        '''
        try:
            target = await self._librarian.summarize(obj.target)
        except KeyError:
            logger.debug(str(obj) + ' target missing from librarian: ' +
                         str(obj.target))
        else:
            for forbidden in (_GidcLite, _GobsLite, _GdxxLite, _GarqLite):
                if isinstance(target, forbidden):
                    raise InvalidTarget(str(obj) + ' target invalid: ' +
                                        str(target))
                    
        await self._validate_dynamic_history(obj)
                    
        return True
        
    async def validate_gdxx(self, obj, target_obj=None):
        ''' Check if target is known, and if it is, validate it.
        '''
        try:
            if target_obj is None:
                target = await self._librarian.summarize(obj.target)
            else:
                target = target_obj
        except KeyError:
            logger.warning(str(obj) + ' validated by Enforcer, but its ' +
                           'target was unknown: ' + str(obj.target))
            logger.debug(str(obj) + ' missing target traceback:\n' +
                         ''.join(traceback.format_exc()))
            
        else:
            # NOTE: if this changes, will need to modify place_gdxx in _Bookie
            for forbidden in (_GidcLite, _GeocLite):
                if isinstance(target, forbidden):
                    raise InvalidTarget(str(obj) + ' target invalid: ' +
                                        str(target))
        return True
        
    async def validate_garq(self, obj):
        ''' No additional validation needed.
        '''
        return True
        
    async def _validate_dynamic_history(self, obj):
        ''' Enforces state flow / progression for dynamic objects. In
        other words, ensures monotonic counter.
        '''
        # Try getting an existing binding.
        try:
            existing = await self._librarian.summarize(obj.ghid)
        
        # TOFU (trust on first upload lulz)
        except KeyError:
            pass
                
        else:
            if existing.counter >= obj.counter:
                logger.debug('New obj frame:     ' +
                             str(obj.frame_ghid))
                logger.debug('New obj tarvec:    ' +
                             str(obj.target_vector))
                logger.debug('Existing frame:    ' +
                             str(existing.frame_ghid))
                logger.debug('Existing tarvec:   ' +
                             str(existing.target_vector))
                raise IllegalDynamicFrame(str(obj.ghid) + ' counter ' +
                                          'decreased from ' +
                                          str(existing.counter) + ' to ' +
                                          str(obj.counter)) class Bookie(metaclass=API):
    ''' Tracks state relationships between objects using **only weak
    references** to them. ONLY CONCERNED WITH LIFETIMES! Does not check
    (for example) consistent authorship.
    '''
    _librarian = weak_property('__librarian')
        
    def assemble(self, librarian):
        # Call before using.
        self._librarian = librarian
        
    async def validate_gidc(self, obj):
        ''' GIDC need no state verification.
        '''
        return True
        
    async def validate_geoc(self, obj):
        ''' GEOC must verify that they are bound.
        '''
        if not (await self._librarian.is_bound(obj)):
            raise UnboundContainer(str(obj))
        
        return True
        
    async def validate_gobs(self, obj):
        if (await self._librarian.is_debound(obj)):
            raise AlreadyDebound(str(obj), ghid=obj.ghid)
            
        return True
        
    async def validate_gobd(self, obj):
        # A deliberate binding can override a debinding for GOBD.
        if (await self._librarian.is_debound(obj)):
            if not (await self._librarian.is_bound(obj)):
                raise AlreadyDebound(str(obj), ghid=obj.ghid)
                
        return True
        
    async def validate_gdxx(self, obj):
        if (await self._librarian.is_debound(obj)):
            raise AlreadyDebound(str(obj), ghid=obj.ghid)
            
        return True
        
    async def validate_garq(self, obj):
        if (await self._librarian.is_debound(obj)):
            raise AlreadyDebound(str(obj), ghid=obj.ghid)
            
        return True class LawyerCore(metaclass=API):
    ''' Enforces authorship requirements, including both having a known
    entity as author/recipient and consistency for eg. bindings and
    debindings.
    
    Threadsafe.
    '''
    _librarian = weak_property('__librarian')
    
    @fixture_api
    def __init__(self, librarian, *args, **kwargs):
        super(LawyerCore.__fixture__, self).__init__(*args, **kwargs)
        self._librarian = librarian
        
    def assemble(self, librarian):
        # Call before using.
        self._librarian = librarian
    
    @fixture_return(True)
    @public_api
    async def _validate_author(self, obj):
        try:
            author = await self._librarian.summarize(obj.author)
        
        except KeyError as exc:
            raise InvalidIdentity('Unknown author: ' +
                                  str(obj.author)) from exc
        
        else:
            if not isinstance(author, _GidcLite):
                raise InvalidIdentity('Invalid author: ' + str(author))
                
        return True
        
    async def validate_gidc(self, obj):
        ''' GIDC need no validation.
        '''
        return True
        
    async def validate_geoc(self, obj):
        ''' Ensure author is known and valid.
        '''
        return (await self._validate_author(obj))
        
    async def validate_gobs(self, obj):
        ''' Ensure author is known and valid.
        '''
        return (await self._validate_author(obj))
        
    async def validate_gobd(self, obj):
        ''' Ensure author is known and valid, and consistent with the
        previous author for the binding (if it already exists).
        '''
        await self._validate_author(obj)
        
        if obj.counter > 0:
            try:
                existing = await self._librarian.summarize(obj.ghid)
            
            except KeyError:
                logger.warning(str(obj) + ' previous binding missing; ' +
                               'cannot validate author self-consistency.')
            
            else:
                if existing.author != obj.author:
                    raise InconsistentAuthor('Existing author for ' +
                                             str(obj) + ': ' +
                                             str(existing.author) +
                                             ' (existing) vs ' +
                                             str(obj.author) + ' (attempted)')
        
        return True
        
    async def validate_gdxx(self, obj, target_obj=None):
        ''' Ensure author is known and valid, and consistent with the
        previous author for the binding.
        
        If other is not None, specifically checks it against that object
        instead of obtaining it from librarian.
        '''
        await self._validate_author(obj)
        
        try:
            if target_obj is None:
                existing = await self._librarian.summarize(obj.target)
            else:
                existing = target_obj
                
        except KeyError:
            logger.debug(str(obj) + ' target missing from librarian; ' +
                         'cannot validate author self-consistency: ' +
                         str(obj.target))
            
        else:
            if isinstance(existing, _GarqLite):
                if existing.recipient != obj.author:
                    raise InconsistentAuthor('Existing author for ' +
                                             str(obj) + ': ' +
                                             str(existing.recipient) +
                                             ' (existing) vs ' +
                                             str(obj.author) + ' (attempted)')
                
            else:
                if existing.author != obj.author:
                    raise InconsistentAuthor('Existing author for ' +
                                             str(obj) + ': ' +
                                             str(existing.author) +
                                             ' (existing) vs ' +
                                             str(obj.author) + ' (attempted)')
        return True
        
    async def validate_garq(self, obj):
        ''' Validate recipient.
        '''
        try:
            recipient = await self._librarian.summarize(obj.recipient)
        
        except KeyError as exc:
            raise InvalidIdentity('Unknown recipient: ' +
                                  str(obj.recipient)) from exc
        
        else:
            if not isinstance(recipient, _GidcLite):
                raise InvalidIdentity('Invalid recipient: ' + str(recipient))
                
        return True class DiskLibrarian(LibrarianCore):
    ''' Librarian that caches data to disk, but keeps all status state
    in memory.
    '''
    
    def __init__(self, cache_dir, executor, loop, *args, **kwargs):
        ''' cache_dir should be relative to current.
        '''
        super().__init__(*args, **kwargs)
        
        cache_dir = pathlib.Path(cache_dir)
        if not cache_dir.exists():
            raise ValueError('Path does not exist: ' + cache_dir.as_posix())
        elif not cache_dir.is_dir():
            raise ValueError('Path is not an available directory: ' +
                             cache_dir.as_posix())
        
        self._loop = loop
        self._executor = executor
        self._cachedir = cache_dir
        
        # This allows us to be lazy when restoring things, without rewriting
        # disk data
        self._restoration_flag = False
        
        # Lookup for dynamic ghid -> frame ghid
        self._dyn_resolver = {}
        
        # Lookup <bound ghid>: set(<binding obj>)
        self._bound_by_ghid = SetMap()
        
        # Lookup <debound ghid>: set(<debinding ghid>)
        self._debound_by_ghid = SetMap()
        
        # Lookup <recipient>: set(<request ghid>)
        self._requests_for_recipient = SetMap()
        
        # Make sure we don't do concurrent write/reads into the cache, which
        # could potentially create contention issues for stuff.
        # TODO: make this keyed, so that we can do simultaneous read/write of
        # different things at the same time (just not the same thing at the
        # same time)
        self._cache_lock = KeyedAsyncioLock(loop=self._loop)
        
    def __read_from_disk(self, ghid):
        ''' Gets a file path from the disk cache, wrapping misses in
        DoesNotExist.
        '''
        fpath = self._make_path(ghid)
        
        try:
            return fpath.read_bytes()
        
        except FileNotFoundError as exc:
            # Remove the filename from the exception context so as not to
            # disclose the full GHID
            raise DoesNotExist(str(ghid)) from None
            
    def __remove_from_disk(self, ghid):
        ''' Removes a ghid from the disk cache, wrapping misses in
        DoesNotExist.
        '''
        try:
            fpath = self._make_path(ghid)
            fpath.unlink()
            
        except FileNotFoundError:
            # Suppress the full name of the file to prevent knowing its whole
            # GHID
            raise DoesNotExist(str(ghid)) from None
        
    async def get_from_cache(self, ghid):
        ''' Returns the raw data associated with the ghid.
        '''
        try:
            ghid = await self.resolve_frame(ghid)
        except KeyError:
            pass
        
        async with self._cache_lock(ghid):
            return (await self._loop.run_in_executor(self._executor,
                                                     self.__read_from_disk,
                                                     ghid))
            
    async def add_to_cache(self, obj, data):
        ''' Adds the passed raw data to the cache.
        '''
        if isinstance(obj, _GobsLite):
            reference_ghid = obj.ghid
            self._bound_by_ghid.add(obj.target, obj.ghid)
            
        elif isinstance(obj, _GobdLite):
            reference_ghid = obj.frame_ghid
            try:
                existing = await self.summarize(obj.ghid)
            except KeyError:
                pass
            else:
                self._bound_by_ghid.remove(existing.target, obj.ghid)
                
            # Now we have a clean slate and need to update things accordingly.
            self._bound_by_ghid.add(obj.target, obj.ghid)
            self._dyn_resolver[obj.ghid] = obj.frame_ghid
                
        elif isinstance(obj, _GdxxLite):
            reference_ghid = obj.ghid
            self._debound_by_ghid.add(obj.target, obj.ghid)
            
        elif isinstance(obj, _GarqLite):
            reference_ghid = obj.ghid
            # NOTE: this is only necessary for a persistence server
            self._requests_for_recipient.add(obj.recipient, obj.ghid)
            
        else:
            reference_ghid = obj.ghid
            
        if not self._restoration_flag:
            fpath = self._make_path(reference_ghid)
            
            async with self._cache_lock(reference_ghid):
                await self._loop.run_in_executor(self._executor,
                                                 fpath.write_bytes, data)
    
    async def remove_from_cache(self, ghid):
        ''' Removes the data associated with the passed ghid from the
        cache.
        '''
        obj = await self.summarize(ghid)
        
        if isinstance(obj, _GobsLite):
            reference_ghid = obj.ghid
            self._bound_by_ghid.discard(obj.target, obj.ghid)
            
        elif isinstance(obj, _GobdLite):
            reference_ghid = obj.frame_ghid
            self._bound_by_ghid.discard(obj.target, obj.ghid)
            
            if self._dyn_resolver.get(obj.ghid, None) == obj.frame_ghid:
                self._dyn_resolver.pop(obj.ghid, None)
            
        elif isinstance(obj, _GdxxLite):
            reference_ghid = obj.ghid
            self._debound_by_ghid.discard(obj.target, obj.ghid)
            
        elif isinstance(obj, _GarqLite):
            reference_ghid = obj.ghid
            self._requests_for_recipient.discard(obj.recipient, obj.ghid)
            
        else:
            reference_ghid = obj.ghid
        
        async with self._cache_lock(reference_ghid):
            await self._loop.run_in_executor(self._executor,
                                             self.__remove_from_disk,
                                             reference_ghid)
        
    async def contains(self, ghid):
        ''' Checks the ghidcache for the ghid.
        '''
        try:
            ghid = await self.resolve_frame(ghid)
        except KeyError:
            pass
        
        fpath = self._make_path(ghid)
        return (await self._loop.run_in_executor(self._executor, fpath.exists))
    
    async def resolve_frame(self, ghid):
        ''' Get the current frame ghid from the dynamic ghid.
        '''
        if not isinstance(ghid, Ghid):
            raise TypeError('Ghid must be a Ghid.')
            
        if ghid in self._dyn_resolver:
            return self._dyn_resolver[ghid]
        else:
            raise KeyError(str(ghid) + ' not known as dynamic ghid.')
    
    async def recipient_status(self, ghid):
        ''' Return a frozenset of ghids assigned to the passed ghid as
        a recipient.
        '''
        return self._requests_for_recipient.get_any(ghid)
    
    async def bind_status(self, ghid):
        ''' Return a frozenset of ghids binding the passed ghid.
        '''
        return self._bound_by_ghid.get_any(ghid)
    
    async def debind_status(self, ghid):
        ''' Return either a ghid, or None.
        '''
        # Note that any particular object can have exactly zero or one VALID
        # debinds, but that a malicious actor could find a race condition and
        # debind something FOR SOMEONE ELSE before the bookie knows about the
        # original object authorship.
        return self._debound_by_ghid.get_any(ghid)
    
    # If subclasses want/need to do anything to restore themselves, they should
    # override this.
    async def restore(self):
        ''' Loads any existing files from the cache.  All existing
        files there will be attempted to be loaded, so it's best not to
        have extraneous stuff in the directory. Will be passed through
        to the core for processing.
        '''
        self._restoration_flag = True
        try:
            # Get all available files (this is a massive contention problem
            # and race condition waiting to happen. DON'T use concurrent copies
            # of the librarian.)
            # Iterate over each file within the cache. We're doing this one
            # time only, so don't bother doing the iterdir in an executor
            for child in self._cachedir.iterdir():
                if child.is_file():
                    data = await self._loop.run_in_executor(self._executor,
                                                            child.read_bytes)
                    obj = await self._percore.attempt_load(data)
                    # Lazily just use store to re-load our previous bookkeeping
                    # state
                    await self.store(obj, data)
                
        # Reset the restoration flag
        finally:
            self._restoration_flag = False
        
    def _make_path(self, ghid):
        ''' Converts the ghid to a file path.
        '''
        fname = ghid.as_str() + '.ghid'
        fpath = self._cachedir / fname
        return fpath class UndertakerCore(loopa.TaskLooper, metaclass=API):
    ''' Note: what about post-facto removal of bindings that have
    illegal targets? For example, if someone uploads a binding for a
    target that isn't currently known, and then it turns out that the
    target, once uploaded, actually doesn't support that binding, what
    should we do?
    
    In theory it shouldn't affect other operations. Should we just bill
    for it and call it a day? We'd need to make some kind of call to the
    bookie to handle that.
    '''
    _librarian = weak_property('__librarian')
    _postman = weak_property('__postman')
    
    def __init__(self, *args, maxlen=25, **kwargs):
        super().__init__(*args, **kwargs)
        self._maxlen = maxlen
        self._triage = None
        
        self._check_lookup = {
            _GidcLite: self._check_gidc,
            _GeocLite: self._check_geoc,
            _GobsLite: self._check_gobs,
            _GobdLite: self._check_gobd,
            _GdxxLite: self._check_gdxx,
            _GarqLite: self._check_garq
        }
        
    async def await_idle(self):
        ''' Wait until the undertaker has no more GC to perform.
        '''
        while self._triage is None:
            await asyncio.sleep(.01)
        
        await self._triage.join()
        
    async def loop_init(self):
        ''' Set up the staging queue.
        '''
        self._triage = asyncio.Queue(maxsize=self._maxlen)
        
    async def loop_stop(self):
        ''' Clear the staging queue.
        '''
        self._triage = None
        
    async def loop_run(self):
        ''' Wait for stuff to be added to the queue, then execute GC on
        it.
        '''
        ghid_to_collect, skip_conn = await self._triage.get()
        logger.debug(str(ghid_to_collect) + ' GC check starting.')
        try:
            obj = await self._librarian.summarize(ghid_to_collect)
            
            try:
                collection_check = self._check_lookup[type(obj)]
            
            except KeyError as exc:
                raise TypeError('Invalid object type received from ' +
                                'librarian.') from exc
                
            else:
                removable = await collection_check(obj, skip_conn)
                if removable:
                    logger.info(str(obj) + ' garbage collection starting.')
                    await self._gc_execute(obj, skip_conn)
                else:
                    logger.debug(str(obj) + ' garbage collection unneeded.')
            
        # except DoesNotExist:
        except KeyError:
            logger.warning(str(ghid_to_collect) + ' missing; could not ' +
                           'garbage collect it.')
            
        finally:
            self._triage.task_done()
        
    def assemble(self, librarian, postman):
        # Call before using.
        self._librarian = librarian
        self._postman = postman
            
    async def _check_gidc(self, obj, skip_conn=None):
        ''' Check whether we should remove a GIDC, and then remove it
        if appropriate. Currently we don't do that, so just leave it
        alone.
        '''
        return False
            
    async def _check_geoc(self, obj, skip_conn=None):
        ''' Check whether we should remove a GEOC, and then remove it if
        appropriate. Pretty simple: is it bound?
        '''
        # Keep if bound
        if (await self._librarian.is_bound(obj)):
            return False
        # Remove if unbound
        else:
            return True
            
    async def _check_gobs(self, obj, skip_conn=None):
        if (await self._librarian.is_debound(obj)):
            # Add our target to the list of GC checks
            await self._triage.put((obj.target, skip_conn))
            return True
        else:
            return False
            
    async def _check_gobd(self, obj, skip_conn=None):
        # If we've been debound, it might be time to die
        if (await self._librarian.is_debound(obj)):
            # Except child bindings can prevent GCing GOBDs
            if (await self._librarian.is_bound(obj)):
                return False
            
            # Dead binding.
            else:
                # Still need to add target
                await self._triage.put((obj.target, skip_conn))
                return True
        
        # Nope, still alive.
        else:
            return False
            
    async def _check_gdxx(self, obj, skip_conn=None):
        # Note that removing a debinding cannot result in a downstream target
        # being GCd, because it wouldn't exist.
        if (await self._librarian.is_debound(obj)):
            return True
        else:
            return False
            
    async def _check_garq(self, obj, skip_conn=None):
        if (await self._librarian.is_debound(obj)):
            return True
        else:
            return False
        
    async def _gc_execute(self, obj, skip_conn):
        # Next, goodbye object.
        await self._librarian.abandon(obj)
        # Now notify the postman, and tell her it's a removal.
        await self._postman.schedule(obj, removed=True, skip_conn=skip_conn)
    
    @fixture_return(None)
    @public_api
    async def alert_gidc(self, obj, skip_conn=None):
        ''' GIDC do not affect GC.
        '''
        # GIDC creates zero triage calls.
        return None
        
    @fixture_return(None)
    @public_api
    async def alert_geoc(self, obj, skip_conn=None):
        ''' GEOC do not affect GC.
        '''
        # GEOC creates zero triage calls.
        return None
        
    @fixture_return(None)
    @public_api
    async def alert_gobs(self, obj, skip_conn=None):
        ''' GOBS do not affect GC.
        '''
        return None
        
    @fixture_return(None)
    @public_api
    async def alert_gobd(self, obj, skip_conn=None):
        ''' GOBD require triage for previous targets.
        '''
        # This will always happen if it's the first frame, so let's be sure
        # to ignore that for logging (also, performance).
        if len(obj.target_vector) > 1:
            try:
                existing = await self._librarian.summarize(obj.ghid)
            
            except KeyError:
                logger.warning(str(obj) + ' existing binding missing; could ' +
                               'not triage old target.')
                triaged = None
            
            else:
                triaged = existing.target
                await self._triage.put((triaged, skip_conn))
        
        else:
            triaged = None
            
        return triaged
        
    @public_api
    async def alert_gdxx(self, obj, skip_conn=None):
        ''' GDXX require triage for new targets.
        '''
        triaged = obj.target
        await self._triage.put((triaged, skip_conn))
        return triaged
        
    @alert_gdxx.fixture
    async def alert_gdxx(self, obj, skip_conn=None):
        ''' Return the obj.target without triaging when fixtured.
        '''
        return obj.target
        
    @fixture_return(None)
    @public_api
    async def alert_garq(self, obj, skip_conn=None):
        ''' GARQ do not affect GC.
        '''
        return None class MrPostman(PostalCore, metaclass=API):
    ''' Postman to use for LOCAL persistence systems -- ie ones that
    have logins and do not support subsription.
    
    Note that MrPostman doesn't need to worry about silencing updates,
    because the persistence ingestion tract will only result in a mail
    run if there's a new object there. So, by definition, any re-sent
    objects will be DOA.
    '''
    _rolodex = weak_property('__rolodex')
    _golcore = weak_property('__golcore')
    _oracle = weak_property('__oracle')
    _salmonator = weak_property('__salmonator')
        
    def assemble(self, golcore, oracle, librarian, rolodex, salmonator):
        super().assemble(librarian)
        self._golcore = golcore
        self._rolodex = rolodex
        self._oracle = oracle
        self._salmonator = salmonator
            
    async def _deliver(self, subscription, notification, skip_conn):
        ''' Do the actual subscription update.
        
        NOTE THAT SKIP_CONN is a weakref.ref (or None).
        '''
        # We just got a garq for our identity. Rolodex handles these.
        if subscription == self._golcore.whoami:
            logger.debug(str(subscription) +
                         'subscription notification handed off to rolodex...')
            await self._rolodex.notification_handler(
                subscription,
                notification
            )
        
        # Anything else is an object subscription. Handle those by directly,
        # but only if we have them in memory.
        # elif subscription in self._oracle:
        # Note: that was causing issues so I'm holding off on it as premature
        # optimization / feature creep
        else:
            # The ingestion pipeline will already have applied any new updates
            # to the ghidproxy.
            try:
                obj = await self._oracle.get_object(GAOCore, subscription)
            
            except KeyError:
                logger.debug(str(subscription) + ' subscription delivery ' +
                             'ignored: unavailable at oracle: ' +
                             str(notification))
                await self._salmonator.deregister(subscription)
                
            else:
                logger.debug(str(subscription) + ' subscription delivery ' +
                             'starting for ' + str(notification))
                await obj.pull(notification) class Dispatcher(metaclass=API):
    ''' The Dispatcher decides which objects should be delivered where.
    This is decided through either:
    
    1. An API identifier (schema) dictating general compatibility as a
        dispatchable
    2. Declaring an object to be private, at which point the Dispatcher
        will internally (though distributedly) maintain that object to
        be exclusively available to an app token.
        
    Objects, once declared as non-private, cannot be retroactively
    privatized. That cat has officially left the bag/building/rodeo.
    However, a private object can be made non-private at a later time,
    provided it has defined an API ID.
    
    Private objects do not track their owning app tokens internally.
    Instead, this is managed through a Dispatcher-internal GAO. As such,
    even a bug resulting in a leaked private object will not result in a
    leaked app token.
    
    Ideally, the dispatcher will eventually also enforce whatever user
    restrictions on sharing are desired BETWEEN INSTALLED APPLICATIONS.
    Sharing restrictions between external parties are within the purview
    of the Rolodex.
    
    TODO: support notification mechanism to push new objects to other
    concurrent hypergolix instances. See note in ipc.ipccore.send_object
    
    TODO: move most of the misc stuff into dedicated new_object and
    get_object methods within dispatch, which (for the most part) just
    pass on straight to the oracle, but add all of the extra tracking
    stuff (and so on) internally.
    '''
    _account = weak_property('__account')
    _ipc_protocol = weak_property('__ipc_protocol')
    _oracle = weak_property('__oracle')
    
    @public_api
    def __init__(self, *args, **kwargs):
        ''' Yup yup yup yup yup yup yup
        '''
        super().__init__(*args, **kwargs)
        
        # Temporarily set distributed state to None.
        # Lookup for all known tokens: set(<tokens>)
        self._all_known_tokens = None
        # Lookup (dict-like) for <app token>: <startup ghid>
        self._startup_by_token = None
        # Lookup (dict-like) for <obj ghid>: <private owner>
        self._private_by_ghid = None
        # Distributed lock for adding app tokens
        self._token_lock = None
        
        # Set of incoming shared ghids that had no conn
        # set(<ghid, sender tuples>)
        self._orphan_incoming_shares = None
        # Setmap-like lookup for share acks that had no conn
        # <app token>: set(<ghids>)
        self._orphan_share_acks = None
        # Setmap-like lookup for share naks that had no conn
        # <app token>: set(<ghids>)
        self._orphan_share_naks = None
        
        # Lookup <app token>: <connection/session/conn>
        self._conn_from_token = weakref.WeakValueDictionary()
        # Reverse lookup <connection/session/conn>: <app token>
        self._token_from_conn = weakref.WeakKeyDictionary()
        
        # Lookup <api ID>: set(<connection/session/conn>)
        self._conns_from_api = WeakSetMap()
        
        # This lookup directly tracks who has a copy of the object
        # Lookup <object ghid>: set(<connection/session/conn>)
        self._update_listeners = WeakSetMap()
        # This lookup preserves a strong reference to the dispatchable **as the
        # key**, allowing the object to be GC'd from local memory only when no
        # connections remain that have it
        self._obj_binding = WeakSetMap()
        
    @__init__.fixture
    def __init__(self, *args, **kwargs):
        ''' Create a dispatch fixture.
        '''
        super(Dispatcher.__fixture__, self).__init__(*args, **kwargs)
        
        self._all_known_tokens = set()
        # Lookup (dict-like) for <app token>: <startup ghid>
        self._startup_by_token = {}
        # Lookup (dict-like) for <obj ghid>: <private owner>
        self._private_by_ghid = {}
        # Distributed lock for adding app tokens
        self._token_lock = NoContext()
        
        # Lookup <api ID>: set(<connection/session/conn>)
        self._conns_from_api = WeakSetMap()
        
        # Lookup <app token>: <connection/session/conn>
        self._conn_from_token = weakref.WeakValueDictionary()
        # Reverse lookup <connection/session/conn>: <app token>
        self._token_from_conn = weakref.WeakKeyDictionary()
        
    @fixture_api
    def RESET(self):
        ''' Reset the fixture to a pristine state.
        '''
        self._all_known_tokens.clear()
        # Lookup (dict-like) for <app token>: <startup ghid>
        self._startup_by_token.clear()
        # Lookup (dict-like) for <obj ghid>: <private owner>
        self._private_by_ghid.clear()
        # Lookup <api ID>: set(<connection/session/conn>)
        self._conns_from_api.clear_all()
        
        # Lookup <app token>: <connection/session/conn>
        self._conn_from_token.clear()
        # Reverse lookup <connection/session/conn>: <app token>
        self._token_from_conn.clear()
        
    def assemble(self, oracle, ipc_protocol):
        self._oracle = oracle
        self._ipc_protocol = ipc_protocol
        
    @public_api
    def bootstrap(self, account):
        ''' Initialize distributed state.
        '''
        self._account = account
        
        # Now init distributed state.
            
        # Lookup for all known tokens: set(<tokens>)
        self._all_known_tokens = account.dispatch_tokens
        # Lookup (set-map-like) for <app token>: set(<startup ghids>)
        self._startup_by_token = account.dispatch_startup
        # Lookup (dict-like) for <obj ghid>: <private owner>
        self._private_by_ghid = account.dispatch_private
        
        # These need to be distributed but aren't yet. TODO!
        # Distributed lock for adding app tokens
        self._token_lock = NoContext()
        
        # Set of incoming shared ghids that had no conn
        # set(<ghid, sender tuples>)
        self._orphan_incoming_shares = account.dispatch_incoming
        # Setmap-like lookup for share acks that had no conn
        # <app token>: set(<ghid, recipient tuples>)
        self._orphan_share_acks = account.dispatch_orphan_acks
        # Setmap-like lookup for share naks that had no conn
        # <app token>: set(<ghid, recipient tuples>)
        self._orphan_share_naks = account.dispatch_orphan_naks
        
        # All known tokens must already contain a key for the null token, which
        # is used by dispatch to indicate a lack of a token in an exchange. Do
        # this as LBYL to avoid false positive _mutated flags in the GAO.
        if AppToken.null() not in self._all_known_tokens:
            self._all_known_tokens.add(AppToken.null())
            
    @public_api
    def token_lookup(self, connection):
        ''' Return the token associated with the connection, or None if
        there is no currently defined token.
        '''
        try:
            return self._token_from_conn[connection]
            
        except KeyError as exc:
            return None
            
    @public_api
    def connection_lookup(self, token):
        ''' Returns the current connection associated with the token, or
        None if there is no currently available connection.
        '''
        try:
            return self._conn_from_token[token]
            
        except KeyError as exc:
            return None
    
    @public_api
    def private_parent_lookup(self, ghid):
        ''' Returns the app_token parent for the passed ghid, if (and
        only if) it's private. Otherwise, returns None.
        '''
        try:
            return self._private_by_ghid[ghid]
        except KeyError:
            return None
    
    @public_api
    async def make_public(self, ghid):
        ''' Makes a private object public.
        '''
        try:
            del self._private_by_ghid[ghid]
            
        except KeyError as exc:
            raise ValueError(
                'Obj w/ passed ghid is unknown or already public: ' + str(ghid)
            ) from exc
            
        else:
            await self._account.flush()
            
    @make_public.fixture
    async def make_public(self, ghid):
        # Bypass all validation
        del self._private_by_ghid[ghid]
        
    @public_api
    async def add_api(self, connection, api_id):
        ''' Register the connection as currently tracking the api_id.
        '''
        self._conns_from_api.add(api_id, connection)
        await self._account.flush()
        
    @add_api.fixture
    async def add_api(self, connection, api_id):
        # Bypass account flushing
        self._conns_from_api.add(api_id, connection)
        
    @public_api
    async def remove_api(self, connection, api_id):
        ''' Remove a connection's registration for the api_id. Happens
        automatically when connections are GC'd.
        '''
        self._conns_from_api.discard(api_id, connection)
        await self._account.flush()
        
    @remove_api.fixture
    async def remove_api(self, connection, api_id):
        # Bypass account flushing
        self._conns_from_api.discard(api_id, connection)
        
    def _make_new_token(self):
        token = None
        # Birthday paradox be damned; we can actually *enforce* uniqueness
        while token is None or token in self._all_known_tokens:
            token = AppToken(os.urandom(AppToken.TOKEN_LEN))
        return token
        
    @public_api
    async def start_application(self, connection, token=None):
        ''' Ensures that an application is known to the dispatcher.
        
        Currently just checks within all known tokens. In the future,
        this be responsible for starting the application (unless we move
        that responsibility elsewhere), and then sending all of the
        startup objects to the application.
        '''
        if connection in self._token_from_conn:
            raise ExistantAppError('Token already registered for connection.')
            
        elif token in self._conn_from_token:
            raise ExistantAppError(
                'Token already started at different connection.'
            )
            
        elif token is None:
            # Token lock is currently a NoContext, but eventually will need to
            # be a distributed lock -- or, the token lookup needs to be made
            # into a no-contention system. Which I'm not sure it possibly can
            # be, since we're best off actually ENFORCING the "no birthday
            # collisions" for tokens.
            with self._token_lock:
                token = self._make_new_token()
                # Do this right away to prevent race condition
                self._all_known_tokens.add(token)
                await self._account.flush()
            
        elif token not in self._all_known_tokens:
            raise UnknownToken('App token unknown to dispatcher.')
            
        # TODO: should these be enclosed within an operations lock?
        self._conn_from_token[token] = connection
        self._token_from_conn[connection] = token
        
        return token
        
    @start_application.fixture
    async def start_application(self, connection, token=None):
        if token is None:
            # ONLY FOR THE FIXTURE! Do we use a pseudorandom token.
            token = AppToken.pseudorandom()
        
        # Don't emulate normal dispatcher behavior here; let everything start,
        # regardless of status re: "exists in tokens"
        self._all_known_tokens.add(token)
        self._conn_from_token[token] = connection
        self._token_from_conn[connection] = token
        
        return token
    
    @fixture_noop
    @public_api
    async def track_object(self, connection, ghid):
        ''' Registers a connection as tracking a ghid.
        
        This is necessary so that upstream updates can be properly
        dispatched to any applications with copies of the object.
        '''
        logger.debug(
            'CONN ' + str(connection) + ' tracking ' + str(ghid) + '...'
        )
        self._update_listeners.add(ghid, connection)
        obj = await self._oracle.get_object(
            gaoclass = _Dispatchable,
            ghid = ghid,
            api_id = None,  # Let _pull() apply this.
            state = None,   # Let _pull() apply this.
            dispatch = self,
            ipc_protocol = self._ipc_protocol,
            account = self._account
        )
        # This keeps the object in memory, allowing it to receive subscriptions
        # When all apps untrack the object, it gets unsubbed.
        self._obj_binding.add(obj, connection)
    
    @fixture_noop
    @public_api
    async def untrack_object(self, connection, ghid):
        ''' Remove a connection as tracking a ghid.
        
        This indicates an application no longer has a copy of the
        object, therefore silencing any updates it would otherwise have
        received.
        '''
        logger.debug(
            'CONN ' + str(connection) + ' untracking ' + str(ghid) + '...'
        )
        self._update_listeners.discard(ghid, connection)
        obj = await self._oracle.get_object(
            gaoclass = _Dispatchable,
            ghid = ghid,
            api_id = None,  # Let _pull() apply this.
            state = None,   # Let _pull() apply this.
            dispatch = self,
            ipc_protocol = self._ipc_protocol,
            account = self._account
        )
        # This keeps the object in memory, allowing it to receive subscriptions
        # When all apps untrack the object, it gets unsubbed.
        self._obj_binding.discard(obj, connection)
        
    @public_api
    async def register_object(self, connection, ghid, private):
        ''' Call this every time a new object is created to register it
        with the dispatcher, recording it as private or distributing it
        to other applications as needed.
        '''
        # If the object is private, register it as such.
        if private:
            try:
                token = self._token_from_conn[connection]
            
            except KeyError as exc:
                raise UnknownToken(
                    'Must register app token before creating private objects.'
                ) from exc
                
            else:
                logger.debug(
                    'Creating private object for ' + str(connection) +
                    '; bypassing distribution.'
                )
                self._private_by_ghid[ghid] = token
            
        # Otherwise, make sure to notify any other interested parties.
        else:
            make_background_future(
                self.distribute_share(
                    ghid,
                    origin = None,
                    skip_conn = connection
                )
            )
            
    @register_object.fixture
    async def register_object(self, connection, ghid, private):
        ''' Just don't distribute the object.
        '''
        if private:
            try:
                token = self._token_from_conn[connection]
            
            except KeyError as exc:
                raise UnknownToken(
                    'Must register app token before creating private objects.'
                ) from exc
                
            else:
                logger.debug(
                    'Creating private object for ' + str(connection) +
                    '; bypassing distribution.'
                )
                self._private_by_ghid[ghid] = token
            
    @public_api
    async def register_startup(self, connection, ghid):
        ''' Registers a ghid to be used as a startup object for token.
        '''
        try:
            token = self._token_from_conn[connection]
            
        except KeyError as exc:
            raise UnknownToken(
                'Must register app token before registering startup objects.'
            ) from exc
            
        else:
            if token in self._startup_by_token:
                raise ValueError(
                    'Startup object already defined for that application. '
                    'Deregister it before registering a new startup object.'
                )
            else:
                self._startup_by_token[token] = ghid
                await self._account.flush()
                
    @public_api
    async def deregister_startup(self, connection):
        ''' Deregisters a ghid to be used as a startup object for token.
        '''
        try:
            token = self._token_from_conn[connection]
            
        except KeyError as exc:
            raise UnknownToken(
                'Must register app token before deregistering startup objects.'
            ) from exc
            
        else:
            if token not in self._startup_by_token:
                raise ValueError(
                    'Startup object has not been defined for that application.'
                )
            else:
                del self._startup_by_token[token]
                await self._account.flush()
    
    @public_api
    async def get_startup_obj(self, token):
        ''' Returns the ghid of the declared startup object for that
        token, or None if none has been declared.
        '''
        if token not in self._all_known_tokens:
            raise UnknownToken()
        
        elif token not in self._startup_by_token:
            return None
        
        else:
            return self._startup_by_token[token]
    
    @fixture_noop
    @public_api
    async def distribute_share(self, ghid, origin, skip_conn=None):
        ''' Perform an actual share distribution.
        
        Pass an explicit None as the origin to indicate self as origin.
        '''
        callsheet = set()
        
        # Note that this is called in exactly two situations: when creating a
        # new object locally, through register_object, or when receiving a new
        # share, through rolodex. As such, this will never be called when the
        # object is private. So, don't check for a private owner.
        
        obj = await self._oracle.get_object(
            gaoclass = _Dispatchable,
            ghid = ghid,
            api_id = None,  # Let _pull() apply this.
            state = None,   # Let _pull() apply this.
            dispatch = self,
            ipc_protocol = self._ipc_protocol,
            account = self._account
        )
        
        callsheet.update(
            # Get any connections that have registered the api_id
            self._conns_from_api.get_any(obj.api_id)
        )
        callsheet.update(
            # Get any connections that have an instance of the object
            self._update_listeners.get_any(obj.ghid)
        )
            
        # Now, check to see if we have anything in the callsheet. Do it before
        # discarding, so that we know if it's actually an orphan share.
        if callsheet:
            # Discard the skipped connection, if one is defined
            callsheet.discard(skip_conn)
            str_origin = origin or 'self'
            logger.debug(
                'Distributing ' + str(ghid) + ' shared object from ' +
                str(str_origin) + ' to ' + repr(callsheet)
            )
            
            # If we still have a callsheet, distribute it.
            if callsheet:
                await self._distribute(
                    self._ipc_protocol.share_obj,    # distr_coro
                    callsheet,
                    ghid,
                    origin,
                    obj.api_id
                )
        
        # Only make an orphan_shares entry if we didn't create the object.
        elif origin is not None:
            sharelog = _ShareLog(ghid, origin)
            self._orphan_incoming_shares.add(sharelog)
            await self._account.flush()
    
    @fixture_noop
    @public_api
    async def distribute_update(self, ghid, deleted=False, skip_conn=None):
        ''' Perform an actual update distribution.
        '''
        # Get any connections that have an instance of the object
        callsheet = set()
        # Note that this call returns a frozenset, hence the copy.
        callsheet.update(self._update_listeners.get_any(ghid))
        # Skip the connection if one is passed.
        callsheet.discard(skip_conn)
        
        logger.debug(
            'Distributing {!s} update to {!r}.'.format(ghid, callsheet)
        )
        
        if deleted:
            await self._distribute(
                self._ipc_protocol.delete_obj,   # distr_coro
                callsheet,
                ghid
            )
            
        else:
            await self._distribute(
                self._ipc_protocol.update_obj,   # distr_coro
                callsheet,
                ghid
            )
            
    @fixture_noop
    @public_api
    async def distribute_share_success(self, ghid, recipient, tokens):
        ''' Notify any available connections for all passed <tokens> of
        a share failure.
        '''
        callsheet = set()
        for token in tokens:
            # Escape any keys that have gone missing during the rat race
            try:
                callsheet.add(self._conn_from_token[token])
            except KeyError:
                logger.info('No connection for token ' + str(token))
        
        # Distribute the share failure to all apps that requested its delivery
        await self._distribute(
            self._ipc_protocol.notify_share_success,   # distr_coro
            callsheet,
            ghid,
            recipient
        )
    
    @fixture_noop
    @public_api
    async def distribute_share_failure(self, ghid, recipient, tokens):
        ''' Notify any available connections for all passed <tokens> of
        a share failure.
        '''
        callsheet = set()
        for token in tokens:
            # Escape any keys that have gone missing during the rat race
            try:
                callsheet.add(self._conn_from_token[token])
            except KeyError:
                logger.info('No connection for token ' + str(token))
        
        # Distribute the share failure to all apps that requested its delivery
        await self._distribute(
            self._ipc_protocol.notify_share_failure,   # distr_coro
            callsheet,
            ghid,
            recipient
        )
                
    async def _distribute(self, distr_coro, callsheet, *args, **kwargs):
        ''' Call distr_coro with *args and **kwargs at each of the
        connections in the callsheet. Log (but don't raise) any errors.
        '''
        logger.info(
            'Dispatching ' + str(len(callsheet)) + ' calls to ' +
            str(distr_coro)
        )
        distributions = set()
        for connection in callsheet:
            # For each connection...
            distributions.add(
                # ...in parallel, schedule a single execution of the
                # distribution coroutine.
                make_background_future(distr_coro(connection, *args, **kwargs))
            )
            
        # And gather the results, logging (but not raising) any exceptions.
        # BUT, only do it if we have some to do.
        if distributions:
            try:
                await asyncio.wait(
                    fs = distributions,
                    return_when = asyncio.ALL_COMPLETED
                )
            
            except asyncio.CancelledError:
                logger.debug('Cancelling pending distributions.')
                for distr in distributions:
                    distr.cancel()
                raise class _Dispatchable(GAOCore, metaclass=API):
    ''' A dispatchable object. Note that privacy is store within the
    dispatcher itself, and not within the individual dispatchables.
    '''
    _dispatch = weak_property('__dispatch')
    _account = weak_property('__account')
    _ipc_protocol = weak_property('__ipc_protocol')
    api_id = immutable_property('_api_id')
    
    def __init__(self, *args, api_id, state, dispatch, ipc_protocol, account,
                 **kwargs):
        super().__init__(*args, **kwargs)
        
        self._dispatch = dispatch
        self._ipc_protocol = ipc_protocol
        self._account = account
        
        # These may be explicitly set to None to allow pull() to apply them.
        if state is not None:
            self.state = state
            
        if api_id is not None:
            self.api_id = api_id
            
    @public_api
    async def _push(self):
        ''' Extend GAOCore._push() to flush the account upon completion.
        '''
        await super()._push()
        await self._account.flush()
        
    @_push.fixture
    async def _push(self):
        ''' Call super() to FIXTURE super instead of dispatchable.
        '''
        await super(_Dispatchable, self)._push()
    
    @public_api
    async def _pull(self):
        ''' Extend GAOCore._pull() to notify IPC about updates.
        '''
        await super()._pull()
        make_background_future(
            self._dispatch.distribute_update(self.ghid)
        )
        
    @_pull.fixture
    async def _pull(self):
        ''' Call super() to FIXTURE super instead of dispatchable.
        '''
        await super(_Dispatchable, self)._pull()
        
    async def pack_gao(self):
        ''' Packs state into a bytes object. May be overwritten in subs
        to pack more complex objects. Should always be a staticmethod or
        classmethod.
        '''
        version = b'\x00'
        return b'hgxd' + version + bytes(self.api_id) + self.state
        
    async def unpack_gao(self, packed):
        ''' Unpacks state from a bytes object. May be overwritten in
        subs to unpack more complex objects. Should always be a
        staticmethod or classmethod.
        '''
        magic = packed[0:4]
        version = packed[4:5]
        api_id = ApiID.from_bytes(packed[5:70])
        
        if magic != b'hgxd':
            raise DispatchError('Object does not appear to be dispatchable.')
        elif version != b'\x00':
            raise DispatchError('Incompatible dispatchable version number.')
        elif getattr(self, 'api_id', None) is None:
            self.api_id = api_id
        else:
            # YES, this does need to be _api_id (with underscore!) because
            # we may need to override something existing.
            self._api_id = api_id
        # Checking for this from other people isn't really that useful. We'll
        # already error out if you try to change it locally, and if someone
        # goes to the effort of changing it elsewhere, well, more power to them
        # I guess.
        # elif api_id != self.api_id:
        #     raise DispatchError('Cannot change API ID.')
            
        self.state = packed[70:]
    
    @fixture_noop
    @public_api
    async def apply_delete(self, debinding):
        ''' Extend GAOCore.apply_delete to notify consumers of deletion.
        '''
        await super().apply_delete(debinding)
        make_background_future(
            self._dispatch.distribute_update(self.ghid, deleted=True)
        )
        
    def __eq__(self, other):
        ''' Ensure the other object is the same gao with the same state.
        Intended pretty much exclusively for use in testing.
        '''
        equal = True
        
        try:
            equal &= (self.dynamic == other.dynamic)
            equal &= (self.ghid == other.ghid)
            equal &= (self.author == other.author)
            equal &= (self.state == other.state)
            equal &= (self.api_id == other.api_id)
            
        except AttributeError:
            equal = False
        
        return equal
        
    def __hash__(self):
        ''' Dispatchable hashes should mix the hashes of the api_id, the
        author, and the ghid together, but only if all of the above are
        defined.
        '''
        defined = self.ghid is not None
        defined &= self.author is not None
        defined &= self.api_id is not None
        
        if not defined:
            raise TypeError(
                'Dispatchable objects may only be hashed after they are ' +
                'fully defined (with a ghid, an author, and an api_id.'
            )
        
        return hash(self.ghid) ^ hash(self.author) ^ hash(self.api_id) class Salmonator(loopa.TaskLooper, metaclass=API):
    ''' Responsible for disseminating Golix objects upstream and
    downstream. Handles all comms with them as well.
    '''
    _golcore = weak_property('__golcore')
    _percore = weak_property('__percore')
    _librarian = weak_property('__librarian')
    _remote_protocol = weak_property('__remote_protocol')
    
    @public_api
    def __init__(self, *args, **kwargs):
        ''' Yarp.
        '''
        super().__init__(*args, **kwargs)
        
        self._deferred = ListMap()
        self._clear_q = None
        
        self._upstream_remotes = set()
        self._downstream_remotes = set()
        
        # Lookup for <registered ghid>
        self._registered = set()
        
        self._bootstrapped = False
        
    @fixture_api
    def RESET(self):
        ''' Clear stuff out, yo.
        '''
        self._upstream_remotes.clear()
        self._downstream_remotes.clear()
        self._registered.clear()
        
    def assemble(self, golcore, percore, librarian, remote_protocol):
        self._golcore = golcore
        self._percore = percore
        self._librarian = librarian
        self._remote_protocol = remote_protocol
    
    @fixture_noop
    @public_api
    def add_upstream_remote(self, task_commander, connection_cls, *args,
                            **kwargs):
        ''' Adds an upstream remote persister.
        *args and **kwargs will be passed to the task_commander task.
        '''
        remote = ConnectionManager(
            connection_cls = connection_cls,
            msg_handler = self._remote_protocol,
            conn_init = self.restore_connection,
        )
        # We have to insert the remote before us at the commander, or shutdown
        # will kill the connections before we can clean them up.
        task_commander.register_task(remote, *args, before_task=self, **kwargs)
        self._upstream_remotes.add(remote)
        
    def add_downstream_remote(self, persister):
        ''' Adds a downstream persister.
        
        PersistenceCore will not attempt to keep a consistent state with
        downstream persisters. Instead, it will simply push updates to
        local objects downstream. It will not, however, look to them for
        updates.
        
        Therefore, to create synchronization **between** multiple
        upstream remotes, also add them as downstream remotes.
        '''
        raise NotImplementedError()
        self._downstream_remotes.add(persister)
        
    async def loop_init(self, *args, **kwargs):
        ''' On top of the usual stuff, set up our queues.
        '''
        await super().loop_init(*args, **kwargs)
        # This "must" be unbounded, because we don't have any control over
        # finalization of objects.
        self._clear_q = asyncio.Queue()
        
    async def loop_stop(self, *args, **kwargs):
        ''' On top of the usual stuff, clear our queues.
        '''
        # Would be good to do, but not currently working
        # for remote in self._persisters:
        #     await self._stop_persister(remote)
        
        await super().loop_stop(*args, **kwargs)
        self._clear_q = None
        
        disconnections = set()
        for remote in self._upstream_remotes:
            if remote.has_connection:
                disconnections.add(
                    make_background_future(remote.disconnect())
                )
        
        # Need to make sure it's not empty
        if disconnections:
            await asyncio.wait(
                fs = disconnections,
                return_when = asyncio.ALL_COMPLETED
            )
        
    async def loop_run(self, *args, **kwargs):
        ''' Wait for object finalizers, and then immediately unsub the
        remotes when the objects are removed from memory.
        '''
        to_clear = await self._clear_q.get()
        await self.deregister(to_clear)
        
    @fixture_noop
    @public_api
    async def notify(self, ghid):
        ''' Notify the salmonator that an upstream subscription
        notification was just successfully completed.
        '''
        # If that was a dynamic object, we also need to pull its target.
        obj = await self._librarian.summarize(ghid)
        if isinstance(obj, _GobdLite):
            if not (await self._librarian.contains(obj.target)):
                await self.pull(obj.target)
    
    @fixture_noop
    @public_api
    async def register(self, ghid):
        ''' Tells the Salmonator to listen upstream for any updates.
        Consumers must independently register an object finalizer for
        salmonator._deregister (not to be called atexit) to remove the
        subscription when the object leaves local memory. TODO: fix that
        leaky abstraction.
        '''
        obj = await self._librarian.summarize(ghid)
        
        if isinstance(obj, _GobdLite):
            logger.info(
                'GAO ' + str(obj.ghid) + ' upstream registration starting.'
            )
            self._registered.add(obj.ghid)
    
            subscriptions = set()
            for remote in self._upstream_remotes:
                if remote.has_connection:
                    subscriptions.add(
                        make_background_future(remote.subscribe(obj.ghid))
                    )
                
            # Need to make sure it's not empty
            if subscriptions:
                await asyncio.wait(
                    fs = subscriptions,
                    return_when = asyncio.ALL_COMPLETED
                )
        
        else:
            logger.debug(
                'GAO ' + str(obj.ghid) + ' cannot be registered upstream: ' +
                'invalid object type: ' + str(type(obj))
            )
    
    @fixture_noop
    @public_api
    def _deregister(self, ghid):
        ''' Finalizer for GAO objects that executes async deregister()
        from within the event loop. Must be called from within our own
        thread, which it should be (finalizers are called from object
        thread, and all gao must be created within event loop's thread).
        '''
        logger.debug('_deregister finalizer called for ' + str(ghid))
        if self._clear_q is not None:
            # This needs to be a function, not a coro, so use nowait.
            self._clear_q.put_nowait(ghid)
    
    @fixture_noop
    @public_api
    async def deregister(self, ghid):
        ''' Tells the salmonator to stop listening for upstream object
        updates.
        '''
        logger.debug('Deregistering updates for ' + str(ghid))
        # This should maybe use remove instead of discard?
        self._registered.discard(ghid)
    
        unsubscriptions = set()
        for remote in self._upstream_remotes:
            unsubscriptions.add(
                make_background_future(remote.unsubscribe(ghid))
            )
        
        # Need to make sure it's not empty
        if unsubscriptions:
            await asyncio.wait(
                fs = unsubscriptions,
                return_when = asyncio.ALL_COMPLETED
            )
    
    @fixture_noop
    @public_api
    async def push(self, ghid):
        ''' Push a single ghid to all remotes.
        
        TODO: smartly defer any pushes to unavailable remotes using
        "if remote.has_connection()"
        '''
        data = await self._librarian.retrieve(ghid)
        
        tasks = set()
        for remote in self._upstream_remotes:
            if remote.has_connection:
                tasks.add(
                    make_background_future(remote.publish(data))
                )
            else:
                self._deferred.append(remote, data)
            
        # Need to make sure it's not empty
        if tasks:
            await asyncio.wait(
                fs = tasks,
                return_when = asyncio.ALL_COMPLETED
            )
    
    @fixture_noop
    @public_api
    async def pull(self, ghid):
        ''' Gets a ghid from upstream. Returns on the first result. Note
        that this is not meant to be called on a dynamic address, as a
        subs update from a slower remote would always be overridden by
        the faster one.
        '''
        pull_complete = None
        tasks_available = set()
        task_to_remote = {}
        for remote in self._upstream_remotes:
            if remote.has_connection:
                task = asyncio.ensure_future(
                    self._attempt_pull_single(ghid, remote)
                )
                tasks_available.add(task)
                task_to_remote[task] = remote
            
        # Wait until the first successful task completion
        # Note that this also shields us against having no tasks
        while tasks_available and not pull_complete:
            finished, pending = await asyncio.wait(
                fs = tasks_available,
                return_when = asyncio.FIRST_COMPLETED
            )
        
            # Despite FIRST_COMPLETED, asyncio may return more than one task
            for task in finished:
                # The task finished, so discard it from the available and reap
                # any exception
                tasks_available.discard(task)
                exc = task.exception()
                
                # If there's been an exception, continue waiting for the rest.
                # Downgrade this to info, since things may be missing from
                # persisters all the time. Let the parent log if, for example,
                # it's missing everywhere.
                if exc is not None:
                    remote = task_to_remote[task]
                    logger.info(
                        'Error while pulling from remote at ' +
                        remote._conn_desc + ':\n' +
                        ''.join(traceback.format_tb(exc.__traceback__)) +
                        repr(exc)
                    )
                    pull_complete = None
                
                # Completed successfully, but it could be a 404 (or other
                # error), which would present as result() = False.
                # Instead of letting the while loop handle this, since more
                # than one task can complete simultaneously, make sure we don't
                # already have a finalized result before blindly assigning the
                # task's result.
                elif not pull_complete:
                    pull_complete = task.result()
                    
                # Multiple tasks completed at once. An earlier one was
                # successful. We need to grab the result to suppress asyncio
                # complaints.
                else:
                    task.result()
                
        # No dice. Either finished is None (no remotes), None (no successful
        # pulls), or False (exactly one remote had the object, but it was
        # unloadable). Raise.
        if not pull_complete:
            raise UnavailableUpstream(
                'Object was unavailable or unacceptable at all '
                'currently-registered remotes.'
            )
            
        # Log success.
        else:
            logger.debug(
                'Successful remote pull for {!s}. Handling...'.format(ghid)
            )
        
        # We may still have some pending tasks. Cancel them. Note that we have
        # not yielded control to the event loop, so there is no race.
        for task in pending:
            logger.debug('Cancelling pending pulls.')
            task.cancel()
            
        # Now handle the result.
        await self._handle_successful_pull(pull_complete)
            
    async def _handle_successful_pull(self, maybe_obj):
        ''' Dispatches the object in the successful pull.
        
        maybe_obj can be either True (if the object already existed
        locally), or the object itself (if the object was new).
        '''
        # If we got a gobd, make sure its target is in the librarian
        if isinstance(maybe_obj, _GobdLite):
            if not (await self._librarian.contains(maybe_obj.target)):
                # Catch unavailableupstream and log a warning.
                # TODO: add logic to retry a few times and then discard
                try:
                    await self.pull(maybe_obj.target)
                    
                except UnavailableUpstream:
                    logger.warning(
                        'Received a subscription notification for ' +
                        str(maybe_obj.ghid) + ', but the sub\'s target was '
                        'missing both locally and upstream.'
                    )
        
        logger.debug('Successful pull handled.')
    
    @fixture_noop
    @public_api
    async def attempt_pull(self, ghid, quiet=False):
        ''' Grabs the ghid from remotes, if available, and puts it into
        the ingestion pipeline.
        '''
        # TODO: check locally, run _inspect, check if mutable before blindly
        # pulling.
        try:
            await self.pull(ghid)
            
        except UnavailableUpstream:
            if not quiet:
                raise
            # Suppress errors if we were called quietly.
            else:
                logger.info(
                    'Object was unavailable or unacceptable upstream, but '
                    'pull was called quietly: ' + str(ghid)
                )
        
    async def _attempt_pull_single(self, ghid, remote):
        ''' Attempt to fetch a single object from a single remote. If
        successful, put it into the ingestion pipeline.
        '''
        # This may error, but any errors here will be caught by the parent.
        data = await remote.get(ghid)
        
        # Call as remotable=False to avoid infinite loops.
        ingested = await self._percore.ingest(data, remotable=False)
        
        # Note that ingest can either return None, if we already have
        # the object, or the object itself, if it's new.
        if ingested:
            return ingested
        else:
            return True
            
    async def bootstrap(self, account):
        ''' Bootstrapping publishes our identity upstream.
        
        ONLY do this if we already have a connection available there, or
        we will hang.
        '''
        for remote in self._upstream_remotes:
            if remote.has_connection:
                await remote.publish(account._identity.second_party.packed)
                await remote.subscribe(account._identity.ghid)
            
        self._bootstrapped = True
    
    @fixture_noop
    @public_api
    async def restore_connection(self, remote, connection):
        ''' Start or re-start a connection.
        '''
        # We need to subscribe to our identity if we're restoring a terminated
        # connection. If we haven't bootstrapped though, that will be handled
        # there.
        if self._bootstrapped:
            await remote.publish(self._golcore._identity.second_party.packed)
            await self._remote_protocol.subscribe(
                connection,
                self._golcore.whoami
            )
        
        # For every every active (salmonator-registered) GAO's ghid...
        tasks = set()
        for registrant in self._registered:
            # Record that we need to perform...
            tasks.add(
                # ...as a background future (which handles its own errors)...
                make_background_future(
                    # ...a subscription call with our connection at protocol
                    self._remote_protocol.subscribe(connection, registrant)
                )
            )
        
        # We need to make sure there's at least one task.
        if tasks:
            # And now just wait for all of that to complete.
            await asyncio.wait(
                fs = tasks,
                return_when = asyncio.ALL_COMPLETED
            )
        
        # Now, we need to destructively iterate over our deferreds until the
        # remote list is exhausted. We have to do this in order, sequentially,
        # and serially, because eg. containers require bindings, etc.
        to_push = collections.deque(self._deferred.pop_key(remote))
        try:
            while to_push:
                data = to_push.popleft()
                await self._remote_protocol.publish(connection, data)
        
        # Restore the last to_push, put it back into the deferred list, and
        # then re-raise
        except Exception:
            to_push.appendleft(data)
            self._deferred.extend(remote, to_push)
            raise class IPCServerProtocol(_IPCSerializer, metaclass=RequestResponseAPI,
                        error_codes=ERROR_CODES, default_version=b'\x00\x00'):
    ''' Defines the protocol for IPC, with handlers specific to servers.
    '''
    _dispatch = weak_property('__dispatch')
    _oracle = weak_property('__oracle')
    _golcore = weak_property('__golcore')
    _rolodex = weak_property('__rolodex')
    _salmonator = weak_property('__salmonator')
        
    @fixture_api
    def __init__(self, whoami, *args, **kwargs):
        super(IPCServerProtocol.__fixture__, self).__init__(*args, **kwargs)
        self._whoami = whoami
        self.shares = collections.deque()
        self.updates = collections.deque()
        self.deletes = collections.deque()
        self.share_successes = collections.deque()
        self.share_failures = collections.deque()
        
    def assemble(self, golcore, oracle, dispatch, rolodex, salmonator):
        # Chicken, egg, etc.
        self._golcore = golcore
        self._oracle = oracle
        self._dispatch = dispatch
        self._rolodex = rolodex
        self._salmonator = salmonator
        
    @request(b'+T')
    async def set_token(self, connection, token=None):
        ''' Register an existing token or get a new token, or notify an
        app of its existing token.
        '''
        # On the server side, this will only be implemented once actual
        # application launching is available.
        raise NotImplementedError()
        
    @set_token.request_handler
    async def set_token(self, connection, body):
        ''' Handles token-setting requests.
        '''
        token = AppToken(body)
        
        # Getting a new token.
        if token == AppToken.null():
            token = await self._dispatch.start_application(connection)
            logger.info(''.join((
                'CONN ', str(connection), ' generating new token: ', str(token)
            )))
        
        # Setting an existing token, but the connection already exists.
        elif self._dispatch.token_lookup(connection):
            raise IPCError(
                'Attempt to reregister a new concurrent token for an ' +
                'existing connection. Each app may use only one token.'
            )
            
        # Setting an existing token, but the token already exists.
        elif self._dispatch.connection_lookup(token) is not None:
            raise IPCError(
                'Attempt to reregister a new concurrent connection for ' +
                'an existing token. Each app may only use one connection.'
            )
        
        # Setting an existing token, with valid state.
        else:
            logger.info(''.join((
                'CONN ', str(connection), ' registering existing token: ',
                str(token)
            )))
            await self._dispatch.start_application(connection, token)
        
        return token
        
    @request(b'+A')
    async def register_api(self, connection):
        ''' Registers the application as supporting an API. Client only.
        '''
        raise NotImplementedError()
        
    @register_api.request_handler
    async def register_api(self, connection, body):
        ''' Handles API registration requests. Server only.
        '''
        api_id = ApiID.from_bytes(body)
        await self._dispatch.add_api(connection, api_id)
        
        return b'\x01'
        
    @request(b'-A')
    async def deregister_api(self, connection):
        ''' Removes any existing registration for the app supporting an
        API. Client only.
        '''
        raise NotImplementedError()
        
    @deregister_api.request_handler
    async def deregister_api(self, connection, body):
        ''' Handles API deregistration requests. Server only.
        '''
        api_id = ApiID.from_bytes(body)
        await self._dispatch.remove_api(connection, api_id)
        
        return b'\x01'
        
    @public_api
    @request(b'?I')
    async def whoami(self, connection):
        ''' Get the current hypergolix fingerprint, or notify an app of
        the current hypergolix fingerprint.
        '''
        # On the server side, this will only be implemented once actual
        # application launching is available.
        raise NotImplementedError()
        
    @whoami.request_handler
    async def whoami(self, connection, body):
        ''' Handles whoami requests.
        '''
        ghid = self._golcore.whoami
        return bytes(ghid)
        
    @request(b'>$')
    async def get_startup_obj(self, connection):
        ''' Request a startup object, or notify an app of its declared
        startup object.
        '''
        token = self._dispatch.token_lookup(connection)
        ghid = await self._dispatch.get_startup_obj(token)
        
        if ghid is not None:
            return bytes(ghid)
        else:
            return b''
        
    @get_startup_obj.request_handler
    async def get_startup_obj(self, connection, body):
        ''' Handles requests for startup objects.
        '''
        token = self._dispatch.token_lookup(connection)
        ghid = await self._dispatch.get_startup_obj(token)
        
        if ghid is not None:
            return bytes(ghid)
        else:
            return b''
        
    @request(b'+$')
    async def register_startup_obj(self, connection, ghid):
        ''' Register a startup object. Client only.
        '''
        raise NotImplementedError()
        
    @register_startup_obj.request_handler
    async def register_startup_obj(self, connection, body):
        ''' Handles startup object registration. Server only.
        '''
        ghid = Ghid.from_bytes(body)
        await self._dispatch.register_startup(connection, ghid)
        return b'\x01'
        
    @request(b'-$')
    async def deregister_startup_obj(self, connection):
        ''' Register a startup object. Client only.
        '''
        raise NotImplementedError()
        
    @deregister_startup_obj.request_handler
    async def deregister_startup_obj(self, connection, body):
        ''' Handles startup object registration. Server only.
        '''
        await self._dispatch.deregister_startup(connection)
        return b'\x01'
        
    @request(b'>O')
    async def get_obj(self, connection):
        ''' Get an object with the specified address. Client only.
        '''
        raise NotImplementedError()
        
    @get_obj.request_handler
    async def get_obj(self, connection, body):
        ''' Handles requests for an object. Server only.
        '''
        ghid = Ghid.from_bytes(body)
        obj = await self._oracle.get_object(
            gaoclass = _Dispatchable,
            ghid = ghid,
            api_id = None,  # Let _pull() apply this.
            state = None,   # Let _pull() apply this.
            dispatch = self._dispatch,
            ipc_protocol = self,
            account = self._dispatch._account
        )
        
        await self._dispatch.track_object(connection, obj.ghid)
            
        if isinstance(obj.state, Ghid):
            is_link = True
            state = bytes(obj.state)
        else:
            is_link = False
            state = obj.state
            
        # For now, anyways.
        # Note: need to add some kind of handling for legroom.
        _legroom = None
        
        # Not a big fan of how this works, seems inelegant to me
        private = bool(self._dispatch.private_parent_lookup(obj.ghid))
        
        return self._pack_object_def(
            obj.ghid,
            obj.author,
            state,
            is_link,
            obj.api_id,
            private,
            obj.dynamic,
            _legroom
        )
        
    @request(b'+O')
    async def new_obj(self, connection):
        ''' Create a new object, or notify an app of a new object
        created by a concurrent instance of the app on a different
        hypergolix session.
        '''
        # Not currently supported.
        raise NotImplementedError()
        
    @new_obj.request_handler
    async def new_obj(self, connection, body):
        ''' Handles requests for new objects.
        '''
        (address,    # Unused and set to None.
         author,     # Unused and set to None.
         state,
         is_link,
         api_id,
         private,
         dynamic,
         legroom) = self._unpack_object_def(body)
        
        if is_link:
            raise NotImplementedError('Linked objects are not yet supported.')
            state = Ghid.from_bytes(state)
        
        obj = await self._oracle.new_object(
            gaoclass = _Dispatchable,
            dispatch = self._dispatch,
            ipc_protocol = self,
            state = state,
            dynamic = dynamic,
            legroom = legroom,
            api_id = api_id,
            account = self._dispatch._account
        )
            
        # Add the endpoint as a listener.
        await self._dispatch.register_object(connection, obj.ghid, private)
        await self._dispatch.track_object(connection, obj.ghid)
        
        return bytes(obj.ghid)
    
    @public_api
    @request(b'!O')
    async def update_obj(self, connection, ghid):
        ''' Update an object or notify an app of an incoming update.
        '''
        try:
            obj = await self._oracle.get_object(
                gaoclass = _Dispatchable,
                ghid = ghid,
                api_id = None,  # Let _pull() apply this.
                state = None,   # Let _pull() apply this.
                dispatch = self._dispatch,
                ipc_protocol = self,
                account = self._dispatch._account
            )
            
        # No CancelledError catch necessary because we're re-raising any exc
            
        except Exception:
            # At some point we'll need some kind of proper handling for this.
            logger.error(
                'Failed to retrieve object at ' + str(ghid) + '\n' +
                ''.join(traceback.format_exc())
            )
            raise
            
        else:
            return self._pack_object_def(
                obj.ghid,
                obj.author,
                obj.state,
                False,      # is_link (currently unsupported)
                obj.api_id,
                None,       # private
                obj.dynamic,
                None        # legroom
            )
    
    @update_obj.fixture
    async def update_obj(self, connection, ghid):
        ''' Manual no-op fixture, courtesy of descriptors not being
        callable or whatever.
        '''
        self.updates.append((connection, ghid))
    
    @update_obj.request_handler
    async def update_obj(self, connection, body):
        ''' Handles update object requests.
        '''
        logger.debug('Handling update request from ' + str(connection))
        (ghid,
         author,    # Unused and set to None.
         state,
         is_link,
         api_id,    # Unused and set to None.
         private,   # TODO: use this.
         dynamic,   # Unused and set to None.
         legroom   # TODO: use this.
         ) = self._unpack_object_def(body)
        
        if is_link:
            raise NotImplementedError('Linked objects are not yet supported.')
            state = Ghid.from_bytes(state)
            
        obj = await self._oracle.get_object(
            gaoclass = _Dispatchable,
            ghid = ghid,
            api_id = None,  # Let _pull() apply this.
            state = None,   # Let _pull() apply this.
            dispatch = self._dispatch,
            ipc_protocol = self,
            account = self._dispatch._account
        )
        obj.state = state
        
        # Converting a private object to a public one
        if self._dispatch.private_parent_lookup(ghid):
            if not private:
                await self._dispatch.make_public(ghid)
        
        else:
            if private:
                raise CatOuttaBagError(
                    'Once made public, objects cannot be made private. They ' +
                    'have already been distributed to other applications.'
                )
            
        await obj.push()
        
        # Schedule an update in the background.
        make_background_future(
            self._dispatch.distribute_update(
                obj.ghid,
                skip_conn = connection
            )
        )
        
        return b'\x01'
        
    @request(b'~O')
    async def sync_obj(self, connection):
        ''' Manually force Hypergolix to check an object for updates.
        Client only.
        '''
        raise NotImplementedError()
        
    @sync_obj.request_handler
    async def sync_obj(self, connection, body):
        ''' Handles manual syncing requests. Server only.
        '''
        ghid = Ghid.from_bytes(body)
        await self._salmonator.attempt_pull(ghid)
        return b'\x01'
    
    @public_api
    @request(b'@O')
    async def share_obj(self, connection, ghid, origin, api_id):
        ''' Request an object share or notify an app of an incoming
        share.
        '''
        if origin is None:
            origin = self._golcore.whoami
        
        return bytes(ghid) + bytes(origin) + bytes(api_id)
        
    @share_obj.fixture
    async def share_obj(self, connection, ghid, origin, api_id):
        ''' Manual no-op fixture, courtesy of descriptors not being
        callable or whatever.
        '''
        self.shares.append(
            _ShareLog(connection, ghid, origin, api_id)
        )
        
    @share_obj.request_handler
    async def share_obj(self, connection, body):
        ''' Handles object share requests.
        '''
        ghid = Ghid.from_bytes(body[0:65])
        recipient = Ghid.from_bytes(body[65:130])
        
        # Instead of forbidding unregistered apps from sharing objects,
        # go for it, but document that you will never be notified of a
        # share success or failure without an app token.
        requesting_token = self._dispatch.token_lookup(connection)
        if requesting_token is None:
            logger.info(
                'CONN ' + str(connection) + ' is sharing ' + str(ghid) +
                ' with ' + str(recipient) + ' without defining an app ' +
                'token, and therefore cannot be notified of share success ' +
                'or failure.'
            )
            
        await self._rolodex.share_object(ghid, recipient, requesting_token)
        return b'\x01'
    
    @public_api
    @request(b'^S')
    async def notify_share_success(self, connection, ghid, recipient):
        ''' Notify app of successful share. Server only.
        '''
        return bytes(ghid) + bytes(recipient)
    
    @notify_share_success.fixture
    async def notify_share_success(self, connection, ghid, recipient):
        ''' Manual no-op fixture, courtesy of descriptors not being
        callable or whatever.
        '''
        self.share_successes.append((connection, ghid, recipient))
    
    @notify_share_success.request_handler
    async def notify_share_success(self, connection, body):
        ''' Handles app notifications for successful shares. Client
        only.
        '''
        raise NotImplementedError()
    
    @public_api
    @request(b'^F')
    async def notify_share_failure(self, connection, ghid, recipient):
        ''' Notify app of unsuccessful share. Server only.
        '''
        return bytes(ghid) + bytes(recipient)
        
    @notify_share_failure.fixture
    async def notify_share_failure(self, connection, ghid, recipient):
        ''' Manual no-op fixture, courtesy of descriptors not being
        callable or whatever.
        '''
        self.share_failures.append((connection, ghid, recipient))
    
    @notify_share_failure.request_handler
    async def notify_share_failure(self, connection, body):
        ''' Handles app notifications for unsuccessful shares. Client
        only.
        '''
        raise NotImplementedError()
        
    @request(b'*O')
    async def freeze_obj(self, connection):
        ''' Creates a new static copy of the object, or notifies an app
        of a frozen copy of an existing object created by a concurrent
        instance of the app.
        '''
        # Not currently supported.
        raise NotImplementedError()
        
    @freeze_obj.request_handler
    async def freeze_obj(self, connection, body):
        ''' Handles object freezing requests.
        '''
        ghid = Ghid.from_bytes(body)
        obj = await self._oracle.get_object(
            gaoclass = _Dispatchable,
            ghid = ghid,
            api_id = None,  # Let _pull() apply this.
            state = None,   # Let _pull() apply this.
            dispatch = self._dispatch,
            ipc_protocol = self,
            account = self._dispatch._account
        )
        frozen_address = await obj.freeze()
        
        return bytes(frozen_address)
        
    @request(b'#O')
    async def hold_obj(self, connection):
        ''' Creates a new static binding for the object, or notifies an
        app of a static binding created by a concurrent instance of the
        app.
        '''
        # Not currently supported.
        raise NotImplementedError()
        
    @hold_obj.request_handler
    async def hold_obj(self, connection, body):
        ''' Handles object holding requests.
        '''
        ghid = Ghid.from_bytes(body)
        obj = await self._oracle.get_object(
            gaoclass = _Dispatchable,
            ghid = ghid,
            api_id = None,  # Let _pull() apply this.
            state = None,   # Let _pull() apply this.
            dispatch = self._dispatch,
            ipc_protocol = self,
            account = self._dispatch._account
        )
        await obj.hold()
        return b'\x01'
        
    @request(b'-O')
    async def discard_obj(self, connection):
        ''' Stop listening to object updates. Client only.
        '''
        raise NotImplementedError()
        
    @discard_obj.request_handler
    async def discard_obj(self, connection, body):
        ''' Handles object discarding requests. Server only.
        '''
        ghid = Ghid.from_bytes(body)
        await self._dispatch.untrack_object(connection, ghid)
        return b'\x01'
    
    @public_api
    @request(b'XO')
    async def delete_obj(self, connection, ghid):
        ''' Request an object deletion or notify an app of an incoming
        deletion.
        '''
        if not isinstance(ghid, Ghid):
            raise TypeError('ghid must be type Ghid or similar.')
        
        return bytes(ghid)
    
    @delete_obj.fixture
    async def delete_obj(self, connection, ghid):
        ''' Manual no-op fixture, courtesy of descriptors not being
        callable or whatever.
        '''
        self.deletes.append((connection, ghid))
    
    @delete_obj.request_handler
    async def delete_obj(self, connection, body):
        ''' Handles object deletion requests.
        '''
        ghid = Ghid.from_bytes(body)
        obj = await self._oracle.get_object(
            gaoclass = _Dispatchable,
            ghid = ghid,
            api_id = None,  # Let _pull() apply this.
            state = None,   # Let _pull() apply this.
            dispatch = self._dispatch,
            ipc_protocol = self,
            account = self._dispatch._account
        )
        await self._dispatch.untrack_object(connection, ghid)
        # TODO: shift to a dispatch.delete_object method that can ignore this
        # connection when the intevitable "deleted that object!" warning comes
        # back
        await obj.delete()
        return b'\x01' class Privateer(metaclass=API):
    ''' Lookup system to get secret from ghid. Loopsafe, but NOT
    threadsafe.
    '''
    _account = weak_property('__account')
    _golcore = weak_property('__golcore')
    
    @public_api
    def __init__(self, *args, **kwargs):
        ''' Temporarily invalidate (but init) the various secrets
        lookups.
        '''
        super().__init__(*args, **kwargs)
        
        # These must be bootstrapped.
        self._secrets_persistent = None
        self._secrets_quarantine = None
        self._secrets = None
        # These two are local only, but included to make the code more explicit
        self._secrets_staging = None
        self._secrets_local = None
        
    @__init__.fixture
    def __init__(self, identity, *args, **kwargs):
        ''' Directly inject a local identity, and then add a ghidproxy.
        '''
        super(Privateer.__fixture__, self).__init__(*args, **kwargs)
        
        self._identity = identity
        self._secrets_persistent = {}
        self._secrets_staging = {}
        self._secrets_local = {}
        self._secrets_quarantine = {}
        self._secrets = collections.ChainMap(
            self._secrets_persistent,
            self._secrets_local,
            self._secrets_staging,
            self._secrets_quarantine,
        )
        self._secrets_committed = collections.ChainMap(
            self._secrets_persistent,
            self._secrets_local
        )
        
    @fixture_api
    def RESET(self):
        # Clear errything internal.
        self._secrets_persistent.clear()
        self._secrets_staging.clear()
        self._secrets_local.clear()
        self._secrets_quarantine.clear()
            
    def __contains__(self, ghid):
        ''' Check if we think we know a secret for the ghid.
        '''
        return ghid in self._secrets
        
    def assemble(self, golcore):
        # Chicken, meet egg.
        self._golcore = golcore
    
    @fixture_noop
    @public_api
    def bootstrap(self, account):
        ''' Initializes the privateer into a distributed state.
        persistent is a GaoDict
        '''
        self._account = account
        
        # We very obviously need to be able to look up what secrets we have.
        # Lookups: <container ghid>: <container secret>
        self._secrets_persistent = account.privateer_persistent
        self._secrets_staging = {}
        self._secrets_local = {}
        self._secrets_deprecated = FiniteDict(maxlen=100)
        self._secrets_quarantine = account.privateer_quarantine
        self._secrets = collections.ChainMap(
            self._secrets_persistent,
            self._secrets_local,
            self._secrets_deprecated,
            self._secrets_staging,
            self._secrets_quarantine,
        )
        self._secrets_committed = collections.ChainMap(
            self._secrets_persistent,
            self._secrets_local
        )
        
        # Note that we just overwrote any chains that were created when we
        # initially loaded the three above resources. So, we may need to
        # re-create them. But, we can use a fake container address for two of
        # them, since they use a different secrets tracking mechanism.
        
    @public_api
    def new_secret(self):
        # Straight pass-through to the golix new_secret bit.
        return self._golcore._identity.new_secret()
        
    @new_secret.fixture
    def new_secret(self):
        # We have a direct identity for the fixture.
        return self._identity.new_secret()
        
    def get(self, ghid):
        ''' Get a secret for a ghid, regardless of status.
        
        Raises KeyError if secret is not present.
        '''
        try:
            return self._secrets[ghid]
        
        except KeyError:
            raise UnknownSecret(str(ghid)) from None
        
    def stage(self, ghid, secret):
        ''' Preliminarily set a secret for a ghid.
        
        If a secret is already staged for that ghid and the ghids are
        not equal, raises ConflictingSecrets.
        '''
        # Is there a reason we can't verify container-ness here?
        logger.debug(''.join((
            'GAO ',
            str(ghid),
            ' secret staging.'
        )))
        self._stage(ghid, secret, self._secrets_staging)
                
    def _stage(self, ghid, secret, lookup):
        ''' Raw staging, bypassing modlock. Only accessed directly
        during bootstrapping.
        '''
        if not isinstance(secret, Secret):
            raise TypeError('Invalid secret type: ' + type(secret).__name__)
        
        if ghid in self._secrets:
            if self._secrets[ghid] != secret:
                self._calc_and_log_diff(self._secrets[ghid], secret)
                raise ConflictingSecrets(
                    'Non-matching secret already known for ' + str(ghid)
                )
        else:
            lookup[ghid] = secret
            
    def quarantine(self, ghid, secret):
        ''' Store a secret temporarily, but distributedly, separate from
        the primary secret store. Used only for incoming secrets in
        shares that have not yet been opened.
        '''
        logger.debug(''.join((
            'GAO ',
            str(ghid),
            ' secret quarantining.'
        )))
        self._stage(ghid, secret, self._secrets_quarantine)
            
    def unstage(self, ghid):
        ''' Remove a staged secret, probably due to a SecurityError.
        Returns the secret.
        '''
        secret = self._secrets_quarantine.pop(ghid, None)
        secret = secret or self._secrets_staging.pop(ghid, None)
        
        if secret is None:
            raise UnknownSecret(
                'No currently staged secret for GHID ' + str(ghid)
            )
                
        return secret
        
    def commit(self, ghid, localize=False):
        ''' Store a secret "permanently". The secret must either:
        1. Be staged, XOR
        2. Be quarantined, XOR
        3. Be committed
        
        Other states will raise UnknownSecret (a subclass of KeyError).
        
        Note that this relies upon staging logic enforcing consistency
        of secrets! If that changes to allow the staging conflicting
        secrets, this will need to be modified.
        
        This is transactional and atomic; any errors (ex: ValueError
        above) will return its state to the previous.
        
        With localize=True, the secret is committed locally only. This
        is intended to be used in combination with a master secret; in
        that case, there's no need to store frame keys -- not to mention
        that it would cause an infinitely recursive chain of secret
        committing.
        
        '''
        # Catch anything with an unknown secret
        if ghid not in self._secrets:
            raise UnknownSecret(
                'Secret not currently staged for GHID ' + str(ghid)
            )
        
        # Note that we cannot stage or quarantine a secret that we already
        # have committed. If that secret matches what we already have, it
        # will indempotently and silently exit, without staging.
        elif ghid in self._secrets_committed:
            return
        
        secret = self._secrets_quarantine.pop(ghid, None)
        secret = secret or self._secrets_staging.pop(ghid, None)
            
        # If it's a bootstrap target, just locally commit it.
        if localize:
            logger.debug(''.join((
                'GAO ',
                str(ghid),
                ' secret committed locally.'
            )))
            self._secrets_local[ghid] = secret
            
        # Nope. Commit globally.
        else:
            logger.debug(''.join((
                'GAO ',
                str(ghid),
                ' secret committed globally.'
            )))
            self._secrets_persistent[ghid] = secret
    
    @classmethod
    def _calc_and_log_diff(cls, secret, other):
        ''' Calculate the difference between two secrets.
        '''
        try:
            cipher_match = (secret.cipher == other.cipher)
            
            if cipher_match:
                key_comp = cls._bitdiff(secret.key, other.key)
                seed_comp = cls._bitdiff(secret.seed, other.seed)
                logger.info('Keys are ' + str(key_comp) + '%% different.')
                logger.info('Seeds are ' + str(seed_comp) + '%% different.')
                
            else:
                logger.info('Secret ciphers do not match. Cannot compare.')
            
        except AttributeError:
            logger.error(
                'Attribute error while diffing secrets. Type mismatch? \n'
                '    ' + repr(type(secret)) + '\n'
                '    ' + repr(type(other))
            )
            
    @staticmethod
    def _bitdiff(this_bytes, other_bytes):
        ''' Calculates the percent of different bits between two byte
        strings.
        '''
        if len(this_bytes) == 0 or len(other_bytes) == 0:
            # By returning None, we can explicitly say we couldn't perform the
            # comparison, whilst also preventing math on a comparison.
            return None
        
        # Mask to extract each bit.
        masks = [
            0b00000001,
            0b00000010,
            0b00000100,
            0b00001000,
            0b00010000,
            0b00100000,
            0b01000000,
            0b10000000,
        ]
        
        # Counters for bits.
        diffbits = 0
        totalbits = 0
        
        # First iterate over each byte.
        for this_byte, other_byte in zip(this_bytes, other_bytes):
            
            # Now, using the masks, iterate over each bit.
            for mask in masks:
                # Extract the bit using bitwise AND.
                this_masked = mask & this_byte
                other_masked = mask & other_byte
                
                # Do a bool comparison of the bits, and add any != results to
                # diffbits. Note that 7 + False == 7 and 7 + True == 8
                diffbits += (this_masked != other_masked)
                totalbits += 1
                
        # Finally, calculate a percent difference
        doubdiff = diffbits / totalbits
        return int(doubdiff * 100)
        
    def abandon(self, ghid, quiet=True):
        ''' Remove a secret. If quiet=True, silence any KeyErrors.
        '''
        # Short circuit any tests if quiet is enabled
        if not quiet and ghid not in self._secrets:
            raise UnknownSecret('Secret not found for ' + str(ghid))
            
        self._secrets_persistent.pop(ghid, None)
        self._secrets_local.pop(ghid, None)
        self._secrets_quarantine.pop(ghid, None)
        self._secrets_staging.pop(ghid, None)
        self._secrets_deprecated.pop(ghid, None)
        
    @fixture_noop
    @public_api
    def deprecate(self, ghid, quiet=True):
        ''' Deprecate a secret. It will still be available locally, but
        will be removed from shared nonlocal state.
        '''
        # Short circuit any tests if quiet is enabled
        if not quiet and ghid not in self._secrets:
            raise UnknownSecret('Secret not found for ' + str(ghid))
        
        else:
            secret1 = self._secrets_persistent.pop(ghid, None)
            secret2 = self._secrets_local.pop(ghid, None)
            secret3 = self._secrets_quarantine.pop(ghid, None)
            secret4 = self._secrets_staging.pop(ghid, None)
            secret5 = self._secrets_deprecated.pop(ghid, None)
            
            secret = secret1 or secret2 or secret3 or secret4 or secret5
            
            self._secrets_deprecated[ghid] = secret
        
    def ratchet_chain(self, proxy, current_target, master_secret=None):
        ''' Gets a new secret for the proxy. Returns the secret, and
        flags the ratchet as in-progress.
        
        If master_secret is supplied, we will use the bootstrap ratchet.
        '''
        try:
            # Without a master_secret, perform a ratchet with the existing
            # secret for the current target.
            if master_secret is None:
                existing_secret = self.get(current_target)
                ratcheted = self._ratchet(
                    secret = existing_secret,
                    proxy = proxy,
                    salt_ghid = current_target
                )
            
            # With a master_secret, perform a ratchet with the master as a
            # static "seed" secret.
            else:
                existing_secret = None
                ratcheted = self._ratchet(
                    secret = master_secret,
                    proxy = proxy,
                    salt_ghid = current_target
                )
                
            return ratcheted
            
        except Exception:
            msec_str = (' with master secret of type ' +
                        type(master_secret).__name__) * bool(master_secret)
            xsec_str = (' with existing secret of type ' +
                        type(existing_secret).__name__) * (not master_secret)
            
            raise RatchetError('Failed ratchet for unknown reasons: ' +
                               str(proxy) + ' from ' + str(current_target) +
                               msec_str + xsec_str) from None
            
    def heal_chain(self, proxy, target_vector, master_secret=None):
        ''' Heals the ratchet for a binding using the gao. Call this any
        time an agent RECEIVES a new EXTERNAL ratcheted object. Stages
        the resulting secret for the most recent frame in binding, BUT
        DOES NOT RETURN (or commit) IT.
        
        Note that this is necessarily being called before the object is
        available at the oracle. It should, however, be available at
        both the ghidproxy and the librarian.
        
        NOTE that the binding is the LITEWEIGHT version from the
        librarian already, so its ghid is already the dynamic one.
        '''
        if len(target_vector) < 1:
            raise RatchetError('Target vector has no historical references.')
            
        # Master secret ratchets never break, and can alway recover from the
        # most recent frame. BUT, to increase failure tolerance, we are going
        # to stage any and all missing secrets, in case we need to load an
        # earlier container.
        if master_secret:
            aktueller_index = len(target_vector) - 1
            
        # Other ratchets, on the other hand, have to start somewhere.
        else:
            # Go from back to front, finding the most recent target vector that
            # exists locally.
            available = [target in self._secrets for target in target_vector]
            # If we didn't find ANY of them, then we're broken.
            if not any(available):
                raise RatchetError('Broken ratchet: ' + str(proxy))
        
            # Get the index for the most recent target that we already know a
            # secret for
            aktueller_index = available.index(True)
        
        # Go from the oldest to the newest, but start after the first one,
        # because we verified that above.
        # We can't use reversed(enumerate(target_vector)) without re-casting
        # target_vector into a dedicated sequence. We'd like to keep the
        # original iterator, so do it manually.
        for ii in range(aktueller_index - 1, -1, -1):
            target = target_vector[ii]
            # Skip the first one (deques don't support slicing)
            # Skip if we already have a secret
            if target in self._secrets:
                continue
            
            # Only do something if we don't know the secret
            else:
                secret = self.ratchet_chain(
                    proxy,
                    current_target = target_vector[ii + 1],
                    master_secret = master_secret
                )
                self.stage(target, secret)
        
    @staticmethod
    def _ratchet(secret, proxy, salt_ghid):
        ''' Ratchets a key using HKDF-SHA512, using the associated
        address as salt. For dynamic files, this should be the previous
        frame ghid (not the dynamic ghid).
        
        Note: this ratchet is bound to a particular dynamic address. The
        ratchet algorithm is:
        
        new_key = HKDF-SHA512(
            IKM = old_secret, (secret IV/nonce | secret key)
            salt = old_frame_ghid, (entire 65 bytes)
            info = dynamic_ghid, (entire 65 bytes)
            new_key_length = len(IV/nonce) + len(key),
            num_keys = 1,
        )
        '''
        cls = type(secret)
        cipher = secret.cipher
        version = secret.version
        len_seed = len(secret.seed)
        len_key = len(secret.key)
        source = bytes(secret.seed + secret.key)
        
        instance = hkdf.HKDF(
            algorithm = hashes.SHA512(),
            length = len_seed + len_key,
            salt = bytes(salt_ghid),
            info = bytes(proxy),
            backend = CRYPTO_BACKEND
        )
        ratcheted = instance.derive(source)
        
        return cls(
            cipher = cipher,
            version = version,
            key = ratcheted[:len_key],
            seed = ratcheted[len_key:]
        ) class Rolodex(metaclass=API):
    ''' Handles sharing, requests, etc.
    
    In the future, will maintain a contacts list to approve/reject
    incoming requests. In the further future, will maintain sharing
    pipelines, negotiated through handshakes, to perform sharing
    symmetrically between contacts.
    '''
    _account = weak_property('__account')
    _golcore = weak_property('__golcore')
    _ghidproxy = weak_property('__ghidproxy')
    _privateer = weak_property('__privateer')
    _percore = weak_property('__percore')
    _librarian = weak_property('__librarian')
    _salmonator = weak_property('__salmonator')
    _dispatch = weak_property('__dispatch')
    
    @public_api
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        
        # Persistent dict-like lookup for
        # request_ghid -> (request_target, request recipient)
        self._pending_requests = None
        # Lookup for <target_obj_ghid, recipient> -> set(<app_tokens>)
        self._outstanding_shares = None
        
    @__init__.fixture
    def __init__(self, *args, **kwargs):
        ''' Init the fixture.
        '''
        super(Rolodex.__fixture__, self).__init__(*args, **kwargs)
        
        self.shared = {}
        self._pending_requests = {}
        self._outstanding_shares = {}
        
    @fixture_api
    def RESET(self):
        ''' Super simple reset alias to __init__.
        '''
        self.shared.clear()
        self._pending_requests.clear()
        self._outstanding_shares.clear()
        
    @fixture_noop
    @public_api
    def bootstrap(self, account):
        ''' Initialize distributed state.
        '''
        self._account = account
        
        # Persistent dict-like lookup for
        # request_ghid -> (request_target, request recipient)
        self._pending_requests = account.rolodex_pending
        
        # These need to be distributed but aren't yet. TODO!
        # Lookup for <target_obj_ghid, recipient> -> set(<app_tokens>)
        self._outstanding_shares = account.rolodex_outstanding
        
    def assemble(self, golcore, ghidproxy, privateer, percore, librarian,
                 salmonator, dispatch):
        # Chicken, meet egg.
        self._golcore = golcore
        self._ghidproxy = ghidproxy
        self._privateer = privateer
        self._percore = percore
        self._librarian = librarian
        self._salmonator = salmonator
        self._dispatch = dispatch
        
    @public_api
    async def share_object(self, target, recipient, requesting_token):
        ''' Share a target ghid with the recipient.
        '''
        if not isinstance(target, Ghid):
            raise TypeError(
                'target must be Ghid or similar.'
            )
        if not isinstance(recipient, Ghid):
            raise TypeError(
                'recipient must be Ghid or similar.'
            )

        sharepair = _SharePair(target, recipient)
            
        # For now, this is just doing a handshake with some typechecking.
        await self._hand_object(*sharepair)
        
        if requesting_token is not None:
            self._outstanding_shares.add(sharepair, requesting_token)
            await self._account.flush()
            
    @share_object.fixture
    async def share_object(self, target, recipient, requesting_token):
        ''' Pretty basic.
        '''
        # Ignore requesting token, because we won't generally set it when
        # testing this.
        self.shared[target] = recipient
        
    async def _hand_object(self, target, recipient):
        ''' Initiates a handshake request with the recipient.
        '''
        if not (await self._librarian.contains(recipient)):
            await self._salmonator.attempt_pull(recipient, quiet=True)
        
        recipient_identity = \
            (await self._librarian.summarize(recipient)).identity
        
        # This is guaranteed to resolve the container fully.
        container_ghid = await self._ghidproxy.resolve(target)
        
        # TODO: fix Golix library so this isn't such a shitshow re:
        # breaking abstraction barriers.
        handshake = self._golcore._identity.make_handshake(
            target = target,
            secret = self._privateer.get(container_ghid)
        )
        
        request = self._golcore._identity.make_request(
            recipient = recipient_identity,
            request = handshake
        )
        
        # Note that this must be called before publishing to the persister, or
        # there's a race condition between them.
        self._pending_requests[request.ghid] = target, recipient
        
        # Note the potential race condition here. Should catch errors with the
        # persister in case we need to resolve pending requests that didn't
        # successfully post.
        await self._percore.direct_ingest(
            obj = _GarqLite.from_golix(request),
            packed = request.packed,
            remotable = True
        )
    
    @fixture_noop
    @public_api
    async def notification_handler(self, subscription, notification):
        ''' Callback to handle any requests.
        '''
        if not (await self._librarian.contains(notification)):
            await self._salmonator.attempt_pull(notification, quiet=True)
        
        # Note that the notification could also be a GDXX.
        request_or_debind = await self._librarian.summarize(notification)
        
        if isinstance(request_or_debind, _GarqLite):
            await self._handle_request(notification)
            
        elif isinstance(request_or_debind, _GdxxLite):
            # This case should only be relevant if we have multiple agents
            # logged in at separate locations at the same time, processing the
            # same GARQs.
            await self._handle_debinding(request_or_debind)
            
        else:
            raise RuntimeError(
                'Unexpected Golix primitive while listening for requests.'
            )
            
    async def _handle_request(self, notification):
        ''' The notification is an asymmetric request. Deal with it.
        '''
        try:
            packed = await self._librarian.retrieve(notification)
            unpacked = await self._golcore.unpack_request(packed)
        
            # TODO: have this filter based on contacts.
            if not (await self._librarian.contains(unpacked.author)):
                await self._salmonator.attempt_pull(
                    unpacked.author,
                    quiet = True
                )
        
            payload = await self._golcore.open_request(unpacked)
            await self._dispatch_payload(payload, notification)

        # Don't forget to (always) debind.
        finally:
            debinding = await self._golcore.make_debinding(notification)
            await self._percore.direct_ingest(
                obj = _GdxxLite.from_golix(debinding),
                packed = debinding.packed,
                remotable = True
            )
        
    async def _handle_debinding(self, debinding):
        ''' The notification is a debinding. Deal with it.
        '''
        # For now we just need to remove any pending requests for the
        # debinding's target.
        try:
            del self._pending_requests[debinding.target]
            await self._account.flush()
        except KeyError:
            logger.debug(
                'Request at ' + str(debinding.ghid) + ' target missing ' +
                'from pending requests: ' + str(debinding.target)
            )
        
    async def _dispatch_payload(self, payload, source_ghid):
        ''' Appropriately handles a request payload.
        '''
        if isinstance(payload, AsymHandshake):
            logger.debug(
                'Share handshake received from ' + str(payload.author) +
                ' at ' + str(source_ghid) + ' for ' + str(payload.target)
            )
            await self._handle_handshake(payload, source_ghid)
            
        elif isinstance(payload, AsymAck):
            logger.debug(
                'Handshake ACK received from ' + str(payload.author) +
                ' at ' + str(source_ghid) + ' for ' + str(payload.target)
            )
            await self._handle_ack(payload)
            
        elif isinstance(payload, AsymNak):
            logger.debug(
                'Handshake NAK received from ' + str(payload.author) +
                ' at ' + str(source_ghid) + ' for ' + str(payload.target)
            )
            await self._handle_nak(payload)
            
        else:
            raise RuntimeError('Encountered an unknown request type.')
            
    async def _handle_handshake(self, request, source_ghid):
        ''' Handles a handshake request after reception.
        '''
        try:
            # First, we need to figure out what the actual container object's
            # address is, and then stage the secret for it.
            # ALWAYS attempt to pull it first, or we may have a stale frame
            # locally, and attempt to stage a new secret for a long-expired
            # frame
            await self._salmonator.attempt_pull(
                request.target,
                quiet = True
            )
            
            binding_or_obj = await self._librarian.summarize(request.target)
            if isinstance(binding_or_obj, _GobdLite):
                container_ghid = binding_or_obj.target
                self._privateer.quarantine(
                    ghid = container_ghid,
                    secret = request.secret
                )
                if not (await self._librarian.contains(container_ghid)):
                    await self._salmonator.attempt_pull(
                        container_ghid,
                        quiet = True
                    )
            elif isinstance(binding_or_obj, _GeocLite):
                self._privateer.quarantine(
                    ghid = binding_or_obj.ghid,
                    secret = request.secret
                )
            else:
                raise TypeError(
                    'Invalid handshake target: ' + str(type(binding_or_obj))
                )
            
            # Note that unless we raise a HandshakeError RIGHT NOW, we'll be
            # sending an ack to the handshake, just to indicate successful
            # receipt of the share. If the originating app wants to check for
            # availability, well, that's currently on them. In the future, add
            # handle for that in SHARE instead of HANDSHAKE?
            
        except asyncio.CancelledError:
            logger.debug('Handshake handler cancelled.')
            raise
            
        except Exception as exc:
            logger.error(
                'Exception encountered while handling a handshake. Returned a '
                'NAK.\n' + ''.join(traceback.format_exc())
            )
            # Erfolglos. Send a nak to whomever sent the handshake
            response_obj = self._golcore._identity.make_nak(
                target = source_ghid
            )
            
        else:
            # Success. Send an ack to whomever sent the handshake
            response_obj = self._golcore._identity.make_ack(
                target = source_ghid
            )
        
        response = await self._golcore.make_request(
            request.author,
            response_obj
        )
        await self._percore.direct_ingest(
            obj = _GarqLite.from_golix(response),
            packed = response.packed,
            remotable = True
        )
            
        await self.share_handler(request.target, request.author)
            
    async def _handle_ack(self, request):
        ''' Handles a handshake ack after reception.
        '''
        try:
            target, recipient = self._pending_requests.pop(request.target)
        except KeyError:
            logger.error(
                'Received an ACK for an unknown origin: ' +
                str(request.target)
            )
        else:
            await self.receipt_ack_handler(target, recipient)
            
    async def _handle_nak(self, request):
        ''' Handles a handshake nak after reception.
        '''
        try:
            target, recipient = self._pending_requests.pop(request.target)
        except KeyError:
            logger.error(
                'Received a NAK for an unknown origin: ' +
                str(request.target)
            )
        else:
            await self.receipt_nak_handler(target, recipient)
    
    @fixture_noop
    @public_api
    async def share_handler(self, target, sender):
        ''' Incoming share targets (well, their ghids anyways) are
        forwarded to the _ipccore.
        
        Only separate from _handle_handshake right now because in the
        future, object sharing will be at least partly handled within
        its own dedicated rolodex pipeline.
        '''
        # Make sure we actually have the target
        if not (await self._librarian.contains(target)):
            await self._salmonator.attempt_pull(target, quiet=True)
        
        # Distribute the share in the background
        make_background_future(
            self._dispatch.distribute_share(target, sender)
        )
    
    @fixture_noop
    @public_api
    async def receipt_ack_handler(self, target, recipient):
        ''' Receives a share ack from the rolodex and passes it on to
        the application that requested the share.
        '''
        sharepair = _SharePair(target, recipient)
        tokens = self._outstanding_shares.pop_any(sharepair)
        await self._account.flush()
        
        # Distribute the share success in the background
        make_background_future(
            self._dispatch.distribute_share_success(
                target,
                recipient,
                tokens
            )
        )
    
    @fixture_noop
    @public_api
    async def receipt_nak_handler(self, target, recipient):
        ''' Receives a share nak from the rolodex and passes it on to
        the application that requested the share.
        '''
        sharepair = _SharePair(target, recipient)
        tokens = self._outstanding_shares.pop_any(sharepair)
        await self._account.flush()
        
        # Distribute the share failure in the background
        make_background_future(
            self._dispatch.distribute_share_failure(
                target,
                recipient,
                tokens
            )
        ) class SetMap:
    ''' Combines a mapping with a set. Threadsafe.
    '''
    
    def __init__(self):
        ''' Create a lookup!
        
        Currently does not support pre-population during __init__().
        '''
        self._mapping = {}
        self._lock = threading.RLock()
    
    def __getitem__(self, key):
        ''' Pass-through to the core lookup. Will return a frozenset.
        Raises keyerror if missing.
        '''
        with self._lock:
            return frozenset(self._mapping[key])
        
    def get_any(self, key):
        ''' Pass-through to the core lookup. Will return a frozenset.
        Will never raise a keyerror; if key not in self, returns empty
        frozenset.
        '''
        with self._lock:
            try:
                return frozenset(self._mapping[key])
            except KeyError:
                return frozenset()
                
    def pop_any(self, key):
        ''' Unlike other methods, pop_any returns the actual set,
        instead of a frozenset copy.
        '''
        with self._lock:
            try:
                return self._mapping.pop(key)
            except KeyError:
                return set()
        
    def __contains__(self, key):
        ''' Check to see if the key exists.
        '''
        with self._lock:
            return key in self._mapping
        
    def contains_within(self, key, value):
        ''' Check to see if the key exists, AND the value exists at key.
        '''
        with self._lock:
            try:
                return value in self._mapping[key]
            except KeyError:
                return False
        
    def add(self, key, value):
        ''' Adds the value to the set at key. Creates a new set there if
        none already exists.
        '''
        with self._lock:
            try:
                self._mapping[key].add(value)
            except KeyError:
                self._mapping[key] = {value}
                
    def update(self, key, value):
        ''' Updates the key with the value. Value must support being
        passed to set.update(), and the set constructor.
        '''
        with self._lock:
            try:
                self._mapping[key].update(value)
            except KeyError:
                self._mapping[key] = set(value)
                
    def update_all(self, other):
        ''' Updates all keys in other to self.
        '''
        with self._lock:
            for key in other:
                try:
                    self._mapping[key].update(other[key])
                except KeyError:
                    self._mapping[key] = set(other[key])
            
    def _remove_if_empty(self, key):
        ''' Removes a key entirely if it no longer has any values. Will
        suppress KeyError if the key is not found.
        '''
        try:
            if len(self._mapping[key]) == 0:
                del self._mapping[key]
        except KeyError:
            pass
        
    def remove(self, key, value):
        ''' Removes the value from the set at key. Will raise KeyError
        if either the key is missing, or the value is not contained at
        the key.
        '''
        with self._lock:
            try:
                self._mapping[key].remove(value)
            finally:
                self._remove_if_empty(key)
        
    def discard(self, key, value):
        ''' Same as remove, but will never raise KeyError.
        '''
        with self._lock:
            try:
                self._mapping[key].discard(value)
            except KeyError:
                pass
            finally:
                self._remove_if_empty(key)
        
    def clear(self, key):
        ''' Clears the specified key. Raises KeyError if key is not
        found.
        '''
        with self._lock:
            del self._mapping[key]
            
    def clear_any(self, key):
        ''' Clears the specified key, if it exists. If not, suppresses
        KeyError.
        '''
        with self._lock:
            try:
                del self._mapping[key]
            except KeyError:
                pass
        
    def clear_all(self):
        ''' Clears the entire mapping.
        '''
        with self._lock:
            self._mapping.clear()
            
    def __len__(self):
        ''' Returns the length of the mapping only.
        '''
        return len(self._mapping)
            
    def __iter__(self):
        ''' Send this through to the dict, bypassing the usual route,
        because otherwise we'll have a deadlock.
        '''
        with self._lock:
            for key in self._mapping:
                yield key
            
    def __eq__(self, other):
        ''' Expand comparison to search insides.
        '''
        with self._lock, other._lock:
            # First make sure both have same mapping keys.
            if set(self._mapping) != set(other._mapping):
                return False
            
            # Okay, now check each key has identical sets
            else:
                for this_key, this_set in self._mapping.items():
                    if other._mapping[this_key] != this_set:
                        return False
                        
        # If we successfully got through all of that, they are identical.
        return True
        
    def combine(self, other):
        ''' Returns a new SetMap with the union of both.
        '''
        new = type(self)()
        # Note that the iterator will take the lock, so we don't need to.
        for key in self:
            new.update(key, self._mapping[key])
        for key in other:
            new.update(key, other._mapping[key])
        return new

    def __getstate__(self):
        ''' Ignore self._lock to support pickling. Basically boilerplate
        copy from the pickle reference docs.
        '''
        # Copy the object's state from self.__dict__ which contains
        # all our instance attributes. Always use the dict.copy()
        # method to avoid modifying the original state.
        state = self.__dict__.copy()
        # Remove the unpicklable entries.
        del state['_lock']
        return state

    def __setstate__(self, state):
        ''' Ignore self._lock to support pickling. Basically boilerplate
        copy from the pickle reference docs.
        '''
        # Restore instance attributes (i.e., filename and lineno).
        self.__dict__.update(state)
        # Restore the lock.
        self._lock = threading.RLock()
        
    def __bool__(self):
        # Pass bool straight to mapping.
        return bool(self._mapping) class AppToken(bytes):
    ''' Extend bytes to be a fixed-length thing.
    '''
    # The size, in bytes, of an app token
    TOKEN_LEN = 16
    
    def __init__(self, *args, **kwargs):
        ''' For simplicity, enforce length AFTER performing init.
        '''
        if len(self) != self.TOKEN_LEN:
            raise ValueError('Improper token length.')
    
    @classmethod
    def null(cls):
        ''' Return the null token (used for passing "no token" in IPC,
        and not a valid token for app usage.)
        '''
        return cls(cls.TOKEN_LEN)
        
    @classmethod
    def pseudorandom(cls):
        ''' Return a PSEUDOrandom (**NOT** secure) app token, for use in
        testing.
        '''
        return cls([random.randint(0, 255) for __ in range(cls.TOKEN_LEN)])
        
    def __repr__(self):
        ''' Add in the class name to repr.
        '''
        return type(self).__name__ + '(' + super().__repr__() + ')'
        
    def __str__(self, *args, **kwargs):
        ''' Add in the class name to repr.
        '''
        return type(self).__name__ + '(' + \
            super().__str__(*args, **kwargs) + ')'
import logging
import threading
import weakref
import os
import random
import loopa
from Crypto.Protocol.KDF import scrypt
from golix import Ghid
from golix import Secret
from golix import FirstParty
from .core import GolixCore
from .core import Oracle
from .core import GhidProxier
from .gao import GAO
from .gao import GAOSet
from .gao import GAODict
from .gao import GAOSetMap
from .persistence import PersistenceCore
from .persistence import Doorman
from .persistence import Enforcer
from .persistence import Bookie
from .lawyer import LawyerCore
from .librarian import DiskLibrarian
from .undertaker import UndertakerCore
from .postal import MrPostman
from .dispatch import Dispatcher
from .dispatch import _Dispatchable
from .remotes import Salmonator
from .ipc import IPCServerProtocol as IPCCore
from .privateer import Privateer
from .rolodex import Rolodex
from .utils import SetMap
from .utils import AppToken
'''
LICENSING
-------------------------------------------------

hypergolix: A python Golix client.
    Copyright (C) 2016 Muterra, Inc.
    
    Contributors
    ------------
    Nick Badger
        badg@muterra.io | badg@nickbadger.com | nickbadger.com

    This library is free software; you can redistribute it and/or
    modify it under the terms of the GNU Lesser General Public
    License as published by the Free Software Foundation; either
    version 2.1 of the License, or (at your option) any later version.

    This library is distributed in the hope that it will be useful,
    but WITHOUT ANY WARRANTY; without even the implied warranty of
    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
    Lesser General Public License for more details.

    You should have received a copy of the GNU Lesser General Public
    License along with this library; if not, write to the
    Free Software Foundation, Inc.,
    51 Franklin Street,
    Fifth Floor,
    Boston, MA  02110-1301 USA

------------------------------------------------------
'''

# Global dependencies



# Intra-package dependencies








# ###############################################
# Boilerplate
# ###############################################


# Control * imports. Therefore controls what is available to toplevel
# package through __init__.py
__all__ = [
    'AgentBootstrap',
]


logger = logging.getLogger(__name__)


# ###############################################
# Utilities, etc
# ###############################################


# Use 2**14 for t<=100ms, 2**20 for t<=5s.
_DEFAULT_SCRYPT_HARDNESS = 2**15
# 256-bit password validator.
_PASSWORD_VALIDATOR_LEN = 32


class AgentBootstrap:
    ''' Agent bootstraps create and assemble the individual components
    needed to run the hypergolix service from a username and password.
    
    Also binds everything within a single namespace, etc etc.
    '''
    
    def __init__(self, cache_dir, aengel=None, debug=False):
        ''' Creates everything and puts it into a singular namespace.
        
        If bootstrap (ghid) is passed, we'll use the credential to
        extract an identity. If bootstrap_ghid is not passed, will use
        the credential to create one.
        
        TODO: move entire bootstrap creation process (or as much as
        possible, anyways) into register().
        '''
        # Before anything, create a global task management system
        self.taskmanager = loopa.TaskCommander(
            # threaded = False,
            # debug = False,
            # aengel = None,
            # reusable_loop = False,
            # start_timeout = None,
            name = 'taskman'
        )
        
        # First we need to create everything.
        self.percore = PersistenceCore()
        self.doorman = Doorman()
        self.enforcer = Enforcer()
        self.lawyer = LawyerCore()
        self.bookie = Bookie()
        self.librarian = DiskLibrarian(cache_dir=cache_dir)
        self.postman = MrPostman()
        self.undertaker = UndertakerCore()
        self.golcore = GolixCore()
        self.privateer = Privateer()
        self.oracle = Oracle()
        self.rolodex = Rolodex()

        self.ghidproxy = GhidProxier()