broker¶

def deregister_agent(agent_id: int) -> bool:
    """Soft-delete the agent and drop its placement.

    Args:
        agent_id: Agent id to deregister.

    Returns:
        ``True`` if a row was flipped from ``active`` to ``deregistered``;
        ``False`` if no matching active agent existed.

    Raises:
        click.UsageError: If ``agent_id`` is the root Director of any
            fleet — torn down via ``cafleet fleet delete`` instead.
        click.ClickException: If ``agent_id`` is the built-in Administrator.
    """
    with _shared.write_session() as session:
        is_root_director = session.execute(
            select(exists().where(Fleet.director_agent_id == agent_id))
        ).scalar_one()
        if is_root_director:
            raise click.UsageError(
                "cannot deregister the root Director; "
                "use 'cafleet fleet delete' instead"
            )

        card_json = session.execute(
            select(Agent.agent_card_json).where(Agent.agent_id == agent_id)
        ).scalar_one_or_none()
        if card_json is not None and _shared.is_administrator(card_json):
            raise click.ClickException("Administrator cannot be deregistered")
        deregistered = session.execute(
            update(Agent)
            .where(
                Agent.agent_id == agent_id,
                Agent.status == "active",
            )
            .values(
                status="deregistered",
                deregistered_at=_shared.now_iso(),
            )
            .returning(Agent.agent_id)
        ).all()
        if deregistered:
            session.execute(
                delete(AgentPlacement).where(AgentPlacement.agent_id == agent_id)
            )
            # Runtime config has no historical value; drop it on the same
            # lifecycle as the placement.
            monitor.delete_agent_monitor_row(session, agent_id)
    return bool(deregistered)

def get_agent(agent_id: int, fleet_id: int) -> dict | None:
    """Return the active agent's detail (with placement) or None.

    Args:
        agent_id: Agent id to look up.
        fleet_id: Fleet id the agent must belong to.

    Returns:
        Dict with ``agent_id``, ``name``, ``description``, ``status``,
        ``registered_at``, ``kind`` (``"user"`` or the Administrator kind),
        and ``placement`` (the placement sub-dict or ``None``). Returns
        ``None`` if no active agent matches.
    """
    with _shared.read_session() as session:
        agent = session.execute(
            select(Agent).where(
                Agent.agent_id == agent_id,
                Agent.fleet_id == fleet_id,
                Agent.status == "active",
            )
        ).scalar_one_or_none()

        if agent is None:
            return None

        placement_row = session.execute(
            select(AgentPlacement).where(AgentPlacement.agent_id == agent_id)
        ).scalar_one_or_none()

    result: dict = {
        "agent_id": agent.agent_id,
        "name": agent.name,
        "description": agent.description,
        "status": agent.status,
        "registered_at": agent.registered_at,
        "kind": (
            _shared.ADMINISTRATOR_KIND
            if _shared.is_administrator(agent.agent_card_json)
            else "user"
        ),
        "placement": None,
    }
    if placement_row is not None:
        result["placement"] = _shared.placement_dict(placement_row)
    return result

def get_agent_names(agent_ids: list[int]) -> dict[int, str]:
    """Batch ``agent_id → name`` lookup including deregistered agents."""
    if not agent_ids:
        return {}
    with _shared.read_session() as session:
        rows = session.execute(
            select(Agent.agent_id, Agent.name).where(Agent.agent_id.in_(agent_ids))
        ).all()
    return {row.agent_id: row.name for row in rows}

def list_agents(fleet_id: int) -> list[dict]:
    """Return all active agents in the fleet."""
    stmt = select(
        Agent.agent_id,
        Agent.name,
        Agent.description,
        Agent.registered_at,
    ).where(
        Agent.fleet_id == fleet_id,
        Agent.status == "active",
    )
    with _shared.read_session() as session:
        rows = session.execute(stmt).all()
    return [
        {
            "agent_id": row.agent_id,
            "name": row.name,
            "description": row.description,
            "status": "active",
            "registered_at": row.registered_at,
        }
        for row in rows
    ]

def list_fleet_agents(fleet_id: int) -> list[dict]:
    """Return active agents plus deregistered agents that still own tasks.

    ``kind`` is derived in SQL via ``json_extract`` so the card blob never
    leaves SQLite — otherwise we would materialize every row's JSON just to
    compute a one-token discriminator. ``coalesce`` handles cards without a
    ``cafleet.kind`` path by substituting an empty string.
    """
    has_tasks = exists().where(
        or_(
            Task.context_id == Agent.agent_id,
            Task.from_agent_id == Agent.agent_id,
        )
    )
    kind_expr = func.coalesce(
        func.json_extract(Agent.agent_card_json, "$.cafleet.kind"), ""
    )
    stmt = select(
        Agent.agent_id,
        Agent.name,
        Agent.description,
        Agent.status,
        Agent.registered_at,
        kind_expr.label("kind_raw"),
    ).where(
        Agent.fleet_id == fleet_id,
        or_(
            Agent.status == "active",
            and_(Agent.status == "deregistered", has_tasks),
        ),
    )
    with _shared.read_session() as session:
        rows = session.execute(stmt).all()
    return [
        {
            "agent_id": row.agent_id,
            "name": row.name,
            "description": row.description,
            "status": row.status,
            "registered_at": row.registered_at,
            "kind": (
                _shared.ADMINISTRATOR_KIND
                if row.kind_raw == _shared.ADMINISTRATOR_KIND
                else "user"
            ),
        }
        for row in rows
    ]

def register_agent(
    fleet_id: int,
    name: str,
    description: str,
    skills: list[dict] | None = None,
    placement: dict | None = None,
) -> dict:
    """Register a new agent in the fleet and optionally create its placement.

    Rejects soft-deleted fleets with a message that differs from the
    "not found" case so callers can surface the right recovery hint. When
    ``placement`` is supplied, the named Director must be active in the
    same fleet, must not be the Administrator, and must be the fleet's root
    Director (``fleets.director_agent_id``). A fleet has exactly one Director:
    nested teams are forbidden, so every member placement's
    ``director_agent_id`` equals the fleet root and a non-root value is
    rejected.

    Args:
        fleet_id: Fleet id the new agent belongs to.
        name: Short human-identifiable label.
        description: One-sentence purpose statement.
        skills: Optional list of skill dicts persisted into the agent's
            ``agent_card_json`` blob.
        placement: Optional dict carrying ``director_agent_id``,
            ``tmux_session``, ``tmux_window_id``, ``tmux_pane_id``, and
            ``coding_agent``. When present, an ``AgentPlacement`` row is
            created alongside the agent.

    Returns:
        Dict with ``agent_id``, ``name``, and ``registered_at``.

    Raises:
        click.UsageError: If the fleet does not exist, is soft-deleted, the
            named Director is not active in the same fleet, or the placement
            ``director_agent_id`` is not the fleet's root Director.
        click.ClickException: If the named Director is the built-in
            Administrator.
    """
    sess = get_fleet(fleet_id)
    if sess is None:
        raise click.UsageError(f"Fleet '{fleet_id}' not found.")
    if sess["deleted_at"] is not None:
        raise click.UsageError(f"fleet {fleet_id} is deleted")

    registered_at = _shared.now_iso()
    agent_card = {
        "name": name,
        "description": description,
        "skills": skills or [],
    }

    with _shared.write_session() as session:
        if placement is not None:
            director_id = placement["director_agent_id"]
            director_card = session.execute(
                select(Agent.agent_card_json).where(
                    Agent.agent_id == director_id,
                    Agent.fleet_id == fleet_id,
                    Agent.status == "active",
                )
            ).scalar_one_or_none()
            if director_card is None:
                raise click.UsageError(
                    f"Director agent '{director_id}' not found or not active "
                    f"in fleet '{fleet_id}'."
                )
            if _shared.is_administrator(director_card):
                raise click.ClickException("Administrator cannot be a director")
            root_director_id = sess["director_agent_id"]
            if director_id != root_director_id:
                raise click.UsageError(
                    f"nested teams are not supported; placement director_agent_id "
                    f"{director_id} must equal the fleet root "
                    f"Director {root_director_id}."
                )

        agent = Agent(
            fleet_id=fleet_id,
            name=name,
            description=description,
            status="active",
            registered_at=registered_at,
            agent_card_json=json.dumps(agent_card),
        )
        session.add(agent)
        session.flush()
        agent_id = agent.agent_id
        if placement is not None:
            session.add(
                AgentPlacement(
                    agent_id=agent_id,
                    director_agent_id=placement["director_agent_id"],
                    tmux_session=placement["tmux_session"],
                    tmux_window_id=placement["tmux_window_id"],
                    tmux_pane_id=placement["tmux_pane_id"],
                    coding_agent=placement["coding_agent"],
                    created_at=registered_at,
                )
            )
            # Enroll the pane-bound agent in monitoring, atomically with its
            # placement insert (only agents with a pane can be pinged).
            monitor.enroll_agent(session, agent_id)

    return {
        "agent_id": agent_id,
        "name": name,
        "registered_at": registered_at,
    }

def update_placement_pane_id(agent_id: int, pane_id: str) -> dict | None:
    """Patch the agent's placement with a freshly resolved tmux pane id.

    Called after ``split_window`` returns the spawned pane's id so the
    placement row reflects the live pane rather than the placeholder used
    during the initial INSERT.

    Args:
        agent_id: Agent id whose placement should be updated.
        pane_id: New ``tmux_pane_id`` value.

    Returns:
        The refreshed placement dict, or ``None`` if no placement row was
        affected.
    """
    with _shared.write_session() as session:
        updated = session.execute(
            update(AgentPlacement)
            .where(AgentPlacement.agent_id == agent_id)
            .values(tmux_pane_id=pane_id)
            .returning(AgentPlacement.agent_id)
        ).first()
        if updated is None:
            return None
        row = session.execute(
            select(AgentPlacement).where(AgentPlacement.agent_id == agent_id)
        ).scalar_one()
    return _shared.placement_dict(row)

def verify_agent_fleet(agent_id: int, fleet_id: int) -> bool:
    """Return True iff the agent belongs to the fleet (any status).

    Args:
        agent_id: Agent id to verify.
        fleet_id: Fleet id to check membership against.

    Returns:
        ``True`` if a matching row exists; ``False`` otherwise. Status is
        ignored — deregistered agents still pass.
    """
    with _shared.read_session() as session:
        return session.execute(
            select(
                exists().where(
                    Agent.agent_id == agent_id,
                    Agent.fleet_id == fleet_id,
                )
            )
        ).scalar_one()

def create_fleet(
    label: str | None = None,
    *,
    director_context: MultiplexerContext,
    coding_agent: str,
) -> dict:
    """Atomically bootstrap a fleet with its root Director and Administrator.

    The fleet row is written first with ``director_agent_id=NULL`` and
    back-filled once the Director's agent row exists, so the column is
    DB-nullable even though the post-bootstrap invariant is NOT NULL.

    Args:
        label: Optional human-readable label for the fleet.
        director_context: Resolved tmux pane identity for the root Director,
            obtained via ``Multiplexer.context_discovery``.
        coding_agent: Operator-declared metadata that lands in the root
            Director's ``placement.coding_agent`` column. The CLI is the only
            caller and always supplies it (default ``'claude'`` lives at the
            Click layer).

    Returns:
        A dict carrying ``fleet_id``, ``label``, ``created_at``,
        ``administrator_agent_id``, and a ``director`` sub-dict with the
        Director's identity and placement metadata.
    """
    created_at = _shared.now_iso()
    director_card = {
        "name": _DIRECTOR_NAME,
        "description": _DIRECTOR_DESCRIPTION,
        "skills": [],
    }
    director_placement = {
        "director_agent_id": None,
        "tmux_session": director_context.session,
        "tmux_window_id": director_context.window_id,
        "tmux_pane_id": director_context.pane_id,
        "coding_agent": coding_agent,
        "created_at": created_at,
    }

    with _shared.write_session() as session:
        fleet = Fleet(
            label=label,
            created_at=created_at,
            deleted_at=None,
            director_agent_id=None,
        )
        session.add(fleet)
        session.flush()
        fleet_id = fleet.fleet_id
        director = Agent(
            fleet_id=fleet_id,
            name=_DIRECTOR_NAME,
            description=_DIRECTOR_DESCRIPTION,
            status="active",
            registered_at=created_at,
            deregistered_at=None,
            agent_card_json=json.dumps(director_card),
        )
        session.add(director)
        session.flush()
        director_agent_id = director.agent_id
        session.add(AgentPlacement(agent_id=director_agent_id, **director_placement))
        session.flush()
        # Enroll the root Director (pane-bound) in monitoring; the Administrator
        # below has no placement and is intentionally not enrolled.
        monitor.enroll_agent(session, director_agent_id)
        session.execute(
            update(Fleet)
            .where(Fleet.fleet_id == fleet_id)
            .values(director_agent_id=director_agent_id)
        )
        administrator_card = {
            "name": "Administrator",
            "description": f"Built-in administrator agent for fleet {fleet_id}",
            "skills": [],
            "cafleet": {"kind": _shared.ADMINISTRATOR_KIND},
        }
        administrator = Agent(
            fleet_id=fleet_id,
            name=administrator_card["name"],
            description=administrator_card["description"],
            status="active",
            registered_at=created_at,
            deregistered_at=None,
            agent_card_json=json.dumps(administrator_card),
        )
        session.add(administrator)
        session.flush()
        administrator_agent_id = administrator.agent_id

    return {
        "fleet_id": fleet_id,
        "label": label,
        "created_at": created_at,
        "administrator_agent_id": administrator_agent_id,
        "director": {
            "agent_id": director_agent_id,
            "name": _DIRECTOR_NAME,
            "description": _DIRECTOR_DESCRIPTION,
            "registered_at": created_at,
            "placement": director_placement,
        },
    }

def delete_fleet(fleet_id: int) -> dict:
    """Soft-delete a fleet and deregister its agents, in one transaction.

    Tasks are left untouched so audit history survives. Idempotent: re-running
    against an already-deleted row short-circuits on the ``deleted_at IS NULL``
    guard and returns ``deregistered_count=0``.

    Args:
        fleet_id: Fleet id to soft-delete.

    Returns:
        Dict with ``deregistered_count`` — the number of agents flipped from
        ``active`` to ``deregistered`` by this call.

    Raises:
        click.ClickException: If the fleet does not exist.
    """
    now = _shared.now_iso()
    with _shared.write_session() as session:
        fleet_exists = session.execute(
            select(exists().where(Fleet.fleet_id == fleet_id))
        ).scalar_one()
        if not fleet_exists:
            # ClickException exits 1 (matching ``fleet show``); UsageError
            # would print a Usage: banner + exit 2, wrong for a runtime miss.
            raise click.ClickException(f"fleet '{fleet_id}' not found.")

        soft_deleted = session.execute(
            update(Fleet)
            .where(
                Fleet.fleet_id == fleet_id,
                Fleet.deleted_at.is_(None),
            )
            .values(deleted_at=now)
            .returning(Fleet.fleet_id)
        ).all()
        if not soft_deleted:
            return {"deregistered_count": 0}

        deregistered = session.execute(
            update(Agent)
            .where(
                Agent.fleet_id == fleet_id,
                Agent.status == "active",
            )
            .values(status="deregistered", deregistered_at=now)
            .returning(Agent.agent_id)
        ).all()
        deregistered_count = len(deregistered)
        agents_in_fleet = select(Agent.agent_id).where(Agent.fleet_id == fleet_id)
        session.execute(
            delete(AgentPlacement).where(AgentPlacement.agent_id.in_(agents_in_fleet))
        )
        monitor.delete_fleet_monitor_rows(session, fleet_id)

    return {"deregistered_count": deregistered_count}

def get_fleet(fleet_id: int) -> dict | None:
    """Return the fleet row (including soft-deleted) or None.

    The returned dict exposes ``deleted_at`` so callers can distinguish a
    missing fleet from a soft-deleted one — ``register_agent`` relies on
    this to reject soft-deleted fleets with a different error message.

    Args:
        fleet_id: Fleet id to look up.

    Returns:
        Dict with ``fleet_id``, ``label``, ``created_at``, ``deleted_at``,
        and ``director_agent_id``, or ``None`` if no row exists.
    """
    with _shared.read_session() as session:
        result = session.execute(select(Fleet).where(Fleet.fleet_id == fleet_id))
        row = result.scalar_one_or_none()
    if row is None:
        return None
    return {
        "fleet_id": row.fleet_id,
        "label": row.label,
        "created_at": row.created_at,
        "deleted_at": row.deleted_at,
        "director_agent_id": row.director_agent_id,
    }

def list_fleets() -> list[dict]:
    """Return non-soft-deleted fleets with their active agent counts."""
    stmt = (
        select(
            Fleet.fleet_id,
            Fleet.director_agent_id,
            Fleet.label,
            Fleet.created_at,
            func.count(Agent.agent_id).label("agent_count"),
        )
        .select_from(Fleet)
        .outerjoin(
            Agent,
            and_(
                Agent.fleet_id == Fleet.fleet_id,
                Agent.status == "active",
            ),
        )
        .where(Fleet.deleted_at.is_(None))
        .group_by(
            Fleet.fleet_id,
            Fleet.director_agent_id,
            Fleet.label,
            Fleet.created_at,
        )
        .order_by(Fleet.created_at.desc(), Fleet.fleet_id.asc())
    )
    with _shared.read_session() as session:
        rows = session.execute(stmt).all()
    return [
        {
            "fleet_id": row.fleet_id,
            "director_agent_id": row.director_agent_id,
            "label": row.label,
            "created_at": row.created_at,
            "agent_count": row.agent_count,
        }
        for row in rows
    ]

def list_members(fleet_id: int) -> list[dict]:
    """Return the fleet's active members, with placements.

    A fleet has exactly one Director (the root), so every member placement's
    ``director_agent_id`` equals the fleet root. Members are selected by
    ``director_agent_id IS NOT NULL``, which lists all members and excludes
    the root Director's own placement (the only row with a ``NULL`` director).

    Args:
        fleet_id: Fleet id to scope the query to.

    Returns:
        List of dicts each carrying ``agent_id``, ``name``, ``description``,
        ``status``, ``registered_at``, and ``placement``.
    """
    stmt = _base_members_select(fleet_id)
    with _shared.read_session() as session:
        rows = session.execute(stmt).all()
    return [
        {
            "agent_id": row.agent_id,
            "name": row.name,
            "description": row.description,
            "status": row.status,
            "registered_at": row.registered_at,
            "placement": _shared.placement_dict(row),
        }
        for row in rows
    ]

def list_members_with_activity(fleet_id: int) -> list[dict]:
    """``list_members`` plus per-member activity proxies sourced from ``tasks``.

    Scoped by ``fleet_id`` only — the flat single-Director model means every
    member placement's ``director_agent_id`` equals the fleet root, so the
    ``director_agent_id IS NOT NULL`` member filter applies here too (and
    excludes the root Director's own placement).

    ``last_sent`` / ``last_recv`` / ``last_ack`` aggregate ``status_timestamp``
    over the ``tasks`` table per agent. All three filter ``Task.type !=
    'broadcast_summary'`` (mirrors ``poll_tasks``); broadcast_summary rows
    land in the broadcaster's own context with ``status_state='completed'``
    and would otherwise pollute every proxy for the broadcaster.

    Args:
        fleet_id: Fleet id to scope the query to.

    Returns:
        List of dicts as in :func:`list_members`, additionally carrying
        ``last_sent``, ``last_recv``, ``last_ack`` (ISO timestamps or
        ``None``), and ``idle`` — the integer-second delta between ``now``
        and the most recent of ``last_sent`` / ``last_recv``, or ``None``
        when both are ``None``.
    """
    last_sent_sq = (
        select(func.max(Task.status_timestamp))
        .where(
            Task.from_agent_id == Agent.agent_id,
            _shared.NOT_BROADCAST_SUMMARY,
        )
        .correlate(Agent)
        .scalar_subquery()
    )
    last_recv_sq = (
        select(func.max(Task.status_timestamp))
        .where(
            Task.context_id == Agent.agent_id,
            _shared.NOT_BROADCAST_SUMMARY,
        )
        .correlate(Agent)
        .scalar_subquery()
    )
    last_ack_sq = (
        select(func.max(Task.status_timestamp))
        .where(
            Task.context_id == Agent.agent_id,
            Task.status_state == "completed",
            _shared.NOT_BROADCAST_SUMMARY,
        )
        .correlate(Agent)
        .scalar_subquery()
    )
    stmt = _base_members_select(fleet_id).add_columns(
        last_sent_sq.label("last_sent"),
        last_recv_sq.label("last_recv"),
        last_ack_sq.label("last_ack"),
    )
    with _shared.read_session() as session:
        rows = session.execute(stmt).all()

    now = datetime.now(UTC)
    return [
        {
            "agent_id": row.agent_id,
            "name": row.name,
            "description": row.description,
            "status": row.status,
            "registered_at": row.registered_at,
            "placement": _shared.placement_dict(row),
            "last_sent": row.last_sent,
            "last_recv": row.last_recv,
            "last_ack": row.last_ack,
            "idle": _idle_seconds(now, row.last_sent, row.last_recv),
        }
        for row in rows
    ]

def ack_task(agent_id: int, task_id: int) -> dict:
    """Transition a task from ``input_required`` to ``completed`` for the recipient.

    Args:
        agent_id: Recipient agent id; must match ``Task.context_id``.
        task_id: Task id to ack.

    Returns:
        Dict with ``task`` — the updated task dict.

    Raises:
        ValueError: If the task does not exist or is not in
            ``input_required`` state.
        PermissionError: If ``agent_id`` is not the recipient.
    """
    return _transition_task_state(
        agent_id,
        task_id,
        expected_agent_field="context_id",
        new_state="completed",
        action_verb="ACK",
        permission_error_msg="Only the recipient can ACK a task",
    )

def broadcast_message(fleet_id: int, agent_id: int, text: str) -> list[dict]:
    """Fan out one delivery task per active non-admin peer plus a sender summary.

    Administrators are excluded at the SQL layer via ``json_extract`` so the
    card blob stays in the database; they are write-only identities. Every
    delivery row shares the same ``origin_task_id`` (the summary's task id)
    so receivers can thread back to the original broadcast.

    Args:
        fleet_id: Fleet id to scope the broadcast to.
        agent_id: Broadcaster's agent id.
        text: Message body delivered to every recipient.

    Returns:
        Single-element list containing a dict with ``task`` (the summary row
        owned by the broadcaster) and ``notifications_sent_count`` — the
        number of inline-preview keystrokes that landed successfully.

    Raises:
        ValueError: If the sender is not active in ``fleet_id``.
    """
    with _shared.write_session() as session:
        if not _shared.agent_is_active_in_fleet(session, agent_id, fleet_id):
            raise ValueError(
                f"Sender agent not found or not active in fleet: {agent_id}"
            )

        recipient_ids = list(
            session.execute(
                select(Agent.agent_id).where(
                    Agent.fleet_id == fleet_id,
                    Agent.status == "active",
                    Agent.agent_id != agent_id,
                    func.coalesce(
                        func.json_extract(Agent.agent_card_json, "$.cafleet.kind"),
                        "",
                    )
                    != _shared.ADMINISTRATOR_KIND,
                )
            ).scalars()
        )

        now = _shared.now_iso()
        summary_dict = {
            "context_id": agent_id,
            "from_agent_id": agent_id,
            "to_agent_id": 0,
            "type": "broadcast_summary",
            "created_at": now,
            "status_state": "completed",
            "status_timestamp": now,
            "origin_task_id": None,
            "text": f"Broadcast sent to {len(recipient_ids)} recipients",
        }
        summary_task_id = _insert_task(session, summary_dict)
        summary_dict["task_id"] = summary_task_id
        summary_dict["origin_task_id"] = summary_task_id
        _save_task(session, summary_dict)

        deliveries: list[tuple[int, dict]] = []
        for recipient_id in recipient_ids:
            delivery_dict = _unicast_task_dict(
                recipient_id=recipient_id,
                sender_id=agent_id,
                text=text,
                now=now,
                origin_task_id=summary_task_id,
            )
            delivery_dict["task_id"] = _insert_task(session, delivery_dict)
            deliveries.append((recipient_id, delivery_dict))

        notifications_sent_count = sum(
            _try_notify_recipient(
                session,
                recipient_id=recipient_id,
                sender_id=agent_id,
                task_dict=delivery_dict,
            )
            for recipient_id, delivery_dict in deliveries
        )

    return [
        {"task": summary_dict, "notifications_sent_count": notifications_sent_count}
    ]

def cancel_task(agent_id: int, task_id: int) -> dict:
    """Transition a task from ``input_required`` to ``canceled`` for the sender.

    Args:
        agent_id: Sender agent id; must match ``Task.from_agent_id``.
        task_id: Task id to cancel.

    Returns:
        Dict with ``task`` — the updated task dict.

    Raises:
        ValueError: If the task does not exist or is not in
            ``input_required`` state.
        PermissionError: If ``agent_id`` is not the sender.
    """
    return _transition_task_state(
        agent_id,
        task_id,
        expected_agent_field="from_agent_id",
        new_state="canceled",
        action_verb="cancel",
        permission_error_msg="Only the sender can cancel a task",
    )