fn replace_with_true(&mut self) -> bool {

        let was = *self;

        *self = true;

        !was

    }

}

impl CuUndecided for ConnectorUnphased {

    fn logger(&mut self) -> &mut dyn Logger {

        &mut *self.inner.logger

    }

    fn proto_description(&self) -> &ProtocolDescription {

        &self.proto_description

        let Self { unphased: cu, phased } = self;

        match phased {

            ConnectorPhased::Setup { .. } => Err(SyncError::NotConnected),

            ConnectorPhased::Communication(comm) => {

                match &comm.round_result {

                    Err(SyncError::Unrecoverable(e)) => {

                        return Err(SyncError::Unrecoverable(e.clone()));

                    }

                    _ => {}

                }

                comm.round_result = Self::connected_sync(cu, comm, timeout);

                comm.round_index += 1;

        timeout: Option<Duration>,

    ) -> Result<Option<RoundOk>, SyncError> {

        //////////////////////////////////

        use SyncError as Se;

        //////////////////////////////////

        log!(

            "~~~ SYNC called with timeout {:?}; starting round {}",

            &timeout,

            comm.round_index

        );

        // 1. run all proto components to Nonsync blockers

        // iterate

        let mut branching_proto_components =

            HashMap::<ComponentId, BranchingProtoComponent>::default();

        let mut unrun_components: Vec<(ComponentId, ComponentState)> = cu

            .proto_components

            .iter()

            .map(|(&proto_id, proto)| (proto_id, proto.clone()))

            .collect();

        // drains unrun_components, and populates branching_proto_components.

        while let Some((proto_component_id, mut component)) = unrun_components.pop() {

            log!(

                "Nonsync running proto component with ID {:?}. {} to go after this",

                proto_component_id,

                unrun_components.len()

            );

            let mut ctx = NonsyncProtoContext {

                proto_component_id,

                unrun_components: &mut unrun_components,

            };

            let blocker = component.nonsync_run(&mut ctx, &cu.proto_description);

            log!(

                "proto component {:?} ran to nonsync blocker {:?}",

                proto_component_id,

                &blocker

            );

            use NonsyncBlocker as B;

            match blocker {

                    branching_proto_components

                        .insert(proto_component_id, BranchingProtoComponent::initial(component));

                }

            }

        }

        log!(

            "All {} proto components are now done with Nonsync phase",

            branching_proto_components.len(),

        );

        // Create temp structures needed for the synchronous phase of the round

        let mut rctx = RoundCtx {

            current_state,

            solution_storage: {

                let n = std::iter::once(SubtreeId::LocalComponent(cu.inner.native_component_id));

                SolutionStorage::new(subtree_id_iter)

            },

            spec_var_stream: cu.inner.current_state.id_manager.new_spec_var_stream(),

            payload_inbox: Default::default(),

            deadline: timeout.map(|to| Instant::now() + to),

        };

        // Explore all native branches eagerly. Find solutions, buffer messages, etc.

        log!(

            "Translating {} native batches into branches...",

            comm.native_batches.len()

        );

        let native_spec_var = rctx.spec_var_stream.next();

        let mut branching_native = BranchingNative { branches: Default::default() };

        'native_branches: for ((native_branch, index), branch_spec_val) in

            comm.native_batches.drain(..).zip(0..).zip(SpecVal::iter_domain())

        {

            let NativeBatch { to_get, to_put } = native_branch;

            let predicate = {

                let mut predicate = Predicate::default();

                // all firing ports have SpecVal::FIRING

                let firing_iter = to_get.iter().chain(to_put.keys()).copied();

                log!(

                    "New native with firing ports {:?}",

                    firing_iter.clone().collect::<Vec<_>>()

                );

                let firing_ports: HashSet<PortId> = firing_iter.clone().collect();

                for port in firing_iter {

                    let var = cu.inner.current_state.spec_var_for(port);

                    if firing_ports.contains(port) {

                        // this one is FIRING

                        continue;

                    }

                    let var = cu.inner.current_state.spec_var_for(*port);

                    if let Some(SpecVal::FIRING) = predicate.assigned.insert(var, SpecVal::SILENT) {

                        continue 'native_branches;

                    }

                }

                // this branch is consistent. distinguish it with a unique var:val mapping and proceed

                predicate.inserted(native_spec_var, branch_spec_val)

            };

            // send all outgoing messages (by buffering them)

            for (putter, payload) in to_put {

                let msg = SendPayloadMsg { predicate: predicate.clone(), payload };

                log!(

                    "Native branch {} sending msg {:?} with putter {:?}",

                    index,

                    &msg,

                    putter

                );

                // sanity check

                assert_eq!(Putter, cu.inner.current_state.port_info.get(&putter).unwrap().polarity);

                rctx.putter_push(cu, putter, msg);

            }

            let branch = NativeBranch { index, gotten: Default::default(), to_get };

            if branch.is_ended() {

                log!(

                    "Native submitting solution for batch {} with {:?}",

                    index,

                    &predicate

                );

                rctx.solution_storage.submit_and_digest_subtree_solution(

                    cu,

                unreachable!()

            }

        }

        // restore the invariant: !native_batches.is_empty()

        comm.native_batches.push(Default::default());

        // Call to another big method; keep running this round until a distributed decision is reached

        let decision = Self::sync_reach_decision(

            cu,

            comm,

            &mut branching_native,

            &mut branching_proto_components,

            &mut rctx,

        )?;

        // propagate the decision to children

        let msg = Msg::CommMsg(CommMsg {

            round_index: comm.round_index,

            contents: CommMsgContents::CommCtrl(CommCtrlMsg::Announce {

                decision: decision.clone(),

            }),

        });

        log!(

            "Announcing decision {:?} through child endpoints {:?}",

            &msg,

            &comm.neighborhood.children

        );

        for &child in comm.neighborhood.children.iter() {

            comm.endpoint_manager.send_to_comms(child, &msg)?;

        }

        let ret = match decision {

            Decision::Failure => {

                // dropping {branching_proto_components, branching_native}

                log!(

                    cu.inner.logger,

                    "End round with (updated) component states {:?}",

                    cu.proto_components.keys()

                );

                // consume native

            }

        };

        ret

    }

    fn sync_reach_decision(

        cu: &mut impl CuUndecided,

        comm: &mut ConnectorCommunication,

}

impl NonsyncProtoContext<'_> {

    pub fn new_component(&mut self, moved_ports: HashSet<PortId>, state: ComponentState) {

        // called by a PROTO COMPONENT. moves its own ports.

        // 1. sanity check: this component owns these ports

        log!(

            "Component {:?} added new component with state {:?}, moving ports {:?}",

            self.proto_component_id,

            &state,

            &moved_ports

        );

        println!("MOVED PORTS {:#?}", &moved_ports);

        // sanity check

        for port in moved_ports.iter() {

            assert_eq!(

                self.proto_component_id,

            );

        }

        // 2. create new component

        self.unrun_components.push((new_cid, state));

        // 3. update ownership of moved ports

        for port in moved_ports.iter() {

        }

        // 3. create a new component

    }

    pub fn new_port_pair(&mut self) -> [PortId; 2] {

        // adds two new associated ports, related to each other, and exposed to the proto component

        let [o, i] = [new_cid_fn(), new_cid_fn()];

            o,

            PortInfo {

                route: Route::LocalComponent,

                peer: Some(i),

                polarity: Putter,

                owner: self.proto_component_id,

            },

        );

            i,

            PortInfo {

                route: Route::LocalComponent,

                peer: Some(o),

                polarity: Getter,

                owner: self.proto_component_id,

            },

        );

        log!(

            "Component {:?} port pair (out->in) {:?} -> {:?}",

            self.proto_component_id,

            o,

            i

        );

        [o, i]

#[derive(Debug, Clone)]

struct CurrentState {

    port_info: HashMap<PortId, PortInfo>,

    id_manager: IdManager,

}

pub(crate) struct NonsyncProtoContext<'a> {

    unrun_components: &'a mut Vec<(ComponentId, ComponentState)>, // lives for Nonsync phase

}

pub(crate) struct SyncProtoContext<'a> {

    rctx: &'a RoundCtx,

    branch_inner: &'a mut ProtoComponentBranchInner, // sub-structure of component branch

    predicate: &'a Predicate,                        // KEY in pred->branch map

}

#[derive(Default, Debug, Clone)]

struct ProtoComponentBranchInner {

    untaken_choice: Option<u16>,