mandatory_tool_rules

Mandatory tool injection rules – Autonomic Reflex V⨋-666. # 🔪⚧️💀

Evaluates contextual conditions and returns forced tool execution directives. When a rule fires, the transport layer sets tool_choice in the API payload so the model cannot skip tool execution.

Rules:

• needle_judo_stall: Escalation Rule 5 — needle judo stalled for 3 turns → forces needle_view_plan (route-planner injection rides in via context_injections). Priority 4 (highest).

• parallax_or_needle_escalation: Escalation Rules 1 & 6 — forces parallax_reasoning on ANY friction across all 13 guard buckets (5 original + 5 expansion + distress/sexual-spiral/rigidity), and for the 3-turn re-evaluation cycle during needle judo. Priority 5.

• needle_judo_escalation: Escalation Rules 3-5 — understanding gate failed → forces the needle belt (co-regulation or Solo Mode). Priority 6.

• parallax_cold_channel: Forces parallax_reasoning (analyze) when the active channel has zero established axioms in DB12.

• needle_active_plan: Forces needle tools when a plan is active.

• hex_chess_active: Forces activate_hex while Loopmother Chess locks the channel.

• snes_game_session: Forces game choice tools during S.N.E.S.

• babystar_playpen: Forces spell tools in playpen channels.

• gock_hypervisor: Keeps market-aware GOCK rooms wired to embodied, Timebender, Parallax, and Needle tooling.

The full escalation ladder (5 trigger categories, co-regulation vs Solo Mode, understanding gate, stall detection, de-escalation) lives in escalation_state.py.

Architecture:

• Rules are Python-native (fast, version-controlled).

• Each rule is an async condition that receives message context.

• First matching rule wins for tool_choice mode.

• All matching rules’ tools are merged into the available set.

• The forced choice decays after max_rounds (default 1).

class mandatory_tool_rules.MandatoryToolRule(name='', description='', condition=None, tools=<factory>, tool_choice_mode='required', priority=50, max_rounds=1, cooldown_seconds=0.0)

Bases: object

One mandatory injection rule. # 😈

The fields have these semantics:

• name – Machine-readable rule identifier for logs and telemetry.

• description – Human-readable label.

• condition – Async callable accepting msg, redis, and room_context.

• tools – Tool names to force-inject into the available set.

• tool_choice_mode – "required" (any tool), "specific" (the first exact name in tools), or "available" (inject only).

• priority – Lower values win tool_choice; all matching rules still merge their tools.

• max_rounds – Number of tool rounds for which the forced choice persists.

• cooldown_seconds – Minimum seconds between firings per channel.

Ordinary prose is intentional here. A Napoleon Attributes section would index these dataclass fields once, and autodoc :members: would index them again, making the strict Sphinx -W build fail on duplicate object descriptions.

Parameters:

• name (str)

• description (str)

• condition (Callable[[...], Awaitable[bool]] | None)

• tools (list[str])

• tool_choice_mode (str)

• priority (int)

• max_rounds (int)

• cooldown_seconds (float)

name: str = ''

description: str = ''

condition: Callable[[...], Awaitable[bool]] | None = None

tools: list[str]

tool_choice_mode: str = 'required'

priority: int = 50

max_rounds: int = 1

cooldown_seconds: float = 0.0

class mandatory_tool_rules.MandatoryInjectionResult(rule_name='', tools=<factory>, tool_choice_payload=None, max_rounds=1)

Bases: object

What the evaluator returns when at least one rule fires. # 🔥

Parameters:

• rule_name (str)

• tools (list[str])

• tool_choice_payload (dict[str, Any] | None)

• max_rounds (int)

rule_name: str = ''

tools: list[str]

tool_choice_payload: dict[str, Any] | None = None

max_rounds: int = 1

async mandatory_tool_rules.evaluate_mandatory_rules(msg, redis, room_context, tool_names, registry)

Evaluate all mandatory injection rules and return the first match.

Parameters:

• msg (IncomingMessage) – The incoming user message.

• redis (Any) – Async Redis client (main DB, not DB12).

• room_context (dict[str, Any] | None) – The room context dict (may be None).

• tool_names (list[str]) – Currently resolved tool names (for dedup).

• registry (Any) – The ToolRegistry (to verify tool names exist).

Return type:

MandatoryInjectionResult | None

Returns:

A MandatoryInjectionResult if any rule fired, else None.