dispatch

dispatch

Sending a command, and telling the truth about what happened.

This is the layer the assistant talks to. It brings together the registry (who), presence (is it there), the tool catalogue (is this allowed), the queue (durability) and the transport (how) — and it is deliberately the only place that decides what the user is told.

The rule it exists to keep is spec §57, restated as code: a command that was merely queued must never be reported as done. Each terminal status carries a French sentence that is true of that status and no other.

Classes

Functions:

dispatch_command

dispatch_command(
    *,
    target_device_id: str,
    tool: str,
    arguments: Mapping[str, Any] | None = None,
    registry: DeviceRegistry | None = None,
    queue: CommandQueue | None = None,
    idempotency_key: str = "",
    transport=None,
) -> DispatchResult

Send tool to target_device_id, or explain precisely why not.

Source code in src/diapason/mesh/dispatch.py

def dispatch_command(
    *,
    target_device_id: str,
    tool: str,
    arguments: Mapping[str, Any] | None = None,
    registry: DeviceRegistry | None = None,
    queue: CommandQueue | None = None,
    idempotency_key: str = "",
    transport=None,
) -> DispatchResult:
    """Send *tool* to *target_device_id*, or explain precisely why not."""
    from diapason.mesh.identity import device_identity, owner_id

    registry = registry or DeviceRegistry()
    queue = queue or CommandQueue()
    send = transport or deliver

    # 1. The tool must exist. An assistant asking for something outside the
    #    catalogue gets nothing — there is no fallback path (spec §21).
    spec = get_remote_tool(tool)
    if spec is None:
        return _refused(
            "UNSUPPORTED",
            f"L'outil « {tool} » n'existe pas.",
            tool=tool,
            target=target_device_id,
        )

    # 2. The target must be a device we know and still trust.
    device = registry.find(target_device_id)
    if device is None:
        return _refused(
            "DENIED",
            "Cet appareil n'est pas appairé.",
            tool=tool,
            target=target_device_id,
        )
    if device.get("trustLevel") != "TRUSTED":
        return _refused(
            "DENIED",
            f"L'appareil « {device.get('name')} » n'est plus autorisé.",
            tool=tool,
            target=target_device_id,
        )

    # 3. Capabilities are checked BEFORE sending, so an impossible request
    #    fails here with an explanation instead of on the far side with a
    #    shrug (spec §6).
    granted = set(device.get("capabilities") or [])
    if spec.capability not in granted:
        return _refused(
            "UNSUPPORTED",
            f"« {device.get('name')} » ne peut pas faire cela.",
            tool=tool,
            target=target_device_id,
        )

    # 4. Build, validate locally, sign. Validating before signing means a
    #    malformed command never travels at all.
    identity = device_identity()
    command = build_command(
        owner_id=owner_id(),
        origin_device_id=identity.device_id,
        target_device_id=target_device_id,
        tool=tool,
        arguments=arguments or {},
        requires_confirmation=spec.requires_confirmation,
        idempotency_key=idempotency_key,
        # A tool that is allowed to wait must be given long enough to be worth
        # waiting for. One minute expires before a sleeping phone ever polls.
        ttl_ms=(
            QUEUED_TTL_MS
            if spec.offline_policy == "QUEUE_UNTIL_EXPIRATION"
            else DEFAULT_TTL_MS
        ),
    )
    try:
        spec.validate(command.arguments)
    except CommandRejected as exc:
        return _refused(exc.code, exc.message, tool=tool, target=target_device_id)

    # 4 bis. SCELLER, entre la validation et la signature. L'ordre n'est pas
    #        indifférent : on valide le CLAIR — le validateur n'a jamais à
    #        connaître le chiffrement — puis on scelle, puis on signe. La
    #        signature couvre donc le chiffré, ce qui la laisse vérifiable
    #        sans la clé de déchiffrement.
    try:
        cle_du_pair = scellement.doit_sceller(device, registry=registry)
    except scellement.ScellementExige as exc:
        return _refused("DENIED", str(exc), tool=tool, target=target_device_id)
    if cle_du_pair:
        try:
            command = scellement.sceller_commande(command, cle_du_pair)
        except Exception as exc:  # noqa: BLE001
            # Un scellement qui échoue ne part PAS en clair : ce serait la
            # panne silencieuse que tout ce chantier existe pour empêcher.
            logger.warning("scellement impossible : %s", exc, exc_info=True)
            return _refused(
                "DENIED",
                "Cette commande n'a pas pu être chiffrée, donc elle n'est pas partie.",
                tool=tool,
                target=target_device_id,
            )

    # 5. Idempotency: the same intent, already sent, returns what we know
    #    rather than doing it twice (spec §45).
    if idempotency_key:
        known = queue.find_by_idempotency(
            idempotency_key, origin_device_id=identity.device_id
        )
        if known is not None and known["status"] not in {"PENDING", "QUEUED"}:
            return known

    signed = sign_command(command)
    # 6. Recorded BEFORE the wire (transactional outbox, spec §16): a command
    #    that left without a row behind it is one nobody can report on.
    queue.enqueue(signed)

    # 7. A device we cannot dial is not a device we cannot reach. A phone has
    #    no address and comes to fetch its own commands, so leaving this one
    #    in the queue is delivery, not failure — and must be worded as such.
    if device_collects_its_own(device):
        return _await_collection(queue, signed, device)

    # 8. Reachability decides the policy, and the policy decides the wording.
    if not is_reachable(device):
        presence = presence_of(device)
        return _apply_offline_policy(queue, signed, device, spec, presence)

    # 9. Deliver.
    queue.record_attempt(signed.command_id)
    try:
        response = send(signed, device)
    except RemoteRefusal as exc:
        # The device answered. Sending its owner to inspect the network
        # would be sending them after a fault that does not exist.
        if not exc.retryable:
            return queue.mark(
                signed.command_id,
                "FAILED",
                user_message=str(exc),
                error_code=f"REFUSED_{exc.status_code}",
            )
        return _apply_offline_policy(
            queue,
            signed,
            device,
            spec,
            presence_of(device),
            reason=str(exc),
            answered=True,
            error_code=f"REFUSED_{exc.status_code}",
        )
    except TransportError as exc:
        return _apply_offline_policy(
            queue, signed, device, spec, presence_of(device), reason=str(exc)
        )
    except Exception as exc:  # noqa: BLE001 - never let a transport bug lie
        logger.exception("mesh dispatch failed")
        return queue.mark(
            signed.command_id,
            "FAILED",
            user_message="L'envoi de la commande a échoué.",
            error_code=type(exc).__name__,
        )

    status = str(response.get("status") or "FAILED").upper()
    return queue.mark(
        signed.command_id,
        status,
        user_message=str(response.get("userSafeMessage") or ""),
        result=response.get("result"),
        error_code=str(response.get("errorCode") or ""),
    )

flush_pending

flush_pending(
    *,
    registry: DeviceRegistry | None = None,
    queue: CommandQueue | None = None,
    transport=None,
    limit_per_device: int = 20,
) -> dict[str, int]

Deliver what is waiting for devices that have come back.

This is the half of « partira dès son retour » that makes the sentence true. Without it that message is a promise nobody keeps: a command for a laptop that was asleep sits QUEUED until it expires, and the user was told it would arrive.

Only push devices are drained here. A polling device fetches its own, and delivering to it twice is not a courtesy — it is the same command arriving on two paths.

Never raises: this runs on a timer, and a peer that cannot be reached is the normal state of a fleet whose devices come and go.

Source code in src/diapason/mesh/dispatch.py

def flush_pending(
    *,
    registry: DeviceRegistry | None = None,
    queue: CommandQueue | None = None,
    transport=None,
    limit_per_device: int = 20,
) -> dict[str, int]:
    """Deliver what is waiting for devices that have come back.

    This is the half of « partira dès son retour » that makes the sentence
    true. Without it that message is a promise nobody keeps: a command for a
    laptop that was asleep sits QUEUED until it expires, and the user was
    told it would arrive.

    Only push devices are drained here. A polling device fetches its own,
    and delivering to it twice is not a courtesy — it is the same command
    arriving on two paths.

    Never raises: this runs on a timer, and a peer that cannot be reached is
    the normal state of a fleet whose devices come and go.
    """
    registry = registry or DeviceRegistry()
    queue = queue or CommandQueue()
    send = transport or deliver

    queue.expire_stale()
    delivered = 0
    skipped = 0

    for device in registry.list_devices(include_revoked=False):
        if device.get("trustLevel") != "TRUSTED":
            continue
        if device_collects_its_own(device) or not is_reachable(device):
            continue
        if not str(device.get("address") or "").strip():
            continue

        for row in queue.pending_for(device["deviceId"], limit=limit_per_device):
            envelope = queue.envelope_of(row["commandId"])
            if envelope is None:
                continue
            queue.record_attempt(row["commandId"])
            try:
                response = send(envelope, device)
            except RemoteRefusal as exc:
                # A refusal is not an absence. Left to the `except Exception`
                # below, this row stayed QUEUED and was retried every fifteen
                # seconds for its whole six-hour life — roughly 1440 times —
                # before expiring under "la commande a expiré avant d'être
                # livrée", which blames a delay for what was a verdict. Worse,
                # the `break` froze the WHOLE queue for this device behind it.
                if not exc.retryable:
                    queue.mark(
                        row["commandId"],
                        "FAILED",
                        user_message=str(exc),
                        error_code=f"REFUSED_{exc.status_code}",
                    )
                    delivered += 1
                    # Not `break`: the device is demonstrably answering, so
                    # the next row deserves its own attempt rather than
                    # inheriting this one's verdict.
                    continue
                skipped += 1
                break
            except Exception:  # noqa: BLE001 - still away; try again next tick
                skipped += 1
                # Stop at the first failure for this device: the rest of its
                # queue will fail the same way, and hammering an unreachable
                # peer with twenty timeouts helps nobody.
                break
            status = str(response.get("status") or "FAILED").upper()
            queue.mark(
                row["commandId"],
                status,
                user_message=str(response.get("userSafeMessage") or ""),
                result=response.get("result"),
                error_code=str(response.get("errorCode") or ""),
            )
            delivered += 1

    return {"delivered": delivered, "skipped": skipped}

receive_command

receive_command(
    raw: Mapping[str, Any],
    *,
    registry: DeviceRegistry | None = None,
    queue: CommandQueue | None = None,
    nonces=None,
    executor=None,
) -> DispatchResult

Verify an inbound command and run it. The far end of the wire.

Source code in src/diapason/mesh/dispatch.py

def receive_command(
    raw: Mapping[str, Any],
    *,
    registry: DeviceRegistry | None = None,
    queue: CommandQueue | None = None,
    nonces=None,
    executor=None,
) -> DispatchResult:
    """Verify an inbound command and run it. The far end of the wire."""
    from diapason.mesh.commands import NonceStore, verify_command
    from diapason.mesh.identity import device_identity, owner_id

    registry = registry or DeviceRegistry()
    queue = queue or CommandQueue()
    nonces = nonces or NonceStore()
    identity = device_identity()

    # Un refus fait ÉCHO à ce qui a été demandé ; il ne se présente pas.
    #
    # Ces deux chemins renvoyaient `identity.device_id` — l'identifiant
    # permanent de cette machine. Un expéditeur légitime le connaît déjà : il
    # vient de l'écrire dans son enveloppe, et le recevoir en retour ne lui
    # apprend rien. Mais cette route vit hors du mur d'authentification, et
    # le 26 août 2026 un POST au corps vide, sans la moindre créance, a
    # obtenu « mac-73d5a8b0c742888a » depuis le Wi-Fi. Un identifiant stable
    # est ce avec quoi on suit une machine d'un réseau à l'autre.
    def _cible_demandee() -> str:
        try:
            return str(raw.get("targetDeviceId") or "")
        except AttributeError:
            return ""

    try:
        command = verify_command(
            raw,
            registry=registry,
            local_device_id=identity.device_id,
            local_owner_id=owner_id(),
            nonces=nonces,
        )
    except CommandRejected as exc:
        return {
            "commandId": str(raw.get("commandId") or ""),
            "targetDeviceId": _cible_demandee(),
            "status": exc.code,
            "errorCode": exc.code,
            "userSafeMessage": exc.message,
        }
    except (CommandError, TypeError, ValueError, AttributeError) as exc:
        # A malformed envelope is a refusal, not a crash. This route is open
        # to anyone who can reach the port, so an exception escaping here
        # became a 500 and a stack trace in the log on demand.
        logger.info("commande illisible refusée : %s", type(exc).__name__)
        return {
            "commandId": "",
            "targetDeviceId": _cible_demandee(),
            "status": "DENIED",
            "errorCode": "MALFORMED",
            "userSafeMessage": "Cette commande est illisible.",
        }

    # Idempotency on the receiving side too: a redelivered command replays
    # its recorded result rather than acting twice (spec §17).
    known = queue.find_by_idempotency(
        command.idempotency_key, origin_device_id=command.origin_device_id
    )
    if known is not None and known["status"] not in {"PENDING", "QUEUED"}:
        return known

    queue.enqueue(command, status="RUNNING")
    if executor is None:
        return queue.mark(
            command.command_id,
            "UNSUPPORTED",
            user_message="Cet appareil ne sait pas encore exécuter cette commande.",
            error_code="NO_EXECUTOR",
        )
    try:
        result = executor(command)
    except Exception as exc:  # noqa: BLE001
        logger.exception("mesh command execution failed")
        return queue.mark(
            command.command_id,
            "FAILED",
            user_message="L'exécution de la commande a échoué sur cet appareil.",
            error_code=type(exc).__name__,
        )
    # 26 August 2026: this was five lines, and it wrote "SUCCESS" without
    # ever looking at what the executor had answered. `_desktop_open` already
    # returned {"ok": False, "userSafeMessage": "Échec de l'ouverture."} —
    # under a comment explaining that announcing "opened" on a failure would
    # be exactly the defect not to commit. The field was written, nobody read
    # it, and the screen showed "SUCCESS — Échec de l'ouverture.": the word
    # "failure" crossed the whole mesh while the status said the opposite
    # (§100).
    #
    # "ok" is REQUIRED, never assumed. A silent executor is one we know
    # nothing about, and inferring success from silence is precisely the
    # shape this defect took.
    verdict: dict[str, Any] = dict(result or {})
    if "ok" not in verdict:
        logger.error(
            "silent executor for %s: no 'ok' field in its result",
            command.tool,
        )
        return queue.mark(
            command.command_id,
            "FAILED",
            user_message="Cet appareil n'a pas su dire si l'action a abouti.",
            error_code="EXECUTOR_SILENT",
            result=verdict or None,
        )
    if not verdict["ok"]:
        return queue.mark(
            command.command_id,
            "FAILED",
            user_message=str(
                verdict.get("userSafeMessage") or "L'action a échoué sur cet appareil."
            ),
            error_code=str(verdict.get("errorCode") or "EXECUTOR_FAILED"),
            result=verdict,
        )
    return queue.mark(
        command.command_id,
        "SUCCESS",
        user_message=str(verdict.get("userSafeMessage") or "C'est fait."),
        result=verdict,
    )