Skip to content

approval_bridge

approval_bridge

Chat tool confirmations, answered through the existing approval bell.

The tool executor calls a synchronous confirm_callback(prompt) -> bool whenever a tool declares requires_confirmation. Until now the server either hardcoded lambda _prompt: True (managed-agents chat) or passed nothing at all (main chat agent), so confirmation-gated tools ran silently or failed outright. This bridge gives that callback a real answer path:

  • agent.tool_approval = "auto" — legacy opt-in that confirms immediately.
  • agent.tool_approval = "ask" — queue the confirmation into the ApprovalStore that already feeds the frontend bell (GET /v1/approvals/pending + approve/deny), then wait for the user's decision. Timeout or denial refuses the tool.

The callback runs in the tool executor's worker thread, so blocking here never blocks the event loop; the store is opened per call because sqlite connections do not travel across threads.

Functions:

current_mode

current_mode() -> str

The configured approval mode, defensively normalized.

Source code in src/diapason/server/approval_bridge.py
def current_mode() -> str:
    """The configured approval mode, defensively normalized."""
    try:
        from diapason.core.config import load_config

        mode = (load_config().agent.tool_approval or "ask").strip().lower()
        return mode if mode in ("auto", "ask") else "ask"
    except Exception:  # noqa: BLE001 - config trouble must fail closed
        return "ask"

resumer_la_demande

resumer_la_demande(prompt: str) -> str

La demande en une ligne lisible, pour le centre de notifications.

Source code in src/diapason/server/approval_bridge.py
def resumer_la_demande(prompt: str) -> str:
    """La demande en une ligne lisible, pour le centre de notifications."""
    propre = " ".join(str(prompt or "").split())
    trouve = _NOM_OUTIL_RE.search(propre)
    if trouve:
        return f"Diapason veut utiliser {trouve.group(1)}"
    return propre[:110] or "Diapason demande une autorisation"

announce_approval

announce_approval(title: str, body: str) -> None

Poser une demande dans les notifications sans bloquer l'appelant.

EN TÂCHE DE FOND, et c'est le point : notifier_macos attend osascript jusqu'à dix secondes (livraison.py), soit près du quart du budget vocal de quarante-cinq. L'attente d'approbation ne doit pas financer sa propre annonce. Best-effort et silencieuse : une notification ratée ne fait pas tomber le tour d'outil qu'elle accompagne.

Source code in src/diapason/server/approval_bridge.py
def announce_approval(title: str, body: str) -> None:
    """Poser une demande dans les notifications sans bloquer l'appelant.

    EN TÂCHE DE FOND, et c'est le point : notifier_macos attend osascript
    jusqu'à dix secondes (livraison.py), soit près du quart du budget vocal
    de quarante-cinq. L'attente d'approbation ne doit pas financer sa propre
    annonce. Best-effort et silencieuse : une notification ratée ne fait pas
    tomber le tour d'outil qu'elle accompagne.
    """

    def _poser() -> None:
        try:
            from diapason.heartbeat.livraison import notifier_macos

            notifier_macos(title, body)
        except Exception:  # noqa: BLE001 - l'annonce est un bonus, jamais une porte
            logger.debug("annonce d'approbation impossible", exc_info=True)

    threading.Thread(target=_poser, daemon=True, name="annonce-approbation").start()

tool_confirm_callback

tool_confirm_callback(
    wait_s: float = DEFAULT_WAIT_S,
) -> Callable[[str], bool]

Build the confirm callback for chat agents.

The mode is read at CALL time, not build time: flipping the composer chip applies to the very next tool, no server restart involved.

Source code in src/diapason/server/approval_bridge.py
def tool_confirm_callback(wait_s: float = DEFAULT_WAIT_S) -> Callable[[str], bool]:
    """Build the confirm callback for chat agents.

    The mode is read at CALL time, not build time: flipping the composer
    chip applies to the very next tool, no server restart involved.
    """

    def confirm(prompt: str) -> bool:
        if current_mode() == "auto":
            return True
        try:
            return _await_decision(str(prompt or "Tool call"), wait_s)
        except Exception:  # noqa: BLE001 - a broken store must fail closed
            logger.exception("tool confirmation bridge failed")
            return False

    return confirm