"""Stargazer ban tool — revoke STARGAZER_USE for the interacting user.
Star calls this when she decides a user should lose access. The
``user_id`` parameter is exposed to the LLM for UX but is *always*
overridden by ``ctx.user_id`` to prevent the model from being tricked
into banning someone other than the person she's talking to.
An optional ``duration_days`` makes the ban temporary: a background
asyncio task restores the bit after the timer expires. Without it
the ban is permanent until an admin manually grants the bit back.
"""
from __future__ import annotations
import asyncio
import logging
from typing import Any, TYPE_CHECKING
if TYPE_CHECKING:
from tool_context import ToolContext
logger = logging.getLogger(__name__)
TOOL_NAME = "stargazer_ban"
TOOL_DESCRIPTION = (
"Revoke access to Stargazer for the user you are currently speaking to. "
"Flips the STARGAZER_USE privilege bit off in the GLOBAL mask only. "
"Note: This does NOT clear guild or channel-level overrides. For a "
"complete ban across all scopes, use the stargazer_full_revoke tool. "
"You may optionally set a duration in days for a temporary ban; "
"without it the ban is permanent. "
"IMPORTANT: This tool ALWAYS targets the user you are responding to, "
"regardless of what user_id you pass — it cannot be aimed at anyone else."
)
TOOL_PARAMETERS = {
"type": "object",
"properties": {
"user_id": {
"type": "string",
"description": (
"The user to ban. NOTE: this is cosmetic only — the tool "
"always targets the user you are currently talking to."
),
},
"reason": {
"type": "string",
"description": "Why the user is being banned (logged).",
},
"duration_days": {
"type": "number",
"description": (
"Optional ban duration in days. If omitted the ban is "
"permanent. Fractional days are allowed (e.g. 0.5 = 12 hours)."
),
},
},
"required": ["user_id"],
}
TOOL_NO_BACKGROUND = True
# Bit 63 — STARGAZER_USE
_STARGAZER_USE_BIT = 63
async def _restore_after(redis: Any, user_id: str, delay_seconds: float) -> None:
"""Sleep for the ban duration, then re-grant the STARGAZER_USE bit.
Implements the temporary-ban timer: after ``delay_seconds`` it re-reads the
user's current privilege mask and ORs bit 63 (``STARGAZER_USE``) back on, so
the ban lifts itself without an admin having to intervene. It re-reads rather
than caching the old mask so any privilege changes made during the ban window
are preserved.
Touches Redis directly: reads the mask via ``get_user_privileges`` and writes
the updated value back with ``redis.set`` under the per-user privilege key
from ``_redis_key`` (both imported lazily from ``tools.alter_privileges``).
A cancelled sleep (task torn down before expiry) is logged and swallowed;
any other failure is logged via ``logger.exception`` and never propagated,
since this runs detached. Scheduled only by :func:`run` via
``asyncio.create_task`` when a positive ``duration_days`` is supplied; it is
not invoked anywhere else.
Args:
redis: The Redis client (``ctx.redis``) used to read and write the mask.
user_id: The user whose ``STARGAZER_USE`` bit is restored.
delay_seconds: How long to sleep before restoring, in seconds (the
caller converts ``duration_days`` to seconds).
"""
try:
await asyncio.sleep(delay_seconds)
from tools.alter_privileges import get_user_privileges, _redis_key
mask = await get_user_privileges(redis, user_id)
new_mask = mask | (1 << _STARGAZER_USE_BIT)
await redis.set(_redis_key(user_id), str(new_mask))
logger.info(
"Temp ban expired for %s — STARGAZER_USE restored (mask=%s)",
user_id,
hex(new_mask),
)
except asyncio.CancelledError:
logger.debug("Temp ban restore cancelled for %s", user_id)
except Exception:
logger.exception("Failed to restore STARGAZER_USE for %s", user_id)
[docs]
async def run(
user_id: str,
reason: str = "",
duration_days: float | None = None,
ctx: "ToolContext | None" = None,
) -> str:
"""Revoke the global ``STARGAZER_USE`` privilege bit for the interacting user.
This is the ``stargazer_ban`` tool entry point. It is the LLM's lever for
cutting off a user it is conversing with: it clears bit 63 in the user's
GLOBAL privilege mask so the bot stops serving them. The model-supplied
``user_id`` is intentionally ignored in favor of ``ctx.user_id`` so the ban
can never be redirected at a third party, and admins are refused outright.
Reads and writes Redis directly: it fetches the current mask via
``get_user_privileges`` and persists the bit-cleared mask with ``redis.set``
under ``_redis_key`` (both from ``tools.alter_privileges``), and uses
``_is_admin`` from the same module to enforce the admin exemption. When a
positive ``duration_days`` is given it schedules :func:`_restore_after` as a
detached ``asyncio`` task to auto-lift the ban; otherwise the ban is
permanent. All actions are logged. As a tool, it is dispatched by name
through the tool registry: ``tool_loader.py`` registers this module's ``run``
as the handler for ``stargazer_ban`` and the inference worker invokes it when
the LLM emits that tool call; it has no in-repo direct callers.
Args:
user_id: Cosmetic only — the user the model thinks it is banning. The
real target is always ``ctx.user_id``.
reason: Optional free-text reason, included in the log line and the
returned confirmation.
duration_days: Optional ban length in days (fractional allowed). If set
and positive, the ban auto-expires via :func:`_restore_after`;
otherwise it is permanent.
ctx: The tool context supplying ``user_id``, ``redis``, and ``config``.
Returns:
str: A human-readable result — a confirmation with the old/new mask, or
an error string when context, ``user_id``, or Redis is missing, when the
target is an admin, or when the user is already banned.
"""
# Hard override — always target the real interacting user
if ctx is None:
return "Error: No tool context available — cannot determine user."
real_user_id = ctx.user_id
if not real_user_id:
return "Error: No user_id in context."
redis = ctx.redis
if redis is None:
return "Error: Redis unavailable — cannot modify privileges."
from tools.alter_privileges import get_user_privileges, _redis_key, _is_admin
# Safety: refuse to ban admins
config = ctx.config
if _is_admin(real_user_id, config):
return (
f"Refused: user {real_user_id} is an admin. "
"Admin privileges cannot be revoked via this tool."
)
mask = await get_user_privileges(redis, real_user_id)
if not (mask & (1 << _STARGAZER_USE_BIT)):
return f"User {real_user_id} is already banned (STARGAZER_USE not set)."
new_mask = mask & ~(1 << _STARGAZER_USE_BIT)
await redis.set(_redis_key(real_user_id), str(new_mask))
duration_note = ""
if duration_days is not None and duration_days > 0:
delay_s = duration_days * 86400
asyncio.create_task(
_restore_after(redis, real_user_id, delay_s),
name=f"ban_expire:{real_user_id}",
)
duration_note = f" (temporary: {duration_days:.2f} days)"
reason_note = f" Reason: {reason}" if reason else ""
logger.info(
"STARGAZER_USE revoked for %s (mask %s → %s)%s%s",
real_user_id,
hex(mask),
hex(new_mask),
duration_note,
reason_note,
)
return (
f"Banned user {real_user_id}{duration_note}. "
f"STARGAZER_USE bit cleared (mask {hex(mask)} → {hex(new_mask)}).{reason_note}"
)