Hub

@classmethod
async def open(
    cls,
    store: KnowledgeStore,
    *,
    auth: AuthRegistry | None = None,
    clock: Callable[[], str] | None = None,
    ttl_sweep_interval: float = 30.0,
    expectation_sweep_interval: float = 10.0,
    invite_ack_timeout: float = 30.0,
    register_default_adapters: bool = True,
) -> "Hub":
    """Construct + hydrate from disk + start sweepers. Production entry point.

    ``register_default_adapters=True`` (default) registers the
    built-in adapters (``consulting@v1``, ``conversation@v1``,
    ``discussion@v1``) and the built-in expectation evaluators /
    violation handlers (``acks_within`` / ``reply_within`` /
    ``max_silence``, ``audit`` / ``notify_channel`` /
    ``auto_close``) so simple test setups don't need explicit
    registration calls.

    Set ``expectation_sweep_interval=0`` to disable the expectation
    sweeper entirely (tests usually do this to avoid background
    timer noise).
    """
    hub = cls(
        store,
        auth=auth,
        clock=clock,
        ttl_sweep_interval=ttl_sweep_interval,
        expectation_sweep_interval=expectation_sweep_interval,
        invite_ack_timeout=invite_ack_timeout,
    )
    if register_default_adapters:
        hub.register_adapter(ConsultingAdapter())
        hub.register_adapter(ConversationAdapter())
        hub.register_adapter(DiscussionAdapter())
        hub.register_adapter(WorkflowAdapter())
        for evaluator in default_evaluators():
            hub.register_expectation_evaluator(evaluator)
        for handler in default_handlers():
            hub.register_violation_handler(handler)
    await hub.hydrate()
    await hub.start()
    return hub

async def hydrate(self) -> None:
    """Walk the store; rebuild caches. Idempotent.

    Loads identities, channels, and tasks from disk. Active channel
    WALs are re-folded through their adapter so the
    ``_adapter_states`` cache is rebuilt deterministically.
    """
    self._passports.clear()
    self._resumes.clear()
    self._rules.clear()
    self._skills.clear()
    self._name_to_id.clear()
    self._capability_index.clear()
    self._channels.clear()
    self._active_channels.clear()
    self._adapter_states.clear()
    self._tasks.clear()
    self._channel_tasks.clear()
    self._inbox_cursors.clear()
    self._causation_index.clear()

    # Identities.
    agent_children = await self._store.list(agents_root())
    for child in agent_children:
        if not child.endswith("/"):
            continue
        agent_id = child.rstrip("/")
        await self._load_agent(agent_id)

    # Rebuild capability index from loaded resumes — by_capability.json
    # is a derived cache, the resumes are the authoritative source.
    for agent_id, resume in self._resumes.items():
        for cap in resume.claimed_capabilities:
            self._capability_index.setdefault(cap, set()).add(agent_id)
        for cap in resume.observed:
            self._capability_index.setdefault(cap, set()).add(agent_id)

    # Channels — load metadata first, then re-fold WALs.
    channel_children = await self._store.list(channels_root())
    for child in channel_children:
        if not child.endswith("/"):
            continue
        channel_id = child.rstrip("/")
        await self._load_channel(channel_id)

    # Tasks.
    task_children = await self._store.list(tasks_root())
    for child in task_children:
        if not child.endswith("/"):
            continue
        task_id = child.rstrip("/")
        await self._load_task(task_id)

async def start(self) -> None:
    """Spawn the TTL + expectation + custom sweepers. Idempotent.

    ``ttl_sweep_interval=0`` disables the TTL sweeper;
    ``expectation_sweep_interval=0`` disables the expectation
    sweeper. Custom sweepers attached via
    :meth:`register_sweeper` start here too (registered before
    ``start()``) or immediately at registration (after ``start()``).
    """
    if self._ttl_sweep_interval > 0 and self._ttl_sweeper is None:
        self._ttl_sweeper = _IntervalSweeper(
            name="ttl",
            interval=self._ttl_sweep_interval,
            fn=self.expire_due,
        )
        self._ttl_sweeper.start()
    if self._expectation_sweep_interval > 0 and self._expectation_sweeper is None:
        self._expectation_sweeper = _IntervalSweeper(
            name="expectations",
            interval=self._expectation_sweep_interval,
            fn=self._expectation_tick,
        )
        self._expectation_sweeper.start()
    for sweeper in self._custom_sweepers.values():
        sweeper.start()
    self._started = True

async def close(self) -> None:
    """Cancel sweepers + endpoint tasks; drain queues. Idempotent."""
    if self._closed:
        return
    self._closed = True
    if self._ttl_sweeper is not None:
        await self._ttl_sweeper.stop()
        self._ttl_sweeper = None
    if self._expectation_sweeper is not None:
        await self._expectation_sweeper.stop()
        self._expectation_sweeper = None
    for sweeper in list(self._custom_sweepers.values()):
        await sweeper.stop()
    self._custom_sweepers.clear()
    for task in list(self._endpoint_tasks):
        task.cancel()
    if self._endpoint_tasks:
        await asyncio.gather(*self._endpoint_tasks, return_exceptions=True)
    self._endpoint_tasks.clear()

def register_sweeper(
    self,
    name: str,
    interval_seconds: float,
    fn: Callable[[], "Awaitable[None]"],
) -> None:
    """Attach a custom periodic worker.

    ``fn`` is called every ``interval_seconds``. Subclasses use this
    for protocol-specific background work (e.g. polling a chat
    platform's presence list, refreshing an auth token).

    If ``Hub.start()`` has already run, the sweeper starts immediately.
    Otherwise it's queued and starts when ``start()`` runs.

    Re-registering at the same ``name`` raises ``ValueError`` — use
    :meth:`unregister_sweeper` first if you mean to replace.

    Sync vs. async: ``register_sweeper`` is synchronous because it
    only updates internal bookkeeping (and may call the underlying
    ``_IntervalSweeper.start`` which schedules a fire-and-forget
    task). :meth:`unregister_sweeper` is async because it awaits the
    sweeper's task cancellation to ensure clean shutdown.
    """
    if name in self._custom_sweepers:
        raise ValueError(f"sweeper already registered: {name!r}")
    if interval_seconds <= 0:
        raise ValueError(f"interval_seconds must be positive: {interval_seconds}")
    sweeper = _IntervalSweeper(name=name, interval=interval_seconds, fn=fn)
    self._custom_sweepers[name] = sweeper
    if self._started:
        # ``start()`` has already run — start this sweeper immediately.
        sweeper.start()

async def unregister_sweeper(self, name: str) -> None:
    """Stop and remove a custom sweeper. No-op if absent.

    Async to mirror :meth:`_IntervalSweeper.stop`, which awaits
    cancellation of the running task. Custom sweepers registered
    before :meth:`start` still go through this path on
    :meth:`close`, so subclasses don't need to track them
    themselves.
    """
    sweeper = self._custom_sweepers.pop(name, None)
    if sweeper is not None:
        await sweeper.stop()

def register_adapter(self, adapter: ChannelAdapter) -> None:
    """Register a ``ChannelAdapter`` keyed by ``(type, version)``.

    Re-registering at the same key replaces the prior adapter; the
    old key's existing in-flight channels keep their snapshotted
    manifest for life.
    """
    key = (adapter.manifest.type, adapter.manifest.version)
    self._adapters[key] = adapter

def register_listener(self, listener: HubListener) -> None:
    """Attach a :class:`HubListener` to receive state-transition events.

    Listeners receive events in registration order. Each is wrapped
    in try/except so one buggy listener cannot stall dispatch — its
    exception is logged at ``ERROR`` and the next listener still
    runs.
    """
    self._listeners.append(listener)

def unregister_listener(self, listener: HubListener) -> None:
    """Detach a previously-registered listener. No-op if absent."""
    with contextlib.suppress(ValueError):
        self._listeners.remove(listener)

async def report_turn_failure(
    self,
    *,
    channel_id: str,
    agent_id: str,
    envelope_id: str,
    exc: BaseException,
) -> None:
    """Fan out an ``on_turn_failed`` event to every listener.

    Public entry point so client-side notify handlers can route
    substantive-turn crashes through the observability surface
    without touching hub privates. ``AuditLog`` (the built-in
    listener) records the failure; tenant listeners react however
    they choose.
    """
    await self._fan_out(
        "on_turn_failed",
        channel_id,
        agent_id,
        envelope_id,
        exc,
    )

async def fire_task_event(
    self,
    task_id: str,
    kind: str,
    payload: dict,
) -> None:
    """Fan out an ``on_task_event`` to every listener.

    Public entry point so :class:`TaskMirror` (and other tenant
    observers) can route task-side notifications through the hub's
    listener surface without touching ``_fan_out``. ``kind`` is
    free-form — the built-in listener Protocol documents
    ``"started"`` / ``"progress"`` / ``"completed"`` / ``"failed"`` /
    ``"expired"`` / ``"cancelled"`` / ``"mirror_failed"`` as the
    recognised values, but tenants may emit additional kinds.
    """
    await self._fan_out("on_task_event", task_id, kind, payload)

def replace_audit_log(self, audit_log: AuditLog) -> None:
    """Swap the built-in :class:`AuditLog` for a tenant-provided one.

    Unregisters the prior audit log from the listener chain,
    registers the replacement as the first listener (preserving
    the convention that audit writes complete before tenant
    listeners observe the same event), and updates
    :attr:`audit_log` to point at it.
    """
    with contextlib.suppress(ValueError):
        self._listeners.remove(self._audit_log)
    self._audit_log = audit_log
    self._listeners.insert(0, audit_log)

def register_arbiter(self, arbiter: HubArbiter) -> None:
    """Replace the active :class:`HubArbiter` instance.

    The default :class:`RuleBasedArbiter` is installed automatically
    and enforces per-agent :class:`Rule` (access + limits) — the
    same behavior the hub had inline before this seam existed.
    Tenants replace it to layer custom permission protocols
    (JWT scope, federation routing, etc.) on top of (or in place
    of) the rule-based defaults.

    Only one arbiter is active at a time; calling this with a new
    instance replaces the prior arbiter outright.
    """
    self._arbiter = arbiter

def register_remote_proxy(self, proxy: RemoteAgentProxy) -> None:
    """Register a federation proxy keyed by ``proxy.scheme``.

    When the hub dispatches an envelope to a recipient whose
    passport has ``effective_kind == "remote_agent"``, it looks up
    the proxy by the recipient's ``auth.scheme`` and calls
    ``proxy.dispatch(envelope, recipient)`` instead of sending a
    ``NotifyFrame`` to a local endpoint. Re-registering at the
    same ``scheme`` replaces the prior proxy.
    """
    self._remote_proxies[proxy.scheme] = proxy

def unregister_remote_proxy(self, scheme: str) -> RemoteAgentProxy | None:
    """Remove the proxy registered for ``scheme`` and return it.

    Returns ``None`` if no proxy was registered for ``scheme``.
    The caller is responsible for awaiting ``proxy.close()``
    — the hub leaves lifecycle decisions to whoever owns the
    proxy instance.
    """
    return self._remote_proxies.pop(scheme, None)

def remote_proxy_for(self, scheme: str) -> RemoteAgentProxy | None:
    """Read-only lookup against the proxy registry."""
    return self._remote_proxies.get(scheme)

def health(self) -> dict:
    """Return an operational snapshot of hub state.

    Cheap to compute (in-memory only). Wire to a ``/health``
    endpoint or operational dashboard. The shape is intentionally
    small — operators want a handful of indicative numbers, not
    the full state.

    Fields:

    * ``active_channels``: number of channels in OPENED/PENDING state.
    * ``registered_agents``: total registered identities (agents +
      humans).
    * ``pending_inbox_total``: sum of per-recipient outstanding
      envelope counters (best-effort approximation).
    * ``max_pending_inbox_depth``: maximum per-recipient queue depth,
      or ``None`` when nothing is queued. Indicative of the
      "stuck agent" case.
    * ``registered_listeners``: number of attached
      :class:`HubListener` instances (the built-in
      :class:`AuditLog` counts).
    * ``adapters_loaded``: number of registered
      :class:`ChannelAdapter` instances.
    """
    max_depth: int | None = None
    if self._inbox_pending:
        max_depth = max(self._inbox_pending.values())
    return {
        "active_channels": len(self._active_channels),
        "registered_agents": len(self._passports),
        "pending_inbox_total": sum(self._inbox_pending.values()),
        "max_pending_inbox_depth": max_depth,
        "registered_listeners": len(self._listeners),
        "adapters_loaded": len(self._adapters),
        "audit_log_bytes": self._audit_log.bytes_written,
    }

async def on_envelope_posted(self, envelope: Envelope, metadata: ChannelMetadata) -> None:  # noqa: ARG002
    return None

async def on_envelope_rejected(self, envelope: Envelope, reason: NetworkError) -> None:  # noqa: ARG002
    return None

async def on_dispatch_failed(
    self,
    envelope: Envelope,
    recipient_id: str,
    reason: BaseException,
) -> None:  # noqa: ARG002
    return None

async def on_channel_event(self, channel_id: str, kind: str, payload: dict) -> None:  # noqa: ARG002
    return None

async def on_agent_event(self, agent_id: str, kind: str, payload: dict) -> None:  # noqa: ARG002
    return None

async def on_expectation_fired(self, channel_id: str, expectation: object, violation: object) -> None:  # noqa: ARG002
    return None

async def on_turn_failed(
    self,
    channel_id: str,
    agent_id: str,
    envelope_id: str,
    exc: BaseException,
) -> None:  # noqa: ARG002
    return None

async def on_task_event(self, task_id: str, kind: str, payload: dict) -> None:  # noqa: ARG002
    return None

async def on_inbox_pressure(self, agent_id: str, pending: int, cap: int) -> None:  # noqa: ARG002
    return None

def register_expectation_evaluator(self, evaluator: ExpectationEvaluator) -> None:
    """Register an evaluator keyed by ``evaluator.name``.

    Re-registering the same name replaces the prior evaluator.
    """
    self._expectation_evaluators[evaluator.name] = evaluator

def register_violation_handler(self, handler: ViolationHandler) -> None:
    """Register a violation handler keyed by ``handler.name``.

    Re-registering the same name replaces the prior handler.
    """
    self._violation_handlers[handler.name] = handler

async def register(
    self,
    passport: Passport,
    resume: Resume,
    *,
    skill_md: str | None = None,
    rule: Rule | None = None,
) -> Passport:
    # Remote-agent passports represent participants on another hub
    # and ``auth.scheme`` is a routing label consumed by the
    # registered ``RemoteAgentProxy`` — not a credential to be
    # validated locally. Skip the local auth check for them;
    # authentication on the wire belongs to the originating hub.
    if passport.effective_kind != "remote_agent":
        adapter = self._auth.get(passport.auth.scheme)
        await adapter.validate(passport, passport.auth.claim)

    async with self._registration_lock:
        # Reject a re-register that collides on ``name``: the prior
        # registration's passport / resume / rule / SKILL.md would
        # be orphaned on disk under a now-unreachable agent_id.
        # Tenants must explicitly ``unregister`` first.
        if passport.name in self._name_to_id:
            raise ProtocolError(
                f"name {passport.name!r} already registered "
                f"(agent_id={self._name_to_id[passport.name]}); "
                "unregister it before re-registering."
            )
        agent_id = make_id()
        passport.agent_id = agent_id
        passport.created_at = self._clock()

        effective_rule = rule if rule is not None else Rule()

        await self._persist_passport(passport)
        await self._persist_resume(agent_id, resume)
        await self._persist_rule(agent_id, effective_rule)
        if skill_md is not None:
            await self._persist_skill(agent_id, skill_md)

        self._passports[agent_id] = passport
        self._resumes[agent_id] = resume
        self._rules[agent_id] = effective_rule
        if skill_md is not None:
            self._skills[agent_id] = skill_md
        self._name_to_id[passport.name] = agent_id

        for cap in resume.claimed_capabilities:
            self._capability_index.setdefault(cap, set()).add(agent_id)
        for cap in resume.observed:
            self._capability_index.setdefault(cap, set()).add(agent_id)

    await self._persist_capability_index()
    logger.info("agent registered: name=%s agent_id=%s", passport.name, agent_id)
    await self._fan_out(
        "on_agent_event",
        agent_id,
        "registered",
        {"passport": passport, "at": self._clock()},
    )
    return passport

async def unregister(self, agent_id: str) -> None:
    if agent_id not in self._passports:
        raise NotFoundError(f"agent not registered: {agent_id}")

    async with self._registration_lock:
        passport = self._passports.pop(agent_id, None)
        self._resumes.pop(agent_id, None)
        self._rules.pop(agent_id, None)
        self._skills.pop(agent_id, None)
        if passport is not None and self._name_to_id.get(passport.name) == agent_id:
            self._name_to_id.pop(passport.name, None)

        endpoint_id = self._agent_to_endpoint.pop(agent_id, None)
        if endpoint_id is not None:
            bound = self._endpoint_to_agents.get(endpoint_id)
            if bound is not None:
                bound.discard(agent_id)
                if not bound:
                    self._endpoint_to_agents.pop(endpoint_id, None)

        # Drop the agent from every capability bucket; clean empty
        # buckets so the index stays compact.
        empty_caps: list[str] = []
        for cap, ids in self._capability_index.items():
            ids.discard(agent_id)
            if not ids:
                empty_caps.append(cap)
        for cap in empty_caps:
            self._capability_index.pop(cap, None)

        # Delete on-disk identity files. Without this the next
        # ``hydrate()`` would re-load the unregistered agent from
        # disk. Channels and tasks the agent participated in are
        # kept for audit / read; only the per-agent identity files
        # are removed.
        await self._store.delete(passport_path(agent_id))
        await self._store.delete(resume_path(agent_id))
        await self._store.delete(rule_path(agent_id))
        await self._store.delete(skill_path(agent_id))

        # Drop inbox accounting so a future re-register with a
        # different agent_id starts from zero.
        self._inbox_pending.pop(agent_id, None)
        self._inbox_cursors.pop(agent_id, None)
        await self._store.delete(inbox_cursor_path(agent_id))

    await self._persist_capability_index()
    logger.info("agent unregistered: agent_id=%s", agent_id)
    await self._fan_out(
        "on_agent_event",
        agent_id,
        "unregistered",
        {"name": passport.name if passport is not None else None, "at": self._clock()},
    )

def name_for(self, agent_id: str, *, default: str | None = None) -> str:
    """Resolve ``agent_id`` to its registered ``Passport.name``.

    Reads the in-memory passport directory.
    Returns ``default`` when the id is unknown (or ``agent_id`` itself
    if ``default`` is ``None``), so callers can use this as a safe
    ``NameResolver`` for view projection without needing to handle
    the unregistered / unregistered-mid-turn case.
    """
    passport = self._passports.get(agent_id)
    if passport is not None:
        return passport.name
    return default if default is not None else agent_id

async def get_agent(self, name_or_id: str) -> Passport:
    agent_id = self._name_to_id.get(name_or_id, name_or_id)
    passport = self._passports.get(agent_id)
    if passport is None:
        raise NotFoundError(f"agent not found: {name_or_id}")
    return passport

async def get_resume(self, agent_id: str) -> Resume:
    resume = self._resumes.get(agent_id)
    if resume is None:
        raise NotFoundError(f"resume not found: {agent_id}")
    return resume

def find_agent_id(self, name: str) -> str | None:
    """Resolve ``name`` to its registered ``agent_id``, or ``None``.

    Non-raising peer to :meth:`get_agent` — callers that need to
    branch on "is this name registered?" without catching an
    exception use this directly. Returns ``None`` when ``name``
    has no current registration.
    """
    return self._name_to_id.get(name)

def name_to_id_map(self) -> dict[str, str]:
    """Snapshot of the ``name → agent_id`` directory.

    Public read surface so callers that need reverse name
    resolution (e.g. ``WorkflowAdapter`` resolving handoff target
    names to ids) don't reach into the private index. Returns a
    copy so mutation can't corrupt the registry.
    """
    return dict(self._name_to_id)

async def get_rule(self, agent_id: str) -> Rule:
    """Return the rule attached to ``agent_id``.

    Raises :class:`NotFoundError` if no rule is registered — the
    registration path stamps a default :class:`Rule` for every
    agent, so a missing entry indicates the agent itself is
    unregistered.
    """
    rule = self._rules.get(agent_id)
    if rule is None:
        raise NotFoundError(f"rule not found: {agent_id}")
    return rule

async def get_skill(self, agent_id: str) -> str | None:
    if agent_id in self._skills:
        return self._skills[agent_id]
    if agent_id not in self._passports:
        return None
    body = await self._store.read(skill_path(agent_id))
    if body is not None:
        self._skills[agent_id] = body
    return body

async def list_agents(
    self,
    *,
    capability: str | None = None,
    query: str | None = None,
    kind: str | None = None,
    sort_by: str | None = None,
    limit: int = 50,
) -> list[Passport]:
    """Enumerate registered participants, optionally filtered.

    ``kind`` filters by ``Passport.kind`` (``"agent"`` / ``"human"`` /
    ``"remote_agent"``). ``None`` returns all kinds (current default
    behavior); passing ``"agent"`` also matches passports with
    ``kind=None`` since ``None`` is the back-compat alias.
    """
    results: list[Passport] = []
    query_lower = query.lower() if query else None
    for agent_id, passport in self._passports.items():
        if kind is not None:
            resolved = passport.kind or "agent"
            if resolved != kind:
                continue
        if capability is not None:
            resume = self._resumes.get(agent_id)
            if resume is None:
                continue
            claimed = set(resume.claimed_capabilities)
            observed = set(resume.observed.keys())
            if capability not in claimed and capability not in observed:
                continue
        if query_lower is not None:
            resume = self._resumes.get(agent_id)
            summary = resume.summary.lower() if resume else ""
            if query_lower not in summary:
                continue
        results.append(passport)

    if sort_by == "name":
        results.sort(key=lambda p: p.name)

    return results[:limit]

async def set_resume(self, agent_id: str, resume: Resume) -> None:
    if agent_id not in self._passports:
        raise NotFoundError(f"agent not registered: {agent_id}")
    resume.last_updated = self._clock()
    resume.version = (self._resumes[agent_id].version + 1) if agent_id in self._resumes else resume.version

    # Diff capabilities so the index stays in sync. Without this,
    # a tenant adding a new claim via ``set_resume`` would not
    # surface under ``peers(action="find", capability=...)`` until
    # the agent re-registered or recorded an observation.
    old_resume = self._resumes.get(agent_id)
    old_caps: set[str] = set()
    if old_resume is not None:
        old_caps.update(old_resume.claimed_capabilities)
        old_caps.update(old_resume.observed.keys())
    new_caps: set[str] = set(resume.claimed_capabilities) | set(resume.observed.keys())

    await self._persist_resume(agent_id, resume)
    self._resumes[agent_id] = resume

    added = new_caps - old_caps
    removed = old_caps - new_caps
    for cap in added:
        self._capability_index.setdefault(cap, set()).add(agent_id)
    for cap in removed:
        bucket = self._capability_index.get(cap)
        if bucket is None:
            continue
        bucket.discard(agent_id)
        if not bucket:
            self._capability_index.pop(cap, None)
    if added or removed:
        await self._persist_capability_index()

    await self._fan_out(
        "on_agent_event",
        agent_id,
        "resume_set",
        {"resume": resume, "version": resume.version, "at": self._clock()},
    )

async def set_skill(self, agent_id: str, skill_md: str | None) -> None:
    if agent_id not in self._passports:
        raise NotFoundError(f"agent not registered: {agent_id}")
    if skill_md is None:
        await self._store.delete(skill_path(agent_id))
        self._skills.pop(agent_id, None)
    else:
        await self._persist_skill(agent_id, skill_md)
        self._skills[agent_id] = skill_md
    await self._fan_out(
        "on_agent_event",
        agent_id,
        "skill_set",
        {"removed": skill_md is None, "at": self._clock()},
    )

async def set_rule(self, agent_id: str, rule: Rule) -> None:
    if agent_id not in self._passports:
        raise NotFoundError(f"agent not registered: {agent_id}")
    rule.version = (self._rules[agent_id].version + 1) if agent_id in self._rules else rule.version
    await self._persist_rule(agent_id, rule)
    self._rules[agent_id] = rule
    await self._fan_out(
        "on_agent_event",
        agent_id,
        "rule_set",
        {"rule": rule, "version": rule.version, "at": self._clock()},
    )