patience_engine

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 patience_engine.ShadowBanManager(redis, kg_manager=None, config=None)

Bases: object

Manages the shadow ban pipeline.

Uses Redis for fast state access and FalkorDB for persistent audit trail and documentation storage.

Parameters:

• redis (Any)

• kg_manager (Any | None)

• config (Any)

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

Determine which shadow effects apply for this message.

Returns a ShadowEffect with the decision. The caller is responsible for acting on it (sleeping, dropping, sending 503).

Return type:

ShadowEffect

Parameters:

• user_id (str)

• model (str)

async format_list()

Format a summary of all active shadow bans.

Return type:

str

async format_status(user_id)

Format status box for a single user’s shadow ban.

Return type:

str

Parameters:

user_id (str)

async get_ban(user_id)

Get shadow ban data for a user, or None if not banned.

Return type:

dict[str, Any] | None

Parameters:

user_id (str)

async get_progress(user_id)

Get current progress (0.0 - 1.0) for a shadow-banned user.

Return type:

float

Parameters:

user_id (str)

async jump_to(user_id, stage_float)

Jump to a specific stage (1.0-5.0). Returns new progress.

Return type:

float

Parameters:

• user_id (str)

• stage_float (float)

async lift_ban(user_id)

Remove a shadow ban. Returns True if one existed.

Return type:

bool

Parameters:

user_id (str)

async list_all()

List all active shadow bans.

Return type:

list[dict[str, Any]]

refine_latency_for_tools(effect, num_tool_calls, overflow_llm_time_s=0.0)

Refine the latency target after knowing the tool call count.

Return type:

None

Parameters:

• effect (ShadowEffect)

• num_tool_calls (int)

• overflow_llm_time_s (float)

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

Start a shadow ban for the given user.

Return type:

dict[str, Any]

Parameters:

• user_id (str)

• duration_days (float)

• reason (str)

• initiated_by (str)

• platform (str)

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

Bases: object

Result of applying shadow ban effects to a message.

Parameters:

• delay_target_s (float)

• drop (bool)

• fake_503_text (str | None)

• blackout (bool)

• ban (bool)

• num_tool_calls (int)

• intensities (dict[str, float])

ban: bool = False

blackout: bool = False

delay_target_s: float = 0.0

drop: bool = False

fake_503_text: str | None = None

num_tool_calls: int = 0

intensities: dict[str, float]