feat(vote): add constrained-choice vote deliberation mode (CAC-09 / #3)

src/conclave/cache.py

@@ -85,6 +85,7 @@ def make_key(
     rounds: int | None = None,
     proposer: str | None = None,
     converge_threshold: float | None = None,
+    choices: list[str] | None = None,
 ) -> str:
     """Build the stable cache key (sha256 hex) for a council run.
 
@@ -133,6 +134,8 @@ def make_key(
         payload["converge_threshold"] = converge_threshold
     if mode == "adversarial":
         payload["proposer"] = proposer
+    if mode == "vote":
+        payload["choices"] = choices or []
 
     canonical = json.dumps(payload, sort_keys=True, separators=(",", ":"), ensure_ascii=False)
     return hashlib.sha256(canonical.encode("utf-8")).hexdigest()

src/conclave/cli.py

@@ -216,6 +216,59 @@ def _render_debate(result: CouncilResult) -> None:
     _print_synthesis(result, title="FINAL SYNTHESIS")
 
 
+def _render_vote(result: CouncilResult) -> None:
+    """Render a vote run: tally table then winner/split line."""
+    _print_skipped(result)
+    vote = result.vote
+    if vote is None:
+        err_console.print("[yellow]No vote result produced.[/yellow]")
+        return
+
+    labels = [chr(65 + i) for i in range(len(vote.choices))]
+    label_to_choice = dict(zip(labels, vote.choices, strict=False))
+
+    table = Table(title="Vote Tally", show_header=True)
+    table.add_column("Choice", style="bold")
+    table.add_column("Label", justify="center")
+    table.add_column("Votes", justify="right")
+    table.add_column("Voters")
+
+    for lbl in labels:
+        choice_text = label_to_choice.get(lbl, lbl)
+        cnt = vote.tally.get(lbl, 0)
+        voters = [name for name, v in vote.votes.items() if v == lbl]
+        voters_str = ", ".join(voters) if voters else "—"
+        style = "green" if lbl == vote.winner else ""
+        table.add_row(choice_text, lbl, str(cnt), voters_str, style=style)
+
+    console.print(table)
+
+    unparsed = [name for name, v in vote.votes.items() if v is None]
+    if unparsed:
+        err_console.print(f"[yellow]Unrecognised/failed responses: {', '.join(unparsed)}[/yellow]")
+
+    if vote.winner is not None:
+        winner_text = label_to_choice.get(vote.winner, vote.winner)
+        console.print(
+            Panel(
+                f"[bold]{vote.winner}. {winner_text}[/bold]",
+                title="[bold green]WINNER[/bold green]",
+                border_style="green",
+            )
+        )
+    elif vote.split:
+        tied = [f"{lbl}. {label_to_choice.get(lbl, lbl)}" for lbl in sorted(vote.tally)]
+        console.print(
+            Panel(
+                f"[yellow]Tie: {' vs '.join(tied)}[/yellow]",
+                title="[bold yellow]SPLIT[/bold yellow]",
+                border_style="yellow",
+            )
+        )
+    else:
+        err_console.print("[yellow]No votes were cast.[/yellow]")
+
+
 def _render_adversarial(result: CouncilResult) -> None:
     """Render an adversarial run: proposal, critiques, then the verdict."""
     _print_skipped(result)
@@ -303,10 +356,11 @@ def on_event(event: StreamEvent) -> None:
     "raw": _render_human,
     "debate": _render_debate,
     "adversarial": _render_adversarial,
+    "vote": _render_vote,
 }
 
 
-_VALID_MODES = {"synthesize", "raw", "debate", "adversarial"}
+_VALID_MODES = {"synthesize", "raw", "debate", "adversarial", "vote"}
 
 # Threshold used when --converge is passed without an explicit --converge-threshold.
 # High by design: an early stop should require answers that are nearly stable
@@ -390,6 +444,14 @@ def ask(
         "-p",
         help="Proposer model name (adversarial mode; defaults to first member).",
     ),
+    choices: str | None = typer.Option(
+        None,
+        "--choices",
+        help=(
+            "Comma-separated list of options for vote mode "
+            "(e.g. 'Option A,Option B,Option C'). Required for --mode vote."
+        ),
+    ),
     as_json: bool = typer.Option(
         False, "--json", help="Emit the full result as JSON instead of panels."
     ),
@@ -436,6 +498,12 @@ def ask(
         )
         raise typer.Exit(code=2)
 
+    if mode_lower == "vote" and not choices:
+        err_console.print(
+            "[red]--mode vote requires --choices (e.g. --choices 'Yes,No,Abstain').[/red]"
+        )
+        raise typer.Exit(code=2)
+
     cfg = load_config()
     members = cfg.resolve_council(council)
     if not members:
@@ -463,6 +531,9 @@ def ask(
         result = c.debate_sync(prompt, rounds=rounds, converge_threshold=threshold)
     elif mode_lower == "adversarial":
         result = c.adversarial_sync(prompt, proposer=proposer)
+    elif mode_lower == "vote":
+        choice_list = [ch.strip() for ch in (choices or "").split(",") if ch.strip()]
+        result = c.vote_sync(prompt, choices=choice_list)
     else:
         result = c.ask_sync(prompt, synthesize=(mode_lower == "synthesize"))
 

src/conclave/council.py

@@ -202,6 +202,7 @@ def _cache_key(
         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.
 
@@ -223,6 +224,7 @@ def _cache_key(
             rounds=rounds,
             proposer=proposer,
             converge_threshold=converge_threshold,
+            choices=choices,
         )
 
     async def _cached_run(
@@ -234,6 +236,7 @@ async def _cached_run(
         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.
 
@@ -251,6 +254,7 @@ async def _cached_run(
             rounds=rounds,
             proposer=proposer,
             converge_threshold=converge_threshold,
+            choices=choices,
         )
         hit = cache_mod.load(key)
         if hit is not None:
@@ -881,3 +885,27 @@ def adversarial_sync(self, prompt: str, proposer: str | None = None) -> CouncilR
         return self._run_sync(
             lambda: self.adversarial(prompt, proposer=proposer), "adversarial_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")

src/conclave/models.py

@@ -173,13 +173,35 @@ def successful_critiques(self) -> list[ModelAnswer]:
         return [c for c in self.critiques if c.ok]
 
 
+class VoteResult(BaseModel):
+    """The tally from a constrained-choice vote run.
+
+    Attributes:
+        choices: The fixed option labels offered to members.
+        votes: Map of member friendly name → the label they chose (or ``None``
+            when a member's answer could not be parsed to a valid choice).
+        tally: Map of label → vote count across all members that returned a
+            parseable vote. Members that errored or returned an unrecognised
+            response are excluded from the tally.
+        winner: The label with the most votes, or ``None`` on a tie.
+        split: ``True`` when no single choice won an outright plurality (tie).
+    """
+
+    choices: list[str] = Field(default_factory=list)
+    votes: dict[str, str | None] = Field(default_factory=dict)
+    tally: dict[str, int] = Field(default_factory=dict)
+    winner: str | None = None
+    split: bool = False
+
+
 class CouncilResult(BaseModel):
     """The full outcome of a council run.
 
     Attributes:
         prompt: The original user prompt.
         mode: The run mode that produced this result
-            (``"synthesize"`` | ``"raw"`` | ``"debate"`` | ``"adversarial"``).
+            (``"synthesize"`` | ``"raw"`` | ``"debate"`` | ``"adversarial"``
+            | ``"vote"``).
         answers: One ``ModelAnswer`` per attempted council member. For
             ``debate`` this mirrors the final round so existing consumers that
             read ``answers``/``synthesis`` keep working unchanged.
@@ -261,6 +283,7 @@ class CouncilResult(BaseModel):
     skipped: list[str] = Field(default_factory=list)
     rounds: list[DebateRound] = Field(default_factory=list)
     adversarial: AdversarialResult | None = None
+    vote: VoteResult | None = None
     cached: bool = False
     converged: bool = False
     convergence_score: float | None = None

src/conclave/modes.py

@@ -22,12 +22,13 @@
 
 from __future__ import annotations
 
+import re
 from difflib import SequenceMatcher
 from typing import TYPE_CHECKING
 
 from . import prompts
 from .logging import get_logger
-from .models import AdversarialResult, CouncilResult, DebateRound, ModelAnswer
+from .models import AdversarialResult, CouncilResult, DebateRound, ModelAnswer, VoteResult
 from .registry import key_present
 
 if TYPE_CHECKING:  # avoid a circular import at runtime; only needed for typing
@@ -36,6 +37,125 @@
 logger = get_logger("modes")
 
 
+async def run_vote(
+    council: Council,
+    prompt: str,
+    choices: list[str],
+) -> CouncilResult:
+    """Run a constrained-choice vote and return a :class:`CouncilResult`.
+
+    Each council member is shown the prompt and a fixed option set labelled A,
+    B, C, ... and asked to respond with a single letter. Responses are tallied;
+    the plurality winner (if any) is stored in ``result.vote.winner``. A tie
+    sets ``result.vote.split = True`` and ``result.vote.winner = None``.
+
+    Args:
+        council: The :class:`Council` providing fan-out and config.
+        prompt: The user question to vote on.
+        choices: Two or more option strings (e.g. ``["Option 1", "Option 2"]``).
+            Each choice is assigned a consecutive uppercase letter starting at A.
+
+    Returns:
+        A :class:`CouncilResult` with ``mode="vote"`` and ``vote`` populated.
+        ``synthesis`` carries a human-readable summary of the tally.
+        Zero available members yields an empty result, not an error.
+    """
+    if len(choices) < 2:
+        raise ValueError("vote mode requires at least 2 choices")
+
+    members, skipped = council._available_members()
+    result = CouncilResult(prompt=prompt, mode="vote", skipped=skipped)
+
+    if not members:
+        logger.warning("no council members have keys available; nothing to vote")
+        result.vote = VoteResult(choices=choices)
+        return result
+
+    labels = [chr(65 + i) for i in range(len(choices))]
+    label_to_choice = dict(zip(labels, choices, strict=False))
+
+    messages_for = _vote_messages_for(prompt, choices)
+    result.answers = await council.fan_out(members, messages_for)
+
+    # Parse each member's response into a label.
+    member_votes: dict[str, str | None] = {}
+    for ans in result.answers:
+        if not ans.ok or not ans.answer:
+            member_votes[ans.name] = None
+            continue
+        raw = ans.answer.strip().upper()
+        # Accept the first standalone label letter (word-boundary match).
+        # A letter buried inside a word (e.g. "cANnot") is not accepted.
+        chosen: str | None = None
+        for m in re.finditer(r"\b([A-Z])\b", raw):
+            if m.group(1) in label_to_choice:
+                chosen = m.group(1)
+                break
+        member_votes[ans.name] = chosen
+
+    # Tally votes.
+    tally: dict[str, int] = {lbl: 0 for lbl in labels}
+    for chosen in member_votes.values():
+        if chosen is not None:
+            tally[chosen] = tally.get(chosen, 0) + 1
+
+    # Remove labels with zero votes for a cleaner tally.
+    tally = {lbl: cnt for lbl, cnt in tally.items() if cnt > 0}
+
+    # Determine winner (plurality = most votes; None on tie).
+    winner: str | None = None
+    split = False
+    if tally:
+        max_votes = max(tally.values())
+        leaders = [lbl for lbl, cnt in tally.items() if cnt == max_votes]
+        if len(leaders) == 1:
+            winner = leaders[0]
+        else:
+            split = True
+
+    vote_result = VoteResult(
+        choices=choices,
+        votes=member_votes,
+        tally=tally,
+        winner=winner,
+        split=split,
+    )
+    result.vote = vote_result
+
+    # Build a human-readable synthesis summarising the outcome.
+    result.synthesis = _vote_summary(vote_result, label_to_choice, len(result.answers))
+    return result
+
+
+def _vote_messages_for(prompt: str, choices: list[str]):
+    """Build the per-member message factory for a vote round."""
+    messages = [
+        {"role": "system", "content": prompts.VOTE_SYSTEM},
+        {"role": "user", "content": prompts.vote_user(prompt, choices)},
+    ]
+    return lambda _name, _model_id: messages
+
+
+def _vote_summary(vote_result: VoteResult, label_to_choice: dict[str, str], n_members: int) -> str:
+    """Build a brief text summary of the vote outcome."""
+    lines = [f"Vote result ({n_members} member(s) polled):"]
+    for lbl, cnt in sorted(vote_result.tally.items(), key=lambda x: -x[1]):
+        choice_text = label_to_choice.get(lbl, lbl)
+        lines.append(f"  {lbl}. {choice_text}: {cnt} vote(s)")
+    unparsed = sum(1 for v in vote_result.votes.values() if v is None)
+    if unparsed:
+        lines.append(f"  (unrecognised/failed responses: {unparsed})")
+    if vote_result.winner is not None:
+        winner_text = label_to_choice.get(vote_result.winner, vote_result.winner)
+        lines.append(f"\nWinner: {vote_result.winner}. {winner_text}")
+    elif vote_result.split:
+        tied = [f"{lbl}. {label_to_choice.get(lbl, lbl)}" for lbl in sorted(vote_result.tally)]
+        lines.append(f"\nTie: {' vs '.join(tied)}")
+    else:
+        lines.append("\nNo votes cast.")
+    return "\n".join(lines)
+
+
 async def run_debate(
     council: Council,
     prompt: str,