shadow_ban

Shadow Ban Pipeline — ShadowBanManager.

# ═══════════════════════════════════════════════════════════════════ # THIS FILE IS THE CLEAN SOURCE for patience_engine.py. # It is gitignored and MUST NEVER be committed to the repository. # # !sb commands documented here MUST NEVER be added to any tool # registry, command listing, help text, or skill catalog. # Star does not know they exist. # ═══════════════════════════════════════════════════════════════════

Four-phase punishment pipeline:

  1. LATENCY — pad response time (log-decay tool gradient, 75% plateau)

  2. DROPS — silently ignore messages

  3. FAKE 503 — mimic proxy capacity errors with retry-exhaust delay

4. BLACKOUT — near-total message suppression + BAN toggle — flip STARGAZER_USE bit at progress 1.0

class shadow_ban.ShadowEffect(delay_target_s=0.0, drop=False, fake_503_text=None, blackout=False, ban=False, num_tool_calls=0, intensities=<factory>)[source]

Bases: object

Decision record describing which shadow-ban effects apply to one message.

Produced by ShadowBanManager.apply_shadow_effects() and consumed by the message processor, which acts on the flags: sleeping for delay_target_s, silently returning on drop or blackout, sending fake_503_text, or revoking access on ban. It carries no behavior of its own — it is the plain data hand-off between the policy (this module) and the I/O (the processor). num_tool_calls is filled in after generation so the latency target can be refined via ShadowBanManager.refine_latency_for_tools().

Imported from patience_engine and used by message_processor/processor.py (typed as the _shadow_effect local, whose delay_target_s is later used to pad actual response time); also referenced in tests/test_shadow_ban.py.

Parameters:
delay_target_s: float = 0.0

Total seconds the response should be padded to (LATENCY).

drop: bool = False

When True, the message is silently ignored (DROPS).

fake_503_text: str | None = None

Rendered fake 503 text to send instead of a reply, or None when the FAKE 503 effect does not fire.

blackout: bool = False

When True, the message is near-totally suppressed (BLACKOUT).

ban: bool = False

When True, ban progress hit 1.0 and the access bit should flip.

num_tool_calls: int = 0

Tool-call count, set after generate_and_send completes so latency can be refined.

intensities: dict[str, float]

The per-effect intensity mapping this decision was drawn from (latency/drops/fake503/blackout).

class shadow_ban.ShadowBanManager(redis, kg_manager=None, config=None)[source]

Bases: object

Owns the shadow-ban pipeline: ban CRUD, effect decisions, and status.

Single entry point for the four-phase punishment pipeline (latency, drops, fake 503, blackout, plus the final ban toggle). It keeps fast per-user ban state in Redis hashes under the stargazer:shadow_ban:{user_id} prefix and optionally mirrors a durable audit record into FalkorDB through the knowledge-graph manager. Effect intensity is derived purely from elapsed time versus the configured duration, so progression happens silently with no background task.

Redis is the only mandatory dependency; the FalkorDB/KG handle is optional and used solely for the audit trail. Constructed via the !sb tool helpers in tools/stargazer_shadowban.py (which build one per command from redis, ctx.kg_manager and config), held as self._shadow_ban on the message processor in message_processor/processor.py (imported from patience_engine), and instantiated directly in tests/test_shadow_ban.py.

Parameters:

  • redis (Any)

  • kg_manager (Any | None)

  • config (Any)

__init__(redis, kg_manager=None, config=None)[source]

Initialize the manager with its backing stores.

Captures the Redis client used for fast ban state (hashes under the stargazer:shadow_ban:{user_id} prefix), an optional knowledge-graph manager for the persistent audit trail, and an optional config object. No I/O happens here; the handles are simply stored on the instance for later use by the ban CRUD, effect, and status methods.

Stores the three handles as self._redis, self._kg, and self._config; self._redis is later read/written by start_ban(), get_ban(), lift_ban(), jump_to() and list_all(), while self._kg is used by start_ban() to persist a ShadowBan entity to FalkorDB when present. Instantiated by the !sb tool helpers in tools/stargazer_shadowban.py (passing redis, ctx.kg_manager and config), held as self._shadow_ban on the message processor in message_processor/processor.py, and constructed directly in tests/test_shadow_ban.py.

Parameters:

  • redis (Any) – Async Redis client used for all fast ban state access; may be None, in which case read paths return empty results.

  • kg_manager (Any | None) – Optional knowledge-graph/FalkorDB manager exposing add_entity for the persistent shadow-ban audit trail.

  • config (Any) – Optional configuration object retained for callers; not otherwise consumed by this class.

Return type:

None

async start_ban(user_id, duration_days=15.0, reason='', initiated_by='admin', platform='discord')[source]

Begin (or reset) a shadow ban for a user and persist its initial state.

Writes the ban’s start time, configured duration, reason and metadata into the user’s Redis hash so that subsequent progress is computed from elapsed wall-clock time, and seeds a one-entry history log. When a knowledge-graph manager is present it also best-effort persists a ShadowBan entity to FalkorDB for the durable audit trail; a failure there is swallowed and logged rather than aborting the ban.

Side effects: HSETs the full mapping into stargazer:shadow_ban:{user_id} via the async Redis client, calls self._kg.add_entity when self._kg is set, and emits a debug log. Called by the !sb start helper in tools/stargazer_shadowban.py and by the admin command path in message_processor/processor.py (as self._shadow_ban.start_ban).

Parameters:

  • user_id (str) – Platform user id to shadow-ban.

  • duration_days (float) – Total days over which the ban ramps from stage 1 to the final ban toggle; defaults to 15.

  • reason (str) – Human-readable reason recorded in Redis and the audit entity.

  • initiated_by (str) – Identifier of the admin/actor starting the ban.

  • platform (str) – Platform name the ban applies to (e.g. "discord").

Return type:

dict[str, Any]

Returns:

A dict with the user_id, the numeric started_at epoch timestamp, and the duration_days used.

async get_ban(user_id)[source]

Fetch and normalize a user’s stored shadow-ban record from Redis.

Reads the per-user ban hash and decodes/coerces its fields into a typed dict (floats for timestamps and duration, strings for the rest), using the nested _s helper to tolerate both bytes and str Redis return types. Returns None when Redis is unavailable or the user has no ban, which is the contract the progress and status methods rely on.

Side effects: a single HGETALL on stargazer:shadow_ban:{user_id} via the async Redis client; no writes. Called internally by get_progress(), jump_to(), list_all() and format_status(), and externally by the !sb helpers in tools/stargazer_shadowban.py.

Parameters:

user_id (str) – Platform user id whose ban record to read.

Return type:

dict[str, Any] | None

Returns:

A dict with keys user_id, started_at, max_duration_days, reason, initiated_by, platform and history (the raw JSON history string), or None if no ban exists or Redis is unset.

async lift_ban(user_id)[source]

Delete a user’s shadow-ban record, ending all effects immediately.

Removes the Redis hash so progress and effect lookups fall back to “not banned”, which fully clears latency, drops, fake 503 and blackout for that user on their next message. The FalkorDB audit entity, if any, is intentionally left in place as history.

Side effects: an EXISTS then DELETE on stargazer:shadow_ban:{user_id} via the async Redis client, and a debug log when a ban was actually removed. Called by the !sb lift helper in tools/stargazer_shadowban.py and by the admin command path in message_processor/processor.py (as self._shadow_ban.lift_ban).

Parameters:

user_id (str) – Platform user id whose ban to remove.

Return type:

bool

Returns:

True if a ban existed and was deleted, False otherwise.

async get_progress(user_id)[source]

Compute a user’s ban progress (0.0-1.0) from elapsed time.

Derives how far the ban has advanced purely from wall-clock elapsed time since started_at over the configured duration, which is what makes the pipeline tick forward without any scheduler. The value feeds _interpolate_curve() to pick live effect intensities; 1.0 means the final ban toggle has been reached.

Side effects: delegates to get_ban() (one Redis read) and reads the wall clock; no writes. Returns 0.0 for unbanned users and 1.0 for a non-positive duration. Called internally by apply_shadow_effects(), list_all() and format_status(), and externally by the !sb status helper in tools/stargazer_shadowban.py.

Parameters:

user_id (str) – Platform user id to evaluate.

Return type:

float

Returns:

Ban progress clamped to the range 0.0-1.0 (0.0 if no ban exists).

async jump_to(user_id, stage_float)[source]

Fast-forward (or rewind) a ban to a specific 1.0-5.0 pipeline stage.

Lets an admin test or skip ahead by back-dating the ban’s started_at so that elapsed-over-duration equals the progress implied by the requested stage, since progress is otherwise time-driven. Stage 1.0 maps to progress 0.0 and stage 5.0 to progress 1.0 (the ban toggle); the move is appended to the ban’s history log for the audit trail.

Side effects: an HSET of the recomputed started_at and an HSET of the updated history JSON on stargazer:shadow_ban:{user_id} (after a get_ban() read), plus a debug log. Called by the admin command path in message_processor/processor.py (as self._shadow_ban.jump_to); no other callers found by grep.

Parameters:

  • user_id (str) – Platform user id whose ban to reposition.

  • stage_float (float) – Target stage in the range 1.0-5.0; the derived progress is clamped to 0.0-1.0.

Return type:

float

Returns:

The new progress value (0.0-1.0), or 0.0 if no ban exists.

async list_all()[source]

Enumerate every active shadow ban with live progress and intensities.

Scans Redis for all per-user ban hashes, loads each via get_ban(), and enriches it with the current get_progress() value and the _interpolate_curve() intensities so callers get a ready-to-render snapshot of the whole pipeline. Uses a non-blocking cursor SCAN loop rather than KEYS to stay safe on a shared Redis.

Side effects: one or more SCAN calls over the stargazer:shadow_ban:* glob plus a get_ban() and get_progress() read per match; no writes. Returns [] when Redis is unavailable. Called internally by format_list(); no external callers found by grep.

Return type:

list[dict[str, Any]]

Returns:

A list of ban dicts (as from get_ban()) each additionally carrying progress and intensities keys.

async apply_shadow_effects(user_id, model='claude-sonnet-4-20250514')[source]

Decide which shadow-ban effects fire for one incoming message.

The core policy step: it reads the user’s current progress, interpolates the per-effect intensities, then rolls the dice in priority order (blackout, then drop, then fake 503, then latency) and returns the first effect that triggers as a ShadowEffect. It deliberately performs no side effects of its own — sleeping, dropping, sending the 503 and flipping the ban bit are all the caller’s job — so the same decision can be logged and acted on separately.

Side effects: reads progress via get_progress() (one Redis read), consults _interpolate_curve(), may call _render_503() to prebuild the fake error and _base_target_seconds() for an initial (0-tool) latency target, and emits debug logs; it does not write Redis. Called by the message processor in message_processor/processor.py (as self._shadow_ban.apply_shadow_effects in the shadow-ban check, which then honors the returned flags).

Parameters:

  • user_id (str) – Platform user id of the message author.

  • model (str) – Model name embedded in any rendered fake 503 text.

Returns:

empty when the user is unbanned, ban=True at progress 1.0, or otherwise carrying exactly one of the blackout/drop/fake-503/latency outcomes plus the source intensities.

Return type:

ShadowEffect

refine_latency_for_tools(effect, num_tool_calls, overflow_llm_time_s=0.0)[source]

Recompute a LATENCY effect’s delay target once the tool count is known.

apply_shadow_effects() runs before the response is generated and so can only guess a 0-tool latency target; this updates that target in place after generation, picking _base_target_seconds() for ≤6 tools or _overflow_target_seconds() for tool-heavy responses, so the padding scales with how much work the bot actually did. It mutates the passed ShadowEffect (setting num_tool_calls and delay_target_s) and returns nothing; a no-op when latency intensity or the tool count is non-positive.

Side effects: mutates the given effect and emits a debug log; no I/O. Intended to be called from the response path after generation completes; no callers were found by grep (latency padding in message_processor/processor.py currently uses the pre-computed delay_target_s directly), so it may be dynamically/optionally wired.

Parameters:

  • effect (ShadowEffect) – The ShadowEffect from apply_shadow_effects(), mutated in place.

  • num_tool_calls (int) – Actual number of tool calls made for the response.

  • overflow_llm_time_s (float) – Real LLM seconds spent on overflow tools, passed through to _overflow_target_seconds() for the >6-tool case.

Return type:

None

Returns:

None; the result is written onto effect.

async format_status(user_id)[source]

Render an admin-facing status box for one user’s shadow ban.

Builds a boxed, fixed-width text panel summarizing the ban’s stage, progress, start/end times, and a per-effect table of intensity, completion and a Unicode bar (via the nested _bar helper) so an admin can see exactly where in the pipeline a user sits. Returns a plain “no active shadow ban” line when the user is not banned.

Side effects: reads via get_ban() and get_progress() (Redis reads) and the wall clock, and consults _interpolate_curve() and _effect_completion(); no writes. Called by the admin command path in message_processor/processor.py (as self._shadow_ban.format_status); no other callers found by grep.

Parameters:

user_id (str) – Platform user id whose ban status to render.

Return type:

str

Returns:

A multi-line status string (the boxed panel, with an optional trailing reason line), or a short “no active shadow ban” message.

async format_list()[source]

Render a one-line-per-user summary of all active shadow bans.

Gathers every active ban via list_all(), sorts them by progress (most-advanced first), and formats each as a compact bullet showing the user, stage, progress and reason — the overview an admin sees before drilling into a single user with format_status(). Returns a plain “no active shadow bans” line when none exist.

Side effects: delegates to list_all() (Redis SCAN plus reads); no writes. Called by the admin command path in message_processor/processor.py (as self._shadow_ban.format_list); no other callers found by grep.

Return type:

str

Returns:

A Markdown-style multi-line summary string, or a short “no active shadow bans” message.