State Engine

Overview

StateEngine classifies the user into one of 9 states. Each state has a response policy that the chat pipeline uses to adjust Luna's tone and proactive behaviour — shorter in focus mode, warmer when tired, silent when sleeping.

Classification uses a priority waterfall: explicit voice signals first, then time-of-day heuristics, then PC activity, then learned patterns from historical StateEvent rows.

User states

Focus apps

These executable names trigger FOCUS_MODE:

_FOCUS_APPS = frozenset({
    "code.exe", "cursor.exe", "pycharm64.exe", "idea64.exe",
    "devenv.exe", "vim.exe", "nvim.exe", "sublime_text.exe",
    "rider64.exe", "clion64.exe", "fleet.exe",
})

Response policies

Each state has a tone, a behavior instruction, and a prompt_note that is injected into the system prompt when that state is active.

update()

Call this after every voice-input turn. Reads PC idle time and active app, classifies the state, logs a StateEvent row, and returns the new UserState.

def update(
    db: Session,
    transcript: str = "",
    emotion: str = "neutral",    # from voice emotion classifier
    volume: float | None = None, # RMS 0–1000
    speech_speed: float | None = None,  # words per minute
    speech_duration: float | None = None,
) -> UserState

from backend.services.state_engine import state_engine
from backend.models.database import SessionLocal

db = SessionLocal()
state = state_engine.update(
    db,
    transcript="I just got home",
    emotion="calm",
    volume=320.0,
    speech_speed=120.0,
)
print(state)  # UserState.BACK_FROM_WORK
db.close()

infer_passive()

Classifies state without a voice turn — uses only time-of-day, PC idle seconds, and active app. Used by the scheduler's state_aware_proactive()to check state between conversations.

def infer_passive(db: Session) -> UserState

state = state_engine.infer_passive(db)
if state == UserState.FOCUS_MODE:
    print("User is in deep work — skip proactive messages")

Classification logic

The _classify() method applies rules in priority order:

1. SLEEPING — 01:00–07:00 + PC idle > 30 min → immediate return.
2. JUST_WOKE_UP — 05:00–10:00, PC just active, prior state was SLEEPING.
3. AWAY — personal variant, last-seen gap > 20 min, PC now active.
4. BACK_FROM_WORK — 16:00–21:00, transcript contains work-return words.
5. STAYING_UP — 23:00–02:00, PC active.
6. LOW_ENERGY — voice emotion is sad/angry, or slow + quiet speech.
7. FOCUS_MODE — active app is in the focus-apps set; or late evening + very low volume.
8. RELAXING — 20:00–23:00, non-focus app.
9. Learned pattern — most common state at this hour from StateEvent history.
10. NORMAL — fallback.

build_state_context()

Returns a short markdown string suitable for injection into the system prompt. Empty string for NORMAL state (no extra context needed).

def build_state_context(state: UserState | None = None) -> str

ctx = state_engine.build_state_context()
# "## User state: FOCUS_MODE (for ~18 min)

#  The user is in deep-work / focus mode. Answer only what is directly asked..."

full_system = ctx + "\n\n" + personality_prompt + "\n\n" + memory_context

Pattern learning

Every call to update() logs a StateEvent row with hour, day of week, emotions, volume, speech speed, PC activity, and inferred state. Over time this builds a habit model:

# Get the most common state at a given hour
state_at_9am = state_engine.common_state_by_hour(db, hour=9)
print(state_at_9am)   # "JUST_WOKE_UP"

# Human-readable pattern summary
print(state_engine.narrative_summary(db))
# Learned habits:
#   Mon 09:00 — just waking up (14×)
#   Mon 14:00 — in deep-work mode (22×)
#   Fri 23:00 — staying up late (8×)

Patterns are surfaced by mine_behavioral_patterns() in the scheduler (runs daily at 04:00) and stored as behavior facts in the memory manager.

The state_engine singleton

A single StateEngine() instance — state_engine — is created at module import. Import it anywhere in the backend:

from backend.services.state_engine import state_engine, UserState

# Check current state without a DB call
print(state_engine.current_state)    # UserState.NORMAL
print(state_engine.state_since)      # datetime of last transition

# Get the response policy dict
policy = state_engine.get_response_policy()
print(policy["tone"])                # "normal"