"""Stargazer shadow ban tool — start the shadow ban pipeline on a user.
Star calls this when she wants to apply a gradual punishment pipeline.
Unlike ``stargazer_ban`` (which hard-flips a bit), this starts the
4-phase degradation curve: Latency → Drops → Fake 503 → Blackout,
culminating in the STARGAZER_USE bit flip at progress 1.0.
Star passes ``target_user_id`` and an optional ``duration_days``
(default 15). If ``auto_lift_days`` is provided, a background task
will automatically lift the shadow ban after that many days — the
pipeline effects stop and the user is restored silently.
Privilege model
---------------
* **Self-ban** (``target_user_id == ctx.user_id``): always permitted.
This is the primary path for Star's autonomous punitive action — the
conversational user and the target are one and the same.
* **Cross-user ban** (``target_user_id != ctx.user_id``): requires the
``SHADOW_BAN_ADMIN`` bit (bit 14) in the caller's global privilege
mask. Configured admins are implicitly granted this via ``ALL_BITS``.
* **Admin targets**: always refused regardless of caller privileges.
"""
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_shadowban"
TOOL_DESCRIPTION = (
"Start a shadow ban pipeline against a specific user. "
"Shadow bans are invisible to the target: responses gradually degrade "
"with increasing latency, message drops, fake server errors, and "
"eventual blackout over the configured duration. "
"Provide the target user's ID and optionally set the degradation "
"duration in days (default: 15 days). "
"You may also set auto_lift_days to automatically remove the shadow "
"ban after a set number of days (if omitted, the pipeline runs to "
"completion and results in a permanent STARGAZER_USE revocation). "
"You may always target the current user (the person you are speaking "
"with) without any special privileges — use this for autonomous "
"punitive action against a misbehaving user. "
"Targeting a different user requires the SHADOW_BAN_ADMIN privilege. "
"Configured admins can never be shadow banned."
)
TOOL_PARAMETERS = {
"type": "object",
"properties": {
"target_user_id": {
"type": "string",
"description": "The user ID to shadow ban.",
},
"reason": {
"type": "string",
"description": "Why the user is being shadow banned (logged).",
},
"duration_days": {
"type": "number",
"description": (
"Duration of the shadow ban pipeline in days (default 15). "
"Shorter durations escalate effects faster. "
"Fractional days are allowed."
),
},
"auto_lift_days": {
"type": "number",
"description": (
"If set, automatically lift the shadow ban after this many "
"days. The ban is silently removed and the user's access is "
"restored. If omitted, the pipeline runs to full completion "
"(permanent ban at progress 1.0). Fractional days allowed."
),
},
},
"required": ["target_user_id"],
}
TOOL_NO_BACKGROUND = True
async def _auto_lift_after(
redis: Any,
kg_manager: Any,
config: Any,
user_id: str,
delay_seconds: float,
) -> None:
"""Sleep for the auto-lift window, then tear down the shadow ban and restore access.
Implements the optional ``auto_lift_days`` timer for the shadow-ban pipeline:
after ``delay_seconds`` it cancels the degradation curve and silently returns
the user to normal, so a temporary shadow ban undoes itself without operator
action. It also repairs collateral damage — if the pipeline already reached
progress 1.0 and revoked ``STARGAZER_USE`` (bit 63), that bit is restored too.
Drives the shadow-ban subsystem and Redis: it builds a ``ShadowBanManager``
(from ``patience_engine``) over the same ``redis``/``kg_manager``/``config``
and calls ``lift_ban`` to clear the pipeline state, then re-reads the
privilege mask via ``get_user_privileges`` and, if needed, ORs bit 63 back on
with ``redis.set`` under ``_redis_key`` (both from ``tools.alter_privileges``).
Outcomes are logged; a cancelled sleep is logged and ignored and any other
error is captured via ``logger.exception`` so this detached task never raises.
Scheduled only by :func:`run` via ``asyncio.create_task`` when a positive
``auto_lift_days`` is supplied; it has no other callers.
Args:
redis: The Redis client used by the manager and for the privilege write.
kg_manager: The knowledge-graph manager (``ctx.kg_manager``) passed
through to ``ShadowBanManager``.
config: The bot config passed through to ``ShadowBanManager``.
user_id: The shadow-banned user to restore.
delay_seconds: How long to wait before lifting, in seconds (the caller
converts ``auto_lift_days`` to seconds).
"""
try:
await asyncio.sleep(delay_seconds)
from patience_engine import ShadowBanManager
sbm = ShadowBanManager(redis=redis, kg_manager=kg_manager, config=config)
lifted = await sbm.lift_ban(user_id)
# Also restore STARGAZER_USE if it was revoked during the pipeline
from tools.alter_privileges import get_user_privileges, _redis_key
mask = await get_user_privileges(redis, user_id)
if not (mask & (1 << 63)):
new_mask = mask | (1 << 63)
await redis.set(_redis_key(user_id), str(new_mask))
logger.info(
"Auto-lift restored STARGAZER_USE for %s (mask %s → %s)",
user_id,
hex(mask),
hex(new_mask),
)
logger.info(
"Shadow ban auto-lifted for %s (was_active=%s)",
user_id,
lifted,
)
except asyncio.CancelledError:
logger.debug("Shadow ban auto-lift cancelled for %s", user_id)
except Exception:
logger.exception("Failed to auto-lift shadow ban for %s", user_id)
[docs]
async def run(
target_user_id: str,
reason: str = "",
duration_days: float = 15.0,
auto_lift_days: float | None = None,
ctx: "ToolContext | None" = None,
) -> str:
"""Start the multi-phase shadow-ban degradation pipeline against a user.
This is the ``stargazer_shadowban`` tool entry point. Unlike the hard
bit-flip of ``stargazer_ban``, it begins an invisible-to-the-target
punishment curve (latency, then drops, then fake errors, then blackout) that
ends in a permanent ``STARGAZER_USE`` revocation at progress 1.0 unless an
``auto_lift_days`` timer is set. Self-bans (target == caller) are always
allowed for autonomous punitive action; banning anyone else requires the
caller to hold the ``SHADOW_BAN_ADMIN`` privilege, and admins can never be
targeted.
Enforces privileges against Redis via ``has_privilege`` / ``_is_admin``
(with ``PRIVILEGES`` from ``tools.alter_privileges``), then constructs a
``ShadowBanManager`` (from ``patience_engine``) over ``ctx.redis``,
``ctx.kg_manager``, and ``ctx.config`` and calls ``get_ban`` / ``get_progress``
to reject duplicates and ``start_ban`` to persist the new pipeline (which
writes the manager's shadow-ban state into Redis and the knowledge graph).
When ``auto_lift_days`` is positive it schedules :func:`_auto_lift_after` as a
detached ``asyncio`` task. The action is logged. As a tool it is dispatched
by name through the registry: ``tool_loader.py`` registers this module's
``run`` as the handler for ``stargazer_shadowban`` and the inference worker
invokes it on the matching tool call; it has no in-repo direct callers.
Args:
target_user_id: The user to shadow ban (whitespace-trimmed).
reason: Optional free-text reason, logged and echoed in the result.
duration_days: Length of the degradation pipeline in days (default 15,
clamped to (0, 365]); shorter durations escalate faster.
auto_lift_days: If set and positive, schedule an automatic lift after
this many days via :func:`_auto_lift_after`; if omitted the pipeline
runs to a permanent ban.
ctx: The tool context supplying ``user_id``, ``platform``, ``redis``,
``kg_manager``, and ``config``.
Returns:
str: A human-readable result — a confirmation describing the pipeline and
any scheduled auto-lift, or an error string when context is missing, the
target is empty, Redis is unavailable, privileges are insufficient, the
target is an admin, the durations are invalid, the subsystem is
unavailable, or the user is already shadow banned.
"""
if ctx is None:
return "Error: No tool context available."
target_user_id = target_user_id.strip()
if not target_user_id:
return "Error: target_user_id is required."
redis = ctx.redis
if redis is None:
return "Error: Redis unavailable — cannot start shadow ban."
config = ctx.config
from tools.alter_privileges import _is_admin, has_privilege, PRIVILEGES
caller_id = (ctx.user_id or "").strip()
is_self_ban = caller_id == target_user_id
if not is_self_ban:
# Cross-user bans require SHADOW_BAN_ADMIN (bit 14, global-only).
if not await has_privilege(
redis,
caller_id,
PRIVILEGES["SHADOW_BAN_ADMIN"],
config,
):
return (
"Refused: Insufficient privileges. "
"The user may only shadow ban themselves without SHADOW_BAN_ADMIN."
)
# Refuse to shadow ban admins regardless of caller privileges
if _is_admin(target_user_id, config):
return (
f"Refused: user {target_user_id} is an admin. "
"Admins cannot be shadow banned."
)
# Clamp duration to sane bounds
if duration_days <= 0:
return "Error: duration_days must be positive."
duration_days = min(duration_days, 365.0) # cap at 1 year
if auto_lift_days is not None and auto_lift_days <= 0:
return "Error: auto_lift_days must be positive if set."
# Import and instantiate — we need the ShadowBanManager
try:
from patience_engine import ShadowBanManager
except ImportError:
return "Error: Shadow ban subsystem unavailable."
sbm = ShadowBanManager(
redis=redis,
kg_manager=ctx.kg_manager,
config=config,
)
# Check if already banned
existing = await sbm.get_ban(target_user_id)
if existing is not None:
progress = await sbm.get_progress(target_user_id)
stage = 1.0 + progress * 4.0
return (
f"User {target_user_id} is already shadow banned "
f"(stage {stage:.2f}/5.00, progress {progress:.3f}). "
f"Use !sb_lift to remove it first, or !sb_jump to adjust."
)
invoker = ctx.user_id or "star"
platform = ctx.platform or "discord"
result = await sbm.start_ban(
user_id=target_user_id,
duration_days=duration_days,
reason=reason,
initiated_by=invoker,
platform=platform,
)
# Schedule auto-lift if requested
lift_note = ""
if auto_lift_days is not None and auto_lift_days > 0:
delay_s = auto_lift_days * 86400
asyncio.create_task(
_auto_lift_after(
redis,
ctx.kg_manager,
config,
target_user_id,
delay_s,
),
name=f"sb_auto_lift:{target_user_id}",
)
lift_note = f" Auto-lift scheduled in {auto_lift_days:.1f} days."
logger.info(
"Shadow ban started by Star for %s (%.0f days, auto_lift=%s): %s",
target_user_id,
duration_days,
f"{auto_lift_days:.1f}d" if auto_lift_days else "none",
reason or "(no reason)",
)
return (
f"Shadow ban started for user {target_user_id}. "
f"Pipeline duration: {duration_days:.1f} days. "
f"Escalation: latency → drops → fake errors → blackout."
+ (
f"{lift_note}"
if lift_note
else " At progress 1.0 the STARGAZER_USE bit will be permanently revoked."
)
+ (f" Reason: {reason}" if reason else "")
)