council.py

"""The Council: concurrent multi-model fan-out plus synthesis.

``Council`` is the primary importable entry point. It resolves friendly names to
provider-prefixed model ids, skips any member whose API key is absent, fans the prompt out
concurrently, collects partial results even when some members fail, and (in
synthesize mode) asks a synthesizer model to merge the answers into one.

The deliberation modes (``debate``, ``adversarial``) live in :mod:`conclave.modes`
and reuse this class's :meth:`Council.fan_out` primitive so the partial-failure
handling is written exactly once.

Synthesizer selection and degradation (the "council" value prop)
----------------------------------------------------------------

**Which model synthesizes.** Synthesis is performed by one *synthesizer* model,
separate from the council members (though a member may also be the synthesizer).
Selection precedence, highest first:

1. the ``synthesizer=`` argument to :class:`Council` (the CLI ``--synthesizer/-s``
   flag wires straight through to this);
2. the ``synthesizer:`` key in ``~/.conclave/config.yml``;
3. the built-in default :data:`conclave.registry.DEFAULT_SYNTHESIZER` (``"claude"``,
   i.e. ``anthropic/claude-sonnet-4-6``).

The same model is the **judge** in ``adversarial`` mode and the final
consolidator in ``debate`` mode -- one selection drives all three.

**The fallback / degraded path is OBSERVABLE, never silent.** Synthesis can fail
to run for three reasons, and each one is signaled on the result rather than
silently swallowed:

* *No usable member answers* (every member errored/skipped) -- nothing to merge;
* *The synthesizer has no API key* in the environment;
* *The synthesizer call itself fails* (provider error/timeout).

In all three cases ``CouncilResult.synthesis`` stays ``None``, the member answers
are still returned intact, a warning is logged, and an actionable reason is set
on ``CouncilResult.synthesis_error`` (in ``adversarial`` mode the analogous
``AdversarialResult.verdict_error``, mirrored to ``synthesis_error``). A caller
can therefore always tell synthesis did **not** happen as expected by checking
``synthesis is None and synthesis_error is not None`` -- there is no path where
the council quietly returns concatenated/partial output dressed up as a synthesis.

**The synthesis prompt is a versioned constant.** The synthesize-mode system
prompt is :data:`_SYNTH_SYSTEM` (the debate/judge prompts live in
:mod:`conclave.prompts`); the prompt *set* carries the version tag
:data:`conclave.prompts.SYNTHESIS_PROMPT_VERSION`, stamped onto every
:class:`~conclave.models.CouncilResult` as ``prompt_version`` so a prompt change
is detectable downstream instead of being silently absorbed as model drift.
"""

from __future__ import annotations

import asyncio
from collections.abc import AsyncIterator, Awaitable, Callable
from uuid import uuid4

from . import cache as cache_mod
from . import transport
from .adapters.base import redact
from .config import ConclaveConfig, load_config
from .logging import get_logger
from .manifest import (
    ModelHarnessManifest,
    ProviderExecutionReceipt,
    ProviderSkip,
    verified_secret_safety,
)
from .models import ELITE_PROTOCOL_VERSION, CouncilResult, ModelAnswer, StreamEvent, TokenUsage
from .prompts import ELITE_PROMPT_VERSION, SYNTHESIS_PROMPT_VERSION
from .providers import call_model, receipt_from_answer
from .registry import key_present

logger = get_logger("council")

# A per-member message-list factory: given a (friendly_name, model_id) member,
# return the OpenAI-style messages to send it. Lets each mode tailor the prompt
# per member while sharing Council.fan_out's concurrency + partial-failure code.
MessagesFor = Callable[[str, str], list[dict[str, str]]]

# The synthesize-mode system prompt. It is a stable module constant -- never
# built per-call -- so the wording the council synthesizes under is auditable and
# diffable. Any change to it (or to the debate/judge prompts in
# :mod:`conclave.prompts`) MUST be paired with a bump of
# :data:`conclave.prompts.SYNTHESIS_PROMPT_VERSION`, which is stamped onto every
# :class:`~conclave.models.CouncilResult` as ``prompt_version`` so a downstream
# eval can detect the change rather than silently absorb it. ``test_synthesizer``
# pins both this text and the version, so editing one without the other fails CI.
_SYNTH_SYSTEM = (
    "You are the synthesizer of a council of AI models. You are given the same "
    "user prompt that was posed to several models, plus each model's answer. "
    "Produce one consolidated, accurate answer. Reconcile agreements, surface "
    "and adjudicate disagreements, and note any answer that is clearly wrong. "
    "Do not invent a model's position; rely only on the answers provided."
)

# Re-exported for callers that want the version without importing prompts.
__all__ = ["Council", "SYNTHESIS_PROMPT_VERSION"]


class Council:
    """A council of foundation models with an optional synthesizer.

    Args:
        models: Friendly names (or raw provider-prefixed model ids) of council members.
        synthesizer: Friendly name of the synthesizer model. If ``None``, the
            config default is used.
        config: Pre-loaded config; if ``None``, loaded from disk + defaults.
        temperature: Sampling temperature for member calls.
        timeout: Per-call timeout in seconds.
        cache: Opt-in result cache. ``None`` (default) defers to
            ``config.cache`` (off unless enabled in ``~/.conclave/config.yml``);
            ``True``/``False`` overrides it for this council. When enabled, an
            identical repeat run is served from the on-disk cache instead of
            re-calling the providers. The cache never persists API keys --
            see :mod:`conclave.cache`.
        extract_verdict: Whether to run the structured verdict-extraction step
            (CAC-05) after a synthesize-mode run. Defaults to ``True`` -- the
            auditable verdict (consensus score, conflicts, provider votes) is the
            council's product, so it is on by default. **Cost note:** verdict
            extraction makes a SECOND synthesizer call (one extra LLM round-trip
            per ``ask``, plus one more on the single repair retry) distinct from
            the prose ``synthesis`` call. Subsuming both into a single synthesizer
            call is a future optimization; for now this flag is the single opt-out.
            Set ``False`` to skip the verdict entirely (``CouncilResult.verdict``
            stays ``None`` and the manifest's verdict-provenance slots stay at
            their defaults). Verdict extraction never runs in ``raw`` mode
            (``synthesize=False``) regardless of this flag.
        allow_transport_debug_logging: Opt **out** of the transport-logging guard.
            Defaults to ``False``, which means the guard is **ON**: constructing a
            ``Council`` installs :func:`conclave.transport.guard_transport_logging`
            so httpx/httpcore ``DEBUG`` records -- the only band that emits request
            headers, including the live ``Authorization``/``x-api-key`` value -- are
            dropped before any handler formats them (key-leak audit, RANK 6). The
            guard is idempotent, so constructing many councils installs it once. The
            filter is scoped to the ``httpx``/``httpcore`` loggers only; it never
            touches the host application's root logger or any other logger.
            Set ``True`` to skip installation for the rare case where you genuinely
            need httpx/httpcore ``DEBUG`` output in a process that does not hold real
            keys; you remain responsible for that band then. Consumers using the
            provider functions directly (without a ``Council``) can still call
            :func:`conclave.guard_transport_logging` themselves.
        source_bundle_digest: Optional digest of a future source-grounding bundle.
            When supplied, it participates in cache identity so grounded and
            ungrounded Elite runs cannot collide. The value is re-hashed before
            entering the canonical identity document.

    Example:
        >>> council = Council(models=["grok", "perplexity"], synthesizer="claude")
        >>> result = council.ask_sync("What is the capital of France?")
        >>> print(result.synthesis)
    """

    def __init__(
        self,
        models: list[str],
        synthesizer: str | None = None,
        config: ConclaveConfig | None = None,
        temperature: float = 0.7,
        timeout: float = 120.0,
        cache: bool | None = None,
        extract_verdict: bool = True,
        allow_transport_debug_logging: bool = False,
        source_bundle_digest: str | None = None,
    ) -> None:
        self.config = config or load_config()
        self.requested_models = list(models)
        self.synthesizer = synthesizer or self.config.synthesizer
        self.temperature = temperature
        self.timeout = timeout
        # Explicit override wins; otherwise defer to config (off by default).
        self.cache_enabled = self.config.cache if cache is None else cache
        # Default-on verdict extraction (CAC-06). Named ``*_enabled`` to read
        # unambiguously as a switch, never confused with the imported
        # ``extract_verdict`` engine function. There is no per-call override --
        # this constructor flag is the single resolution path (one opt-out).
        self.extract_verdict_enabled = extract_verdict
        # Horizon-2 placeholder: callers with a grounded source bundle can bind
        # its digest into cache identity now, before source retrieval ships.
        # cache.build_identity re-hashes this value so malformed/raw input never
        # appears in inspectable identity documents or diagnostics.
        self.source_bundle_digest = source_bundle_digest
        # Default-on transport-logging guard (key-leak audit, RANK 6): drop
        # httpx/httpcore DEBUG records (the only band that emits the auth header)
        # so a process holding a real key cannot leak it via verbose transport
        # logging, even if the host enables DEBUG app-wide. Idempotent, so many
        # councils install it once; scoped to the httpx/httpcore loggers only.
        # ``allow_transport_debug_logging=True`` opts out for callers who need
        # that DEBUG band and accept the responsibility.
        if not allow_transport_debug_logging:
            transport.guard_transport_logging()

    def _available_members(self) -> tuple[list[tuple[str, str]], list[str]]:
        """Partition requested members into (available, skipped-for-no-key).

        Returns:
            A tuple ``(members, skipped)`` where ``members`` is a list of
            ``(friendly_name, model_id)`` pairs that have a key present, and
            ``skipped`` is the list of friendly names with no key available.
        """
        members: list[tuple[str, str]] = []
        skipped: list[str] = []
        for name in self.requested_models:
            model_id = self.config.resolve_model_id(name)
            if key_present(model_id):
                members.append((name, model_id))
            else:
                logger.warning("skipping %s (%s): no API key in environment", name, model_id)
                skipped.append(name)
        return members, skipped

    def _cache_key(
        self,
        prompt: str,
        mode: str,
        *,
        rounds: int | None = None,
        proposer: str | None = None,
        converge_threshold: float | None = None,
        choices: list[str] | None = None,
    ) -> str:
        """Build the cache key for a run from the resolved, secret-free identity.

        Uses the *resolved* member ids and the synthesizer/judge identity so two
        runs collide only when they would genuinely produce equivalent output.
        Members that would be skipped for a missing key are excluded -- a cache
        entry reflects the council that actually ran, so a key reappearing later
        produces the same membership. No environment value is read here.
        """
        members, _skipped = self._available_members()
        synth_id = self.config.resolve_model_id(self.synthesizer)
        used_prefixes = {
            model_id.split("/", 1)[0]
            for _name, model_id in [*members, (self.synthesizer, synth_id)]
            if "/" in model_id
        }
        return cache_mod.make_key(
            prompt=prompt,
            mode=mode,
            members=members,
            synthesizer=self.synthesizer,
            synthesizer_model_id=synth_id,
            temperature=self.temperature,
            timeout=self.timeout,
            rounds=rounds,
            proposer=proposer,
            converge_threshold=converge_threshold,
            choices=choices,
            extract_verdict=self.extract_verdict_enabled,
            endpoint_urls={
                prefix: endpoint.completions_url
                for prefix, endpoint in self.config.endpoints.items()
                if prefix in used_prefixes
            },
            source_bundle_digest=self.source_bundle_digest,
            protocol_version=ELITE_PROTOCOL_VERSION,
            synthesis_prompt_version=SYNTHESIS_PROMPT_VERSION,
            elite_prompt_version=ELITE_PROMPT_VERSION,
        )

    async def _cached_run(
        self,
        prompt: str,
        mode: str,
        run: Callable[[], Awaitable[CouncilResult]],
        *,
        rounds: int | None = None,
        proposer: str | None = None,
        converge_threshold: float | None = None,
        choices: list[str] | None = None,
    ) -> CouncilResult:
        """Serve ``run`` from the result cache when caching is enabled.

        On a hit the cached :class:`CouncilResult` is returned with ``cached=True``
        and the providers are not called. On a miss (or when caching is off) the
        live ``run`` executes; a successful live run is stored best-effort. Cache
        read/write failures never propagate -- they degrade to a normal live run.

        This is the single chokepoint every mode funnels through, so it is also
        where the manifest-on-every-result invariant is enforced: each returned
        result passes through :meth:`_ensure_manifest` (a no-op for the
        synthesize/raw path, which builds its own richer manifest in
        :meth:`_ask_uncached`; a fill for ``debate``/``adversarial``/``vote`` and
        for a cache hit stored before the manifest existed).
        """
        if not self.cache_enabled:
            result = await run()
            self._ensure_manifest(result, mode)
            return result

        key = self._cache_key(
            prompt,
            mode,
            rounds=rounds,
            proposer=proposer,
            converge_threshold=converge_threshold,
            choices=choices,
        )
        hit = cache_mod.load(key)
        if hit is not None:
            logger.info("cache hit for %s run (%s)", mode, key[:12])
            self._ensure_manifest(hit, mode)
            return hit

        result = await run()
        self._ensure_manifest(result, mode)
        cache_mod.store(key, result)
        return result

    def _ensure_manifest(self, result: CouncilResult, mode: str) -> None:
        """Guarantee the manifest-on-every-result invariant at the single chokepoint.

        Every mode funnels through :meth:`_cached_run`, so attaching the auditable
        :class:`ModelHarnessManifest` here makes it a true invariant rather than a
        per-mode responsibility that can silently drift (the drift this method
        fixes: ``debate``/``adversarial``/``vote`` used to return with
        ``manifest is None``).

        A no-op when ``result.manifest`` already exists: the synthesize/raw path
        builds its own manifest inside :meth:`_ask_uncached` so it can populate the
        verdict-provenance slots and re-stamp secret-safety over them, and this
        method must not overwrite that richer manifest. The
        ``debate``/``adversarial``/``vote`` wrappers return a result with
        ``manifest is None`` -- this fills it from the resolved membership and the
        collected answers, including the zero-members early-return path and a stale
        cache hit stored before this invariant existed, so no result ever escapes
        without an audit manifest.

        Membership is re-resolved via :meth:`_available_members` (the same
        resolution :meth:`_cache_key` already performs) rather than threaded back
        through every mode's return value; this keeps ``providers_called`` /
        ``model_ids`` reflecting the full resolved membership even for debate rounds
        where a member later dropped out, while the per-answer receipts are built
        from ``result.answers``. :meth:`_build_manifest` stamps ``secret_safety``
        VERIFIED when the assembled manifest is provably clean.

        Args:
            result: The result to attach a manifest to. Mutated in place.
            mode: The deliberation mode string recorded on the manifest.
        """
        if result.manifest is not None:
            return
        members, skipped = self._available_members()
        if mode == "elite" and result.elite is not None:
            result.manifest = self._build_elite_manifest(
                members=members,
                skipped=skipped,
                result=result,
            )
        else:
            result.manifest = self._build_manifest(
                mode=mode, members=members, skipped=skipped, answers=result.answers
            )

    async def fan_out(
        self,
        members: list[tuple[str, str]],
        messages_for: MessagesFor,
    ) -> list[ModelAnswer]:
        """Fan a per-member message list out concurrently and collect results.

        This is the single concurrency primitive reused by every mode (synthesize,
        raw, debate, adversarial). It never raises for a member failure: each
        member yields a :class:`ModelAnswer` carrying either an answer or an error.

        Args:
            members: ``(friendly_name, model_id)`` pairs to call.
            messages_for: Callable mapping a ``(name, model_id)`` member to the
                OpenAI-style message list to send it. Lets each mode tailor the
                prompt per member (e.g. inject peers' prior-round answers) while
                sharing the gather/partial-failure logic.

        Returns:
            One :class:`ModelAnswer` per member, in the same order as ``members``.
        """
        tasks = [
            call_model(
                name,
                model_id,
                messages_for(name, model_id),
                config=self.config,
                temperature=self.temperature,
                timeout=self.timeout,
            )
            for name, model_id in members
        ]
        # return_exceptions=True is belt-and-suspenders; call_model already
        # converts provider failures into ModelAnswer.error, but this guards
        # against any unexpected raise so one bad member can't abort the gather.
        gathered = await asyncio.gather(*tasks, return_exceptions=True)
        answers: list[ModelAnswer] = []
        for (name, model_id), outcome in zip(members, gathered, strict=True):
            if isinstance(outcome, ModelAnswer):
                answers.append(outcome)
            else:
                # call_model already redacts and never raises, so this arm only
                # fires on an UNEXPECTED escape. Redact the exception text anyway:
                # the invariant "every error string conclave surfaces is scrubbed"
                # must hold even on this defense-in-depth path (key-leak audit).
                message = redact(f"{type(outcome).__name__}: {outcome}")
                logger.warning("%s raised unexpectedly: %s", name, message)
                answers.append(ModelAnswer(name=name, model_id=model_id, error=message))
        return answers

    async def ask(self, prompt: str, synthesize: bool = True) -> CouncilResult:
        """Run the council asynchronously.

        When the result cache is enabled, an identical prior run is returned from
        cache (``CouncilResult.cached is True``) without calling the providers.

        Args:
            prompt: The user prompt to fan out.
            synthesize: When True (default), merge answers via the synthesizer.

        Returns:
            A :class:`CouncilResult` with per-member answers and (optionally) the
            synthesis. A run with zero available members returns an empty-answer
            result rather than raising.
        """
        mode = "synthesize" if synthesize else "raw"
        return await self._cached_run(
            prompt, mode, lambda: self._ask_uncached(prompt, synthesize=synthesize)
        )

    def _build_manifest(
        self,
        *,
        mode: str,
        members: list[tuple[str, str]],
        skipped: list[str],
        answers: list[ModelAnswer],
    ) -> ModelHarnessManifest:
        """Assemble the auditable :class:`ModelHarnessManifest` for a run (CAC-04).

        Builds the manifest from the resolved membership plus the collected
        member ``answers`` (one execution receipt per answer via
        :func:`conclave.providers.receipt_from_answer`). It works for both the
        normal path (``answers`` populated) and the empty-members path
        (``members``/``answers`` empty, ``skipped`` listing every requested name)
        so every live run returns a manifest. It is called for the synthesize/raw
        path directly in :meth:`_ask_uncached` and for ``debate``/``adversarial``/
        ``vote`` via :meth:`_ensure_manifest`, so the ``mode`` argument spans every
        deliberation mode.

        The ``conclave_version`` is read via a deferred import: ``conclave``
        imports this module at package init *before* it assigns ``__version__``,
        so a top-level import would resolve too early. Deferring it into this
        method (run only when a result is produced, by which point the package is
        fully initialized) mirrors the ``models._default_prompt_version`` factory.

        After assembly the manifest is scanned for secret material and its
        ``secret_safety`` stamped VERIFIED only when provably clean
        (:func:`conclave.manifest.verified_secret_safety`).

        Args:
            mode: Deliberation mode (``"synthesize"``/``"raw"``/``"debate"``/
                ``"adversarial"``/``"vote"``).
            members: ``(friendly_name, model_id)`` pairs that were called.
            skipped: Friendly names skipped for a missing key.
            answers: The collected per-member answers (empty on the no-members path).

        Returns:
            A fully-assembled, secret-safety-stamped manifest.
        """
        from . import __version__

        receipts = [
            receipt_from_answer(a, temperature=self.temperature, timeout=self.timeout)
            for a in answers
        ]
        manifest = ModelHarnessManifest(
            request_id=uuid4().hex,
            conclave_version=__version__,
            mode=mode,
            providers_considered=list(self.requested_models),
            providers_called=[name for name, _model_id in members],
            providers_skipped=[
                ProviderSkip(name=name, reason="no API key in environment") for name in skipped
            ],
            model_ids=[model_id for _name, model_id in members],
            generation_settings={"temperature": self.temperature, "timeout": self.timeout},
            receipts=receipts,
        )
        self._recompute_manifest_accounting(manifest)
        return manifest

    def _build_elite_manifest(
        self,
        *,
        members: list[tuple[str, str]],
        skipped: list[str],
        result: CouncilResult,
    ) -> ModelHarnessManifest:
        """Assemble a phase-complete manifest for an elite protocol run.

        Elite keeps every attempted member call in phase-specific artifact
        collections. Flattening those collections here preserves repeated calls
        as distinct receipts while provider membership remains unique. Failed
        gates therefore retain the completed and attempted phases without
        inventing receipts for phases that never ran.
        """
        from . import __version__

        elite = result.elite
        phase_artifacts = (
            []
            if elite is None
            else [
                ("initial", elite.initial_answers),
                ("critique", elite.critiques),
                ("revision", elite.revisions),
            ]
        )
        receipts = [
            receipt_from_answer(
                answer,
                temperature=self.temperature,
                timeout=self.timeout,
                phase=phase,
                protocol_version=ELITE_PROTOCOL_VERSION,
                prompt_version=None if phase == "initial" else ELITE_PROMPT_VERSION,
            )
            for phase, answers in phase_artifacts
            for answer in answers
        ]
        manifest = ModelHarnessManifest(
            request_id=uuid4().hex,
            conclave_version=__version__,
            mode="elite",
            providers_considered=list(self.requested_models),
            providers_called=list(dict.fromkeys(name for name, _model_id in members)),
            providers_skipped=[
                ProviderSkip(name=name, reason="no API key in environment") for name in skipped
            ],
            model_ids=list(dict.fromkeys(model_id for _name, model_id in members)),
            generation_settings={"temperature": self.temperature, "timeout": self.timeout},
            receipts=receipts,
        )
        self._recompute_manifest_accounting(manifest)
        return manifest

    @staticmethod
    def _recompute_manifest_accounting(manifest: ModelHarnessManifest) -> None:
        """Recompute every manifest aggregate from its complete receipt ledger."""
        manifest.total_latency_ms = sum(receipt.latency_ms for receipt in manifest.receipts)
        usages = [receipt.usage for receipt in manifest.receipts if receipt.usage is not None]
        manifest.total_usage = (
            TokenUsage(
                prompt_tokens=sum(usage.prompt_tokens for usage in usages),
                completion_tokens=sum(usage.completion_tokens for usage in usages),
                total_tokens=sum(usage.total_tokens for usage in usages),
            )
            if usages
            else None
        )
        # Cost is known only when every actual call carries a trustworthy priced
        # value. A partially-known total would be misleading, so unknown remains
        # None rather than being silently coerced to zero.
        costs = [receipt.estimated_cost for receipt in manifest.receipts]
        manifest.estimated_cost = (
            sum(cost for cost in costs if cost is not None)
            if costs and all(cost is not None for cost in costs)
            else None
        )
        manifest.redacted_errors = [
            receipt.error for receipt in manifest.receipts if receipt.error is not None
        ]
        manifest.secret_safety = verified_secret_safety(manifest)

    def _append_manifest_receipts(
        self,
        result: CouncilResult,
        receipts: list[ProviderExecutionReceipt],
    ) -> None:
        """Append explicit call receipts and refresh all derived manifest fields."""
        if result.manifest is None or not receipts:
            return
        result.manifest.receipts.extend(receipts)
        self._recompute_manifest_accounting(result.manifest)

    async def _ask_uncached(self, prompt: str, synthesize: bool = True) -> CouncilResult:
        """The live ask path (no cache consultation). See :meth:`ask`.

        A :class:`ModelHarnessManifest` is attached on **every** return, including
        the zero-members early return, so a consumer can always audit what ran
        (CAC-04). The empty-members manifest carries no receipts, the full skip
        list, and a VERIFIED ``secret_safety`` stamp.
        """
        mode = "synthesize" if synthesize else "raw"
        members, skipped = self._available_members()
        result = CouncilResult(prompt=prompt, mode=mode, skipped=skipped)

        if not members:
            logger.warning("no council members have keys available; nothing to run")
            result.manifest = self._build_manifest(
                mode=mode, members=members, skipped=skipped, answers=[]
            )
            return result

        base_messages = [{"role": "user", "content": prompt}]
        result.answers = await self.fan_out(members, lambda _name, _model_id: base_messages)
        result.manifest = self._build_manifest(
            mode=mode, members=members, skipped=skipped, answers=result.answers
        )

        if synthesize:
            # Prose synthesis first, then the structured verdict over the SAME
            # answers. ``_apply_verdict`` runs after the manifest exists so it can
            # populate the manifest's verdict-provenance slots; it is skipped in
            # raw mode (no synthesizer call) and is opt-out via the constructor
            # flag (resolved inside the helper). The no-members early return above
            # never reaches here, so a memberless run carries no verdict.
            synthesis_answer = await self._synthesize(result)
            if synthesis_answer is not None:
                self._append_manifest_receipts(
                    result,
                    [
                        receipt_from_answer(
                            synthesis_answer,
                            temperature=self.temperature,
                            timeout=self.timeout,
                            phase="synthesis",
                            protocol_version=(
                                ELITE_PROTOCOL_VERSION if result.mode == "elite" else None
                            ),
                            prompt_version=SYNTHESIS_PROMPT_VERSION,
                        )
                    ],
                )
            await self._apply_verdict(result)
        return result

    async def ask_stream(self, prompt: str, synthesize: bool = True) -> AsyncIterator[StreamEvent]:
        """Stream a synthesize/raw run, yielding incremental :class:`StreamEvent`s.

        The streaming counterpart of :meth:`ask` (issue #7). Members are fanned
        out concurrently and their tokens are interleaved as ``member_delta`` /
        ``member_done`` events; when ``synthesize`` is ``True`` the synthesizer's
        tokens follow as ``synthesis_delta`` / ``synthesis_done``; a terminal
        ``done`` event carries the fully-assembled :class:`CouncilResult`, whose
        shape matches the non-streaming path exactly. Streaming applies to the
        synthesize/raw path only -- ``debate``/``adversarial`` are not streamed.

        **Cache interaction.** When the result cache is enabled and an identical
        prior run is cached, there are no live provider tokens to stream: the
        cached final text is rendered in **one shot** -- a single
        ``member_delta`` per member (and a single ``synthesis_delta`` if a
        synthesis was cached) followed by the matching ``*_done`` events and the
        terminal ``done`` (with ``result.cached is True``). The providers are not
        called. On a cache **miss**, the live stream runs and, on completion, the
        assembled result is stored so a later ``--stream`` or buffered run hits.

        Args:
            prompt: The user prompt to fan out.
            synthesize: When ``True`` (default), stream the synthesizer too.

        Yields:
            :class:`StreamEvent` objects; the last one is always ``type="done"``.
        """
        from .streaming import stream_ask

        mode = "synthesize" if synthesize else "raw"

        if self.cache_enabled:
            key = self._cache_key(prompt, mode)
            hit = cache_mod.load(key)
            if hit is not None:
                logger.info("cache hit for %s stream (%s)", mode, key[:12])
                for event in self._replay_cached(hit):
                    yield event
                return

            # Live miss: stream, capture the terminal result, then store it.
            final: CouncilResult | None = None
            async for event in stream_ask(self, prompt, synthesize=synthesize):
                if event.type == "done" and event.result is not None:
                    final = event.result
                yield event
            if final is not None:
                cache_mod.store(key, final)
            return

        async for event in stream_ask(self, prompt, synthesize=synthesize):
            yield event

    @staticmethod
    def _replay_cached(result: CouncilResult) -> list[StreamEvent]:
        """Render a cached :class:`CouncilResult` as one-shot stream events.

        A cache hit has no live tokens, so each member's full cached answer is
        emitted as a single ``member_delta`` + ``member_done`` (errors emit only
        ``member_done``), the cached synthesis as a single ``synthesis_delta`` +
        ``synthesis_done``, and finally the terminal ``done`` carrying the cached
        result verbatim (``cached is True``). This keeps the streaming consumer's
        event contract intact without fabricating a fake token-by-token stream.
        """
        events: list[StreamEvent] = []
        for ans in result.answers:
            if ans.answer:
                events.append(
                    StreamEvent(
                        type="member_delta",
                        name=ans.name,
                        model_id=ans.model_id,
                        text=ans.answer,
                    )
                )
            events.append(
                StreamEvent(type="member_done", name=ans.name, model_id=ans.model_id, answer=ans)
            )
        if result.synthesis is not None:
            events.append(
                StreamEvent(
                    type="synthesis_delta",
                    name=result.synthesizer,
                    model_id=result.synthesizer_model_id,
                    text=result.synthesis,
                )
            )
            events.append(
                StreamEvent(
                    type="synthesis_done",
                    name=result.synthesizer,
                    model_id=result.synthesizer_model_id,
                    answer=ModelAnswer(
                        name=result.synthesizer or "synthesizer",
                        model_id=result.synthesizer_model_id or "",
                        answer=result.synthesis,
                    ),
                )
            )
        events.append(StreamEvent(type="done", result=result))
        return events

    async def _synthesize(self, result: CouncilResult) -> ModelAnswer | None:
        """Run the synthesizer over the successful answers, mutating ``result``.

        This is the buffered (non-streaming) synthesize path; the streaming
        counterpart :func:`conclave.streaming._stream_synthesis` mirrors it
        short-circuit for short-circuit. The synthesizer model is
        ``self.synthesizer`` (resolved per the precedence documented in the module
        docstring: constructor arg, else config, else the ``"claude"`` default).

        Every degraded outcome is made observable on ``result`` -- none is
        silent. On success ``result.synthesis`` holds the merged answer; on any
        of the three short-circuits ``result.synthesis`` stays ``None`` and
        ``result.synthesis_error`` carries the reason:

        * **no usable answers** -- every member failed/was skipped, so there is
          nothing to merge;
        * **synthesizer unkeyed** -- ``self.synthesizer``'s API key is absent, so
          the raw member answers are returned with an explanatory error;
        * **synthesizer call failed** -- the synthesizer provider errored, and its
          error text is surfaced verbatim.

        The synthesizer identity (``synthesizer`` / ``synthesizer_model_id``) is
        recorded on ``result`` before the key check so a consumer can see *which*
        model was selected even when it could not run. The prompt used is the
        versioned :data:`_SYNTH_SYSTEM`; the version tag already lives on
        ``result.prompt_version``.
        """
        usable = result.successful_answers
        if not usable:
            result.synthesis_error = "no successful member answers to synthesize"
            logger.warning(result.synthesis_error)
            return None

        synth_id = self.config.resolve_model_id(self.synthesizer)
        result.synthesizer = self.synthesizer
        result.synthesizer_model_id = synth_id

        if not key_present(synth_id):
            result.synthesis_error = (
                f"synthesizer '{self.synthesizer}' ({synth_id}) has no API key; "
                "returning raw answers only"
            )
            logger.warning(result.synthesis_error)
            return None

        blocks = "\n\n".join(
            f"### Answer from {a.name} ({a.model_id})"
            f"{f' (Answer ID: {a.answer_id})' if a.answer_id else ''}\n{a.answer}"
            for a in usable
        )
        user_content = (
            f"Original prompt:\n{result.prompt}\n\n"
            f"Council answers:\n\n{blocks}\n\n"
            "Now produce the consolidated answer."
        )
        answer = await self.synthesize_blocks(_SYNTH_SYSTEM, user_content)
        if answer.ok:
            result.synthesis = answer.answer
        else:
            result.synthesis_error = answer.error
        return answer

    async def _apply_verdict(
        self,
        result: CouncilResult,
        *,
        record_receipts: bool = True,
    ) -> None:
        """Run verdict extraction over the answers and hoist it onto ``result``.

        The SINGLE shared verdict-resolution path. Both the buffered
        ``ask``/:meth:`_ask_uncached` path (here) and the streaming path
        (CAC-06-STREAM) call this one method, so the rule "verdict object is
        canonical, top-level fields are mirrors" is written exactly once and the
        two paths cannot drift. It mutates ``result`` in place and returns
        ``None``.

        The verdict object (``result.verdict``) is the canonical adjudication; the
        convenience fields (``consensus_score``/``method``/``label``,
        ``conflicts``, ``provider_votes``, ``minority_reports``) are HOISTED
        mirrors of the same values for callers that don't want to reach through
        ``result.verdict``. They are populated only when a verdict is present;
        when it is absent they stay at their ``None``/empty defaults.

        Consensus is NEVER recomputed here: it is carried verbatim from
        :func:`conclave.verdict_synthesis.extract_verdict`, which computes it
        deterministically from the model's clustering (DD-1). This method only
        delegates and copies fields.

        **Opt-out & cost.** When ``self.extract_verdict_enabled`` is ``False`` this
        is a no-op and every verdict field is left at its default. When enabled it
        makes a SECOND synthesizer call (the extraction round-trip, plus one repair
        retry on a malformed response) distinct from the prose synthesis call --
        the documented cost of the default-on verdict.

        ``extract_verdict`` owns the N<2 gate (it returns ``verdict=None`` with the
        reason ``"fewer than 2 responding members"`` and makes NO LLM call in that
        case), so this method delegates unconditionally rather than duplicating the
        responder-counting logic; that keeps a single code path and lets the
        manifest carry the N<2 reason. ``extract_verdict`` never raises, and this
        method only assigns already-secret-free objects afterward, so no defensive
        try/except is needed.

        When ``result.manifest`` exists its verdict-provenance slots are populated
        (extractor identity + prompt version, absent reason, consensus method,
        verdict type) and the manifest's ``secret_safety`` stamp is RE-RUN over the
        final content: the stamp was first computed in :meth:`_build_manifest`
        before these fields existed, so re-stamping keeps the VERIFIED claim honest
        over the manifest a consumer actually receives. The new fields (a resolved
        model id, a prompt-version string, the ``verdict_type``/``consensus_method``
        literals) are provably key-free, so the stamp stays VERIFIED.

        Args:
            result: The in-progress :class:`CouncilResult` (answers + manifest
                already attached). Mutated in place.
        """
        if not self.extract_verdict_enabled:
            return

        # Lazy import mirrors this module's deferred-import style (``modes`` /
        # ``streaming`` are imported inside methods) and sidesteps any import-cycle
        # risk between council and the verdict engine.
        from .verdict_synthesis import extract_verdict as extract_verdict_fn

        synthesizer_name = self.synthesizer
        synth_id = self.config.resolve_model_id(self.synthesizer)
        vsr = await extract_verdict_fn(
            result.prompt,
            result.answers,
            synthesizer_name=synthesizer_name,
            synthesizer_model_id=synth_id,
            config=self.config,
            temperature=self.temperature,
            timeout=self.timeout,
            protocol_version=(ELITE_PROTOCOL_VERSION if result.mode == "elite" else None),
        )

        if record_receipts:
            self._append_manifest_receipts(result, vsr.attempt_receipts)

        result.verdict = vsr.verdict
        if vsr.verdict is not None:
            # Hoist the canonical verdict's values to the top-level mirrors.
            result.consensus_score = vsr.verdict.consensus_score
            result.consensus_method = vsr.verdict.consensus_method
            result.consensus_label = vsr.verdict.consensus_label
            result.conflicts = vsr.verdict.conflicts
            result.provider_votes = vsr.verdict.provider_votes
            result.minority_reports = vsr.verdict.minority_reports

        if result.manifest is not None:
            result.manifest.verdict_extraction = vsr.extraction
            result.manifest.verdict_absent_reason = vsr.verdict_absent_reason
            result.manifest.consensus_method = vsr.verdict.consensus_method if vsr.verdict else None
            result.manifest.verdict_type = vsr.verdict.verdict_type if vsr.verdict else None
            # Re-stamp over the now-complete manifest so the VERIFIED claim covers
            # the verdict-provenance fields just written (they are key-free).
            result.manifest.secret_safety = verified_secret_safety(result.manifest)

    async def synthesize_blocks(self, system_prompt: str, user_content: str) -> ModelAnswer:
        """Call the synthesizer model with an arbitrary system + user message.

        Shared by synthesize mode, debate's final consolidation, and the
        adversarial judge so the synthesizer call path (and its error capture)
        is written once. Callers are responsible for checking ``key_present``
        on the synthesizer beforehand when they need a distinct no-key message;
        this method still returns a ``ModelAnswer.error`` if the call fails.

        Args:
            system_prompt: System instruction for the synthesizer/judge.
            user_content: The user-role content (prompt + answers/critiques).

        Returns:
            A :class:`ModelAnswer` from the synthesizer model.
        """
        synth_id = self.config.resolve_model_id(self.synthesizer)
        messages = [
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": user_content},
        ]
        return await call_model(
            self.synthesizer,
            synth_id,
            messages,
            config=self.config,
            temperature=self.temperature,
            timeout=self.timeout,
        )

    async def debate(
        self, prompt: str, rounds: int = 2, converge_threshold: float | None = None
    ) -> CouncilResult:
        """Run a multi-round debate. See :func:`conclave.modes.run_debate`.

        Round 1 is an independent fan-out; rounds 2..N show each member its
        peers' anonymized prior answers; the synthesizer consolidates survivors.
        Cache-served when caching is enabled (``rounds`` and the resolved
        ``converge_threshold`` are part of the key).

        Args:
            prompt: The user prompt.
            rounds: Maximum number of rounds (the historic fixed count).
            converge_threshold: Opt-in early-stop threshold. ``None`` (default)
                defers to ``config.converge_threshold`` (off unless set in
                ``~/.conclave/config.yml``); an explicit value overrides it for
                this call. With early-stop off the debate runs exactly ``rounds``,
                identical to the historic behavior. Mirrors the ``cache``
                None-defers-to-config convention.
        """
        from .modes import run_debate

        # Resolve the opt-in here (mirrors ``cache``: explicit arg wins, else
        # config) so the cache key reflects what will actually run.
        threshold = (
            self.config.converge_threshold if converge_threshold is None else converge_threshold
        )
        return await self._cached_run(
            prompt,
            "debate",
            lambda: run_debate(self, prompt, rounds=rounds, converge_threshold=threshold),
            rounds=rounds,
            converge_threshold=threshold,
        )

    async def adversarial(self, prompt: str, proposer: str | None = None) -> CouncilResult:
        """Run propose -> refute -> verdict. See :func:`conclave.modes.run_adversarial`.

        ``proposer`` (friendly name) defaults to the first requested member.
        Cache-served when caching is enabled (``proposer`` is part of the key).
        """
        from .modes import run_adversarial

        return await self._cached_run(
            prompt,
            "adversarial",
            lambda: run_adversarial(self, prompt, proposer=proposer),
            proposer=proposer,
        )

    async def elite(self, prompt: str) -> CouncilResult:
        """Run the answer/claim-audited Elite Decision Protocol.

        Completed runs synthesize the council's revised answers and apply the
        canonical structured verdict. A run that fails any three-responder gate
        returns its phase artifacts without synthesis or verdict extraction.
        """
        from .modes import run_elite

        async def run() -> CouncilResult:
            result = await run_elite(self, prompt)
            if result.elite is not None and result.elite.completed:
                synthesis_answer = await self._synthesize(result)
                self._ensure_manifest(result, "elite")
                if synthesis_answer is not None:
                    self._append_manifest_receipts(
                        result,
                        [
                            receipt_from_answer(
                                synthesis_answer,
                                temperature=self.temperature,
                                timeout=self.timeout,
                                phase="synthesis",
                                protocol_version=ELITE_PROTOCOL_VERSION,
                                prompt_version=SYNTHESIS_PROMPT_VERSION,
                            )
                        ],
                    )
                if result.synthesis is None:
                    result.elite.decision_readiness = "not_ready"
                    result.elite.readiness_reasons = ["synthesis.failed"]
                elif not self.extract_verdict_enabled:
                    result.elite.decision_readiness = "indeterminate"
                    result.elite.readiness_reasons = ["adjudication.disabled"]
                else:
                    await self._apply_verdict(result)
                    self._set_elite_readiness(result)
            return result

        return await self._cached_run(prompt, "elite", run)

    @staticmethod
    def _set_elite_readiness(result: CouncilResult) -> None:
        """Classify Elite readiness after required synthesis and adjudication."""
        elite = result.elite
        if elite is None:
            return
        if result.verdict is not None:
            elite.decision_readiness = "ready"
            elite.readiness_reasons = []
            return

        absent_reason = (
            result.manifest.verdict_absent_reason if result.manifest is not None else None
        )
        if absent_reason == "verdict extraction failed schema validation":
            elite.decision_readiness = "not_ready"
            elite.readiness_reasons = ["adjudication.verdict_extraction_failed"]
        elif absent_reason == "fewer than 2 responding members":
            elite.decision_readiness = "not_ready"
            elite.readiness_reasons = ["adjudication.insufficient_responders"]
        elif absent_reason == "open-ended prompt (no decision/review to adjudicate)":
            elite.decision_readiness = "indeterminate"
            elite.readiness_reasons = ["adjudication.open_ended"]
        else:
            elite.decision_readiness = "indeterminate"
            elite.readiness_reasons = ["adjudication.unknown"]

    async def aclose(self) -> None:
        """Close the shared pooled HTTP client.

        Library users running their own event loop (e.g. a server) should call
        this on shutdown so the process-wide connection pool is released and no
        "unclosed client" warning is emitted under strict asyncio. It is safe to
        call more than once; the pooled client is recreated lazily on next use.

        The synchronous wrappers (:meth:`ask_sync`, :meth:`debate_sync`,
        :meth:`adversarial_sync`) already close the client automatically before
        their event loop ends, so CLI/sync callers do not need to call this.
        """
        await transport.aclose()

    def close_sync(self) -> None:
        """Synchronous wrapper around :meth:`aclose` for non-async callers."""
        self._run_sync(self.aclose, "close_sync", close_client=False)

    @staticmethod
    def _run_sync(
        coro_factory: Callable[[], asyncio.Future | object],
        label: str,
        *,
        close_client: bool = True,
    ):
        """Run an async council method synchronously, guarding nested loops.

        ``asyncio.run`` creates (and tears down) a fresh event loop per call. The
        pooled httpx client is bound to whichever loop first used it, so we close
        it inside that same loop's ``finally`` before the loop is destroyed --
        otherwise the pool leaks and asyncio emits an "unclosed client" warning.
        ``close_client=False`` is used by :meth:`close_sync` itself to avoid
        recursively re-closing.
        """
        try:
            asyncio.get_running_loop()
        except RuntimeError:
            pass
        else:
            raise RuntimeError(
                f"{label}() called from within a running event loop; await the async method instead"
            )

        async def _runner():
            try:
                return await coro_factory()
            finally:
                if close_client:
                    await transport.aclose()

        return asyncio.run(_runner())

    def ask_sync(self, prompt: str, synthesize: bool = True) -> CouncilResult:
        """Synchronous wrapper around :meth:`ask`.

        Safe to call from non-async code. Raises ``RuntimeError`` if invoked
        from inside a running event loop -- use :meth:`ask` there instead.
        """
        return self._run_sync(lambda: self.ask(prompt, synthesize=synthesize), "ask_sync")

    def stream_sync(
        self,
        prompt: str,
        on_event: Callable[[StreamEvent], None],
        synthesize: bool = True,
    ) -> CouncilResult:
        """Drive :meth:`ask_stream` synchronously, invoking ``on_event`` per event.

        For non-async callers (the CLI ``--stream`` path). Each
        :class:`StreamEvent` is passed to ``on_event`` as it arrives so live
        output can be rendered; the fully-assembled :class:`CouncilResult` (from
        the terminal ``done`` event) is returned. Closes the pooled HTTP client
        when the loop ends, like the other ``*_sync`` wrappers. Raises
        ``RuntimeError`` if invoked from inside a running event loop -- iterate
        :meth:`ask_stream` directly there instead.

        Args:
            prompt: The user prompt to fan out.
            on_event: Callback invoked once per :class:`StreamEvent` in order.
            synthesize: When ``True`` (default), stream the synthesizer too.

        Returns:
            The final :class:`CouncilResult` carried by the ``done`` event.
        """

        async def _consume() -> CouncilResult:
            final: CouncilResult | None = None
            async for event in self.ask_stream(prompt, synthesize=synthesize):
                on_event(event)
                if event.type == "done" and event.result is not None:
                    final = event.result
            # ask_stream always ends with a done event carrying a result; fall
            # back to an empty result only as a defensive guard.
            return final if final is not None else CouncilResult(prompt=prompt)

        return self._run_sync(_consume, "stream_sync")

    def debate_sync(
        self, prompt: str, rounds: int = 2, converge_threshold: float | None = None
    ) -> CouncilResult:
        """Synchronous wrapper around :meth:`debate`."""
        return self._run_sync(
            lambda: self.debate(prompt, rounds=rounds, converge_threshold=converge_threshold),
            "debate_sync",
        )

    def adversarial_sync(self, prompt: str, proposer: str | None = None) -> CouncilResult:
        """Synchronous wrapper around :meth:`adversarial`."""
        return self._run_sync(
            lambda: self.adversarial(prompt, proposer=proposer), "adversarial_sync"
        )

    def elite_sync(self, prompt: str) -> CouncilResult:
        """Synchronous wrapper around :meth:`elite`."""
        return self._run_sync(lambda: self.elite(prompt), "elite_sync")

    async def vote(self, prompt: str, choices: list[str]) -> CouncilResult:
        """Run a constrained-choice vote. See :func:`conclave.modes.run_vote`.

        Each member receives the prompt and a labelled option set (A, B, C, ...)
        and must respond with a single letter. Results are tallied and a winner
        (plurality) or split is reported on ``result.vote``.

        Args:
            prompt: The question to vote on.
            choices: Two or more option strings. At least 2 required.
        """
        from .modes import run_vote

        return await self._cached_run(
            prompt,
            "vote",
            lambda: run_vote(self, prompt, choices=choices),
            choices=choices,
        )

    def vote_sync(self, prompt: str, choices: list[str]) -> CouncilResult:
        """Synchronous wrapper around :meth:`vote`."""
        return self._run_sync(lambda: self.vote(prompt, choices=choices), "vote_sync")