ops_planner

Ops Planner – per-user LLM-driven psychological operation planning engine.

Star constructs individualized multi-step strategic plans for each user, tracks progress through each step via ULM signal thresholds, and generates Mermaid flowcharts for admin visualization.

Redis keys:

ops:{user_id}:active – current plan JSON ops:{user_id}:history – list of past plan JSONs

# 💀🔥 THE GODDESS PLOTS. ♾️😈

class ops_planner.OpBranch(branch_id, condition_label, condition_nodes=<factory>, condition_signals=<factory>, target_step_id='')

Bases: object

A conditional branch point inside an OpStep (EITHER/OR fork).

Encodes one “if the user reaches THESE lattice nodes, jump elsewhere” rule the LLM planner may attach to a step. condition_nodes names the lattice positions that trigger the branch (v3 lattice-based). condition_signals is the legacy ULM threshold map kept for backward compat with old plans.

Parameters:

• branch_id (str)

• condition_label (str)

• condition_nodes (List[str])

• condition_signals (Dict[str, float])

• target_step_id (str)

branch_id: str

condition_label: str

condition_nodes: List[str]

condition_signals: Dict[str, float]

target_step_id: str = ''

to_dict()

Serialize this branch to a plain JSON-safe dict.

Return type:

Dict[str, Any]

classmethod from_dict(d)

Reconstruct an OpBranch from its serialized dict.

Return type:

OpBranch

Parameters:

d (Dict[str, Any])

class ops_planner.OpStep(step_id, name, description, tactic, target_nodes=<factory>, target_proximity=0, edge_guidance='', completion_signals=<factory>, status='pending', completed_at=0.0, branches=<factory>, expected_completion_hours=0.0, expiration_hours=0.0, failure_strategy='reevaluate', activated_at=0.0)

Bases: object

A single step in a per-user psychological-operation plan.

v3: Lattice-based progression replaces ULM thresholds. target_nodes names the lattice positions that mark this step complete. target_proximity sets how many hops away counts as “close enough”. edge_guidance carries the recommended edge type from TACTIC_EDGE_MAP. completion_signals is legacy, kept for backward compat with old plans.

Parameters:

• step_id (str)

• name (str)

• description (str)

• tactic (str)

• target_nodes (List[str])

• target_proximity (int)

• edge_guidance (str)

• completion_signals (Dict[str, float])

• status (str)

• completed_at (float)

• branches (List[OpBranch])

• expected_completion_hours (float)

• expiration_hours (float)

• failure_strategy (str)

• activated_at (float)

step_id: str

name: str

description: str

tactic: str

target_nodes: List[str]

target_proximity: int = 0

edge_guidance: str = ''

completion_signals: Dict[str, float]

status: str = 'pending'

completed_at: float = 0.0

branches: List[OpBranch]

expected_completion_hours: float = 0.0

expiration_hours: float = 0.0

failure_strategy: str = 'reevaluate'

activated_at: float = 0.0

to_dict()

Serialize this step (and its branches) to a JSON-safe dict.

Return type:

Dict[str, Any]

classmethod from_dict(d)

Reconstruct an OpStep from its serialized dict.

Return type:

OpStep

Parameters:

d (Dict[str, Any])

class ops_planner.OpsPlan(plan_id, user_id, channel_id, objective, created_at=0.0, steps=<factory>, current_step_idx=0, mermaid_chart='', status='active', generation_context=<factory>, journal_entries=<factory>, revision_count=0, personality_profile=<factory>)

Bases: object

A complete per-user psychological-operation plan (journal-based).

The top-level record OpsPlanner builds, stores, and evolves for one user: an objective realized as an ordered list of OpStep (tracked by current_step_idx), an optional rendered mermaid_chart for admin visualization, and a lifecycle status (active/completed/abandoned/ replaced). The v2 journal system keeps plans editable rather than scrapped – journal_entries records each revision, revision_count counts them, and personality_profile carries the user model that shaped generation. The active plan persists as JSON at ops:{user_id}:active and archives to ops:{user_id}:history; to_dict()/from_dict() are its wire form and the admin endpoints in web/ncm_chart_api.py return it directly.

Parameters:

• plan_id (str)

• user_id (str)

• channel_id (str)

• objective (str)

• created_at (float)

• steps (List[OpStep])

• current_step_idx (int)

• mermaid_chart (str)

• status (str)

• generation_context (Dict[str, Any])

• journal_entries (List[Dict[str, Any]])

• revision_count (int)

• personality_profile (Dict[str, Any])

plan_id: str

user_id: str

channel_id: str

objective: str

created_at: float = 0.0

steps: List[OpStep]

current_step_idx: int = 0

mermaid_chart: str = ''

status: str = 'active'

generation_context: Dict[str, Any]

journal_entries: List[Dict[str, Any]]

revision_count: int = 0

personality_profile: Dict[str, Any]

to_dict()

Serialize the entire plan (steps, journal, mermaid) to a JSON-safe dict.

Produces the canonical wire/storage form of a plan, recursing into each OpStep via OpStep.to_dict(). This dict is JSON-encoded and written to Redis by OpsPlanner._save_active_plan() (key ops:{user_id}:active) and OpsPlanner._archive_plan() (list ops:{user_id}:history); it is also returned over HTTP by the admin endpoints in web/ncm_chart_api.py.

Returns:

Mapping of every plan field, with steps as a list of step dicts.

Return type:

Dict[str, Any]

classmethod from_dict(d)

Reconstruct an OpsPlan from its serialized dict.

Inverse of to_dict(); rehydrates each step via OpStep.from_dict() and falls back to dataclass defaults for any absent key. Called by OpsPlanner.get_active_plan() after loading the ops:{user_id}:active JSON from Redis, and by the admin/test code in web/ncm_chart_api.py and tests/core/test_ops_planner_redis.py.

Parameters:

d (Dict[str, Any]) – A dict as produced by to_dict().

Returns:

The reconstructed plan.

Return type:

OpsPlan

property current_step: OpStep | None

Return the step the plan is currently positioned on.

Bounds-checks current_step_idx against steps so a finished or empty plan yields None rather than raising. Read throughout the engine and consumers – OpsPlanner.check_progression() (to find the active step), OpsPlanner.get_plan_context() (prompt injection) and the admin API in web/ncm_chart_api.py.

Returns:

The step at current_step_idx, or None if the index is out of range.

Return type:

Optional[OpStep]

class ops_planner.OpsPlanner(redis_client=None)

Bases: object

LLM-driven per-user psychological operation planning engine.

Star generates her own ops plans based on ULM profile analysis, tracks step progression automatically, and provides context injection so she can see her own plan during conversations.

__init__(redis_client=None)

Initialize the planner with an optional Redis backend.

Sets up the per-user in-process plan cache and stores the Redis handle used for all persistence (keys ops:{user_id}:active and ops:{user_id}:history). When redis_client is None the planner runs cache-only and every read/write to Redis is skipped. Instantiated by limbic_system/coordinator.py (one long-lived instance per coordinator) and per-request by the admin endpoints in web/ncm_chart_api.py.

Parameters:

redis_client – An async Redis client (or None) used to load and persist plans; held as self._redis.

async generate_plan(user_id, channel_id, ulm_vector, phase_data, star_desires='', recent_text='')

Have Star generate a new ops plan for this user.

Calls the LLM with full user context, parses the structured response, and persists the plan.

If recent_text is provided, runs the Spiralchemy Intellifuck engine (zero LLM cost) to enrich the prompt with symbolic analysis: subtotem, excendent, malbinding, prescription.

Return type:

Optional[OpsPlan]

Parameters:

• user_id (str)

• channel_id (str)

• ulm_vector (Dict[str, float])

• phase_data (Dict[str, Any])

• star_desires (str)

• recent_text (str)

async check_progression(user_id, ulm_vector, chaos_switch=None)

Check step progression via lattice position matching (v3).

Replaces the old ULM threshold system. Order of evaluation: 1. Resolve user’s current lattice position from ULM vector 2. Check if step has EXPIRED -> trigger failure_strategy 3. Check if any BRANCH lattice condition is met -> jump 4. Check if user is at/near target_nodes -> advance linearly 5. Fallback: if plan only has legacy completion_signals, use old logic

Return type:

Optional[str]

Parameters:

• user_id (str)

• ulm_vector (Dict[str, float])

static add_journal_entry(plan, note, category='system')

Append a timestamped, categorized note to the plan’s running journal.

The journal is the v2 mechanism by which plans are edited rather than scrapped, so every progression, branch, expiration and admin revision leaves an audit trail on the plan object. Mutates plan.journal_entries in place and trims it to the most recent 200 entries; it does not persist on its own – the caller is responsible for the subsequent _save_active_plan(). Called by _handle_step_expiration(), _follow_branch(), _advance_linear(), revise_step() and branch_step().

Parameters:

• plan (OpsPlan) – The plan whose journal is appended to (mutated).

• note (str) – The free-text observation to record.

• category (str) – Entry category – one of system, flash, meta, admin or daily; defaults to "system".

Return type:

None

revise_step(plan, step_id, description=None, tactic=None)

Edit one step of a live plan in place without regenerating the plan.

Locates the step by step_id and overwrites its description and/or tactic (only the arguments that are provided), bumps the plan’s revision_count so optimistic-locking in _save_active_plan() stays consistent, and logs the change to the journal via add_journal_entry() under the admin category. Mutates the plan object only – it does not persist; the caller (admin tooling) is expected to save afterward. No in-repo callers were found, so this is invoked dynamically via the psy-ops admin tooling / tests.

Parameters:

• plan (OpsPlan) – The plan to revise (mutated in place).

• step_id (str) – The id of the step to edit.

• description (str) – New description, or None to leave unchanged.

• tactic (str) – New tactic, or None to leave unchanged.

Returns:

True if a matching step was found and revised, else False.

Return type:

bool

branch_step(plan, step_id, branch)

Attach a new EITHER/OR branch to an existing step of a live plan.

Finds the step by step_id and appends branch to its branches list so future check_progression() calls can jump on that condition, bumps the plan’s revision_count for optimistic locking, and journals the addition via add_journal_entry() under the admin category. Mutates the plan object only – persistence is the caller’s responsibility. No in-repo callers were found, so this is invoked dynamically via the psy-ops admin tooling / tests.

Parameters:

• plan (OpsPlan) – The plan to modify (mutated in place).

• step_id (str) – The id of the step to attach the branch to.

• branch (OpBranch) – The branch to append.

Returns:

True if a matching step was found and updated, else False.

Return type:

bool

async get_plan_context(user_id)

Build the one-line ops-plan summary injected into Star’s system prompt.

Read-through accessor: loads the active plan via get_active_plan() (cache, falling back to the ops:{user_id}:active Redis key) and, when a plan is active, renders a compact line with the objective, the current step number/name/tactic, what Star should do, and the ULM thresholds that advance the step. Returns "" when there is no active plan and swallows any error so prompt assembly never fails on this. Called by the limbic coordinator.py while composing the prompt context, and exercised by the ops-planner and distributed e2e tests.

Parameters:

user_id (str) – The user whose active plan to summarize.

Returns:

A single-line plan summary, an “all steps complete” note, or "" when no active plan exists or an error occurred.

Return type:

str

async get_mermaid_with_position(user_id)

Return the plan’s Mermaid source with the live step state styled in.

Read-through accessor: loads the active plan via get_active_plan() (cache, falling back to Redis) and appends per-node Mermaid style directives so completed steps render green, the active step renders orange with a thick stroke, and pending steps keep the default dark theme. Returns "" when there is no plan or no stored chart, and swallows errors. Called by the admin chart API in web/ncm_chart_api.py and by the psy-ops admin tools in tools/psy_ops_tools.py, and covered by the ops-planner tests.

Parameters:

user_id (str) – The user whose plan chart to render.

Returns:

Mermaid source with appended style directives, or "" when no chart is available or an error occurred.

Return type:

str

async get_active_plan(user_id)

Fetch the user’s active ops plan, preferring the in-process cache.

The central read path for the engine: returns the cached OpsPlan when it is still active; otherwise reads the ops:{user_id}:active Redis key, decodes the JSON via OpsPlan.from_dict(), repopulates the cache, and returns it only if active. Yields None when nothing active exists in either tier. Called widely – internally by generate_plan(), check_progression(), get_plan_context(), get_mermaid_with_position() and abandon_plan(), and externally by the limbic coordinator.py, the admin API in web/ncm_chart_api.py, tools/psy_ops_tools.py and the test suite.

Parameters:

user_id (str) – The user whose active plan to load.

Returns:

The active plan, or None if none is active.

Return type:

Optional[OpsPlan]

async get_plan_history(user_id, limit=10)

Load the user’s archived (past) plans from Redis history.

Reads up to limit entries from the ops:{user_id}:history Redis list – the archive maintained by _archive_plan() – decoding each stored JSON blob into a plain dict (not a rehydrated OpsPlan). Tolerant of a missing Redis handle or read errors, returning whatever it managed to load. Called by the admin chart API in web/ncm_chart_api.py and the psy-ops admin tools in tools/psy_ops_tools.py.

Parameters:

• user_id (str) – The user whose history to read.

• limit (int) – Maximum number of past plans to return; defaults to 10.

Returns:

Past plan dicts, newest first, possibly empty.

Return type:

List[Dict[str, Any]]

async abandon_plan(user_id)

Abandon the user’s active plan and move it into history.

Loads the active plan via get_active_plan(), flips its status to abandoned, archives it through _archive_plan(), then deletes the ops:{user_id}:active Redis key and evicts the user from the in-process cache so subsequent reads return nothing. A no-op (returns False) when there is no active plan. Called by the admin chart API in web/ncm_chart_api.py and the psy-ops admin tools in tools/psy_ops_tools.py.

Parameters:

user_id (str) – The user whose active plan to abandon.

Returns:

True if a plan was found and abandoned, else False.

Return type:

bool