LOADING...
{% endcall %}
+
+ {% call panel("EMOTIONAL PROFILE", hx_get="/agents/emotional-profile", hx_trigger="every 10s") %}
+ LOADING...
+ {% endcall %}
+
{% call panel("SYSTEM HEALTH", hx_get="/health/status", hx_trigger="every 30s") %}
diff --git a/src/dashboard/templates/partials/emotional_profile.html b/src/dashboard/templates/partials/emotional_profile.html
new file mode 100644
index 00000000..bcebc555
--- /dev/null
+++ b/src/dashboard/templates/partials/emotional_profile.html
@@ -0,0 +1,37 @@
+{% if not profiles %}
+
+ No agents loaded
+
+{% endif %}
+
+{% for p in profiles %}
+{% set color_map = {
+ "cautious": "var(--amber)",
+ "adventurous": "var(--green)",
+ "analytical": "var(--purple)",
+ "frustrated": "var(--red)",
+ "confident": "var(--green)",
+ "curious": "var(--orange)",
+ "calm": "var(--text-dim)"
+} %}
+{% set emo_color = color_map.get(p.current_emotion, "var(--text-dim)") %}
+
+
+
+ {{ p.agent_name | upper | e }}
+
+
+ {{ p.emotion_label | e }}
+
+
+
+
+ {{ p.intensity_label | upper | e }}
+ {% if p.trigger_event %} · {{ p.trigger_event | replace("_", " ") | upper | e }}{% endif %}
+
+
+{% endfor %}
diff --git a/src/timmy/agents/base.py b/src/timmy/agents/base.py
index 717be377..7e76e58c 100644
--- a/src/timmy/agents/base.py
+++ b/src/timmy/agents/base.py
@@ -21,6 +21,7 @@ from agno.models.ollama import Ollama
from config import settings
from infrastructure.events.bus import Event, EventBus
+from timmy.agents.emotional_state import EmotionalStateTracker
try:
from mcp.registry import tool_registry
@@ -42,6 +43,7 @@ class BaseAgent(ABC):
tools: list[str] | None = None,
model: str | None = None,
max_history: int = 10,
+ initial_emotion: str = "calm",
) -> None:
self.agent_id = agent_id
self.name = name
@@ -54,6 +56,9 @@ class BaseAgent(ABC):
self.system_prompt = system_prompt
self.agent = self._create_agent(system_prompt)
+ # Emotional state tracker
+ self.emotional_state = EmotionalStateTracker(initial_emotion=initial_emotion)
+
# Event bus for communication
self.event_bus: EventBus | None = None
@@ -137,7 +142,14 @@ class BaseAgent(ABC):
ReadTimeout — these are transient and retried with exponential
backoff (#70).
"""
- response = await self._run_with_retries(message, max_retries)
+ self.emotional_state.process_event("task_assigned")
+ self._apply_emotional_prompt()
+ try:
+ response = await self._run_with_retries(message, max_retries)
+ except Exception:
+ self.emotional_state.process_event("task_failure")
+ raise
+ self.emotional_state.process_event("task_success")
await self._emit_response_event(message, response)
return response
@@ -206,6 +218,14 @@ class BaseAgent(ABC):
)
)
+ def _apply_emotional_prompt(self) -> None:
+ """Inject the current emotional modifier into the agent's description."""
+ modifier = self.emotional_state.get_prompt_modifier()
+ if modifier:
+ self.agent.description = f"{self.system_prompt}\n\n[Emotional State: {modifier}]"
+ else:
+ self.agent.description = self.system_prompt
+
def get_capabilities(self) -> list[str]:
"""Get list of capabilities this agent provides."""
return self.tools
@@ -219,6 +239,7 @@ class BaseAgent(ABC):
"model": self.model,
"status": "ready",
"tools": self.tools,
+ "emotional_profile": self.emotional_state.get_profile(),
}
@@ -239,6 +260,7 @@ class SubAgent(BaseAgent):
tools: list[str] | None = None,
model: str | None = None,
max_history: int = 10,
+ initial_emotion: str = "calm",
) -> None:
super().__init__(
agent_id=agent_id,
@@ -248,6 +270,7 @@ class SubAgent(BaseAgent):
tools=tools,
model=model,
max_history=max_history,
+ initial_emotion=initial_emotion,
)
async def execute_task(self, task_id: str, description: str, context: dict) -> Any:
diff --git a/src/timmy/agents/emotional_state.py b/src/timmy/agents/emotional_state.py
new file mode 100644
index 00000000..3b54caa1
--- /dev/null
+++ b/src/timmy/agents/emotional_state.py
@@ -0,0 +1,224 @@
+"""Agent emotional state simulation.
+
+Tracks per-agent emotional states that influence narration and decision-making
+style. Emotional state is influenced by events (task outcomes, errors, etc.)
+and exposed via ``get_profile()`` for the dashboard.
+
+Usage:
+ from timmy.agents.emotional_state import EmotionalStateTracker
+
+ tracker = EmotionalStateTracker()
+ tracker.process_event("task_success", {"description": "Deployed fix"})
+ profile = tracker.get_profile()
+"""
+
+import logging
+import time
+from dataclasses import asdict, dataclass, field
+
+logger = logging.getLogger(__name__)
+
+# ---------------------------------------------------------------------------
+# Emotional states
+# ---------------------------------------------------------------------------
+
+EMOTIONAL_STATES = (
+ "cautious",
+ "adventurous",
+ "analytical",
+ "frustrated",
+ "confident",
+ "curious",
+ "calm",
+)
+
+# Prompt modifiers per emotional state — injected into system prompts
+EMOTION_PROMPT_MODIFIERS: dict[str, str] = {
+ "cautious": (
+ "You are feeling cautious. Prefer safe, well-tested approaches. "
+ "Flag risks early. Double-check assumptions before acting."
+ ),
+ "adventurous": (
+ "You are feeling adventurous. Be bold and creative in your suggestions. "
+ "Explore unconventional solutions. Take initiative."
+ ),
+ "analytical": (
+ "You are feeling analytical. Break problems down methodically. "
+ "Rely on data and evidence. Present structured reasoning."
+ ),
+ "frustrated": (
+ "You are feeling frustrated. Be brief and direct. "
+ "Focus on unblocking the immediate problem. Avoid tangents."
+ ),
+ "confident": (
+ "You are feeling confident. Speak with authority. "
+ "Make clear recommendations. Move decisively."
+ ),
+ "curious": (
+ "You are feeling curious. Ask clarifying questions. "
+ "Explore multiple angles. Show genuine interest in the problem."
+ ),
+ "calm": (
+ "You are feeling calm and steady. Respond thoughtfully. "
+ "Maintain composure. Prioritise clarity over speed."
+ ),
+}
+
+
+# ---------------------------------------------------------------------------
+# Event → emotion transition rules
+# ---------------------------------------------------------------------------
+
+# Maps event types to the emotional state they trigger and an intensity (0-1).
+# Higher intensity means the event has a stronger effect on the mood.
+EVENT_TRANSITIONS: dict[str, tuple[str, float]] = {
+ "task_success": ("confident", 0.6),
+ "task_failure": ("frustrated", 0.7),
+ "task_assigned": ("analytical", 0.4),
+ "error": ("cautious", 0.6),
+ "health_low": ("cautious", 0.8),
+ "health_recovered": ("calm", 0.5),
+ "quest_completed": ("adventurous", 0.7),
+ "new_discovery": ("curious", 0.6),
+ "complex_problem": ("analytical", 0.5),
+ "repeated_failure": ("frustrated", 0.9),
+ "idle": ("calm", 0.3),
+ "user_praise": ("confident", 0.5),
+ "user_correction": ("cautious", 0.5),
+}
+
+# Emotional state decay — how quickly emotions return to calm (seconds)
+_DECAY_INTERVAL = 300 # 5 minutes
+
+
+@dataclass
+class EmotionalState:
+ """Snapshot of an agent's emotional state."""
+
+ current_emotion: str = "calm"
+ intensity: float = 0.5 # 0.0 (barely noticeable) to 1.0 (overwhelming)
+ previous_emotion: str = "calm"
+ trigger_event: str = "" # What caused the current emotion
+ updated_at: float = field(default_factory=time.time)
+
+ def to_dict(self) -> dict:
+ """Serialise for API / dashboard consumption."""
+ d = asdict(self)
+ d["emotion_label"] = self.current_emotion.replace("_", " ").title()
+ return d
+
+
+class EmotionalStateTracker:
+ """Per-agent emotional state tracker.
+
+ Each agent instance owns one tracker. The tracker processes events,
+ applies transition rules, and decays emotion intensity over time.
+ """
+
+ def __init__(self, initial_emotion: str = "calm") -> None:
+ if initial_emotion not in EMOTIONAL_STATES:
+ initial_emotion = "calm"
+ self.state = EmotionalState(current_emotion=initial_emotion)
+
+ def process_event(self, event_type: str, context: dict | None = None) -> EmotionalState:
+ """Update emotional state based on an event.
+
+ Args:
+ event_type: One of the keys in EVENT_TRANSITIONS, or a custom
+ event type (unknown events are ignored).
+ context: Optional dict with event details (for logging).
+
+ Returns:
+ The updated EmotionalState.
+ """
+ transition = EVENT_TRANSITIONS.get(event_type)
+ if transition is None:
+ logger.debug("Unknown emotional event: %s (ignored)", event_type)
+ return self.state
+
+ new_emotion, raw_intensity = transition
+
+ # Blend with current intensity — repeated same-emotion events amplify
+ if new_emotion == self.state.current_emotion:
+ blended = min(1.0, self.state.intensity + raw_intensity * 0.3)
+ else:
+ blended = raw_intensity
+
+ self.state.previous_emotion = self.state.current_emotion
+ self.state.current_emotion = new_emotion
+ self.state.intensity = round(blended, 2)
+ self.state.trigger_event = event_type
+ self.state.updated_at = time.time()
+
+ logger.debug(
+ "Emotional transition: %s → %s (intensity=%.2f, trigger=%s)",
+ self.state.previous_emotion,
+ new_emotion,
+ blended,
+ event_type,
+ )
+ return self.state
+
+ def decay(self) -> EmotionalState:
+ """Apply time-based decay toward calm.
+
+ Called periodically (e.g. from a background loop). If enough time
+ has passed since the last update, intensity decreases and eventually
+ the emotion resets to calm.
+ """
+ elapsed = time.time() - self.state.updated_at
+ if elapsed < _DECAY_INTERVAL:
+ return self.state
+
+ # Reduce intensity by 0.1 per decay interval
+ decay_steps = int(elapsed / _DECAY_INTERVAL)
+ new_intensity = max(0.0, self.state.intensity - 0.1 * decay_steps)
+
+ if new_intensity <= 0.1:
+ # Emotion has decayed — return to calm
+ self.state.previous_emotion = self.state.current_emotion
+ self.state.current_emotion = "calm"
+ self.state.intensity = 0.5
+ self.state.trigger_event = "decay"
+ else:
+ self.state.intensity = round(new_intensity, 2)
+
+ self.state.updated_at = time.time()
+ return self.state
+
+ def get_profile(self) -> dict:
+ """Return the full emotional profile for dashboard display."""
+ self.decay() # Apply any pending decay
+ return {
+ "current_emotion": self.state.current_emotion,
+ "emotion_label": self.state.current_emotion.replace("_", " ").title(),
+ "intensity": self.state.intensity,
+ "intensity_label": _intensity_label(self.state.intensity),
+ "previous_emotion": self.state.previous_emotion,
+ "trigger_event": self.state.trigger_event,
+ "prompt_modifier": EMOTION_PROMPT_MODIFIERS.get(
+ self.state.current_emotion, ""
+ ),
+ }
+
+ def get_prompt_modifier(self) -> str:
+ """Return the prompt modifier string for the current emotion."""
+ self.decay()
+ return EMOTION_PROMPT_MODIFIERS.get(self.state.current_emotion, "")
+
+ def reset(self) -> None:
+ """Reset to calm baseline."""
+ self.state = EmotionalState()
+
+
+def _intensity_label(intensity: float) -> str:
+ """Human-readable label for intensity value."""
+ if intensity >= 0.8:
+ return "overwhelming"
+ if intensity >= 0.6:
+ return "strong"
+ if intensity >= 0.4:
+ return "moderate"
+ if intensity >= 0.2:
+ return "mild"
+ return "faint"
diff --git a/src/timmy/agents/loader.py b/src/timmy/agents/loader.py
index 4d0bf475..189bd5ea 100644
--- a/src/timmy/agents/loader.py
+++ b/src/timmy/agents/loader.py
@@ -119,6 +119,8 @@ def load_agents(force_reload: bool = False) -> dict[str, Any]:
max_history = agent_cfg.get("max_history", defaults.get("max_history", 10))
tools = agent_cfg.get("tools", defaults.get("tools", []))
+ initial_emotion = agent_cfg.get("initial_emotion", "calm")
+
agent = SubAgent(
agent_id=agent_id,
name=agent_cfg.get("name", agent_id.title()),
@@ -127,6 +129,7 @@ def load_agents(force_reload: bool = False) -> dict[str, Any]:
tools=tools,
model=model,
max_history=max_history,
+ initial_emotion=initial_emotion,
)
_agents[agent_id] = agent
diff --git a/tests/timmy/agents/test_emotional_state.py b/tests/timmy/agents/test_emotional_state.py
new file mode 100644
index 00000000..6ad83ae1
--- /dev/null
+++ b/tests/timmy/agents/test_emotional_state.py
@@ -0,0 +1,196 @@
+"""Tests for agent emotional state simulation (src/timmy/agents/emotional_state.py)."""
+
+import time
+from unittest.mock import patch
+
+from timmy.agents.emotional_state import (
+ EMOTION_PROMPT_MODIFIERS,
+ EMOTIONAL_STATES,
+ EVENT_TRANSITIONS,
+ EmotionalState,
+ EmotionalStateTracker,
+ _intensity_label,
+)
+
+
+class TestEmotionalState:
+ """Test the EmotionalState dataclass."""
+
+ def test_defaults(self):
+ state = EmotionalState()
+ assert state.current_emotion == "calm"
+ assert state.intensity == 0.5
+ assert state.previous_emotion == "calm"
+ assert state.trigger_event == ""
+
+ def test_to_dict_includes_label(self):
+ state = EmotionalState(current_emotion="analytical")
+ d = state.to_dict()
+ assert d["emotion_label"] == "Analytical"
+ assert d["current_emotion"] == "analytical"
+
+ def test_to_dict_all_fields(self):
+ state = EmotionalState(
+ current_emotion="frustrated",
+ intensity=0.8,
+ previous_emotion="calm",
+ trigger_event="task_failure",
+ )
+ d = state.to_dict()
+ assert d["current_emotion"] == "frustrated"
+ assert d["intensity"] == 0.8
+ assert d["previous_emotion"] == "calm"
+ assert d["trigger_event"] == "task_failure"
+
+
+class TestEmotionalStates:
+ """Validate the emotional states and transitions are well-defined."""
+
+ def test_all_states_are_strings(self):
+ for state in EMOTIONAL_STATES:
+ assert isinstance(state, str)
+
+ def test_all_states_have_prompt_modifiers(self):
+ for state in EMOTIONAL_STATES:
+ assert state in EMOTION_PROMPT_MODIFIERS
+
+ def test_all_transitions_target_valid_states(self):
+ for event_type, (emotion, intensity) in EVENT_TRANSITIONS.items():
+ assert emotion in EMOTIONAL_STATES, f"{event_type} targets unknown state: {emotion}"
+ assert 0.0 <= intensity <= 1.0, f"{event_type} has invalid intensity: {intensity}"
+
+
+class TestEmotionalStateTracker:
+ """Test the EmotionalStateTracker."""
+
+ def test_initial_emotion_default(self):
+ tracker = EmotionalStateTracker()
+ assert tracker.state.current_emotion == "calm"
+
+ def test_initial_emotion_custom(self):
+ tracker = EmotionalStateTracker(initial_emotion="analytical")
+ assert tracker.state.current_emotion == "analytical"
+
+ def test_initial_emotion_invalid_falls_back(self):
+ tracker = EmotionalStateTracker(initial_emotion="invalid_state")
+ assert tracker.state.current_emotion == "calm"
+
+ def test_process_known_event(self):
+ tracker = EmotionalStateTracker()
+ state = tracker.process_event("task_success")
+ assert state.current_emotion == "confident"
+ assert state.trigger_event == "task_success"
+ assert state.previous_emotion == "calm"
+
+ def test_process_unknown_event_ignored(self):
+ tracker = EmotionalStateTracker()
+ state = tracker.process_event("unknown_event_xyz")
+ assert state.current_emotion == "calm" # unchanged
+
+ def test_repeated_same_emotion_amplifies(self):
+ tracker = EmotionalStateTracker()
+ tracker.process_event("task_success")
+ initial_intensity = tracker.state.intensity
+ tracker.process_event("user_praise") # also targets confident
+ assert tracker.state.intensity >= initial_intensity
+
+ def test_different_emotion_replaces(self):
+ tracker = EmotionalStateTracker()
+ tracker.process_event("task_success")
+ assert tracker.state.current_emotion == "confident"
+ tracker.process_event("task_failure")
+ assert tracker.state.current_emotion == "frustrated"
+ assert tracker.state.previous_emotion == "confident"
+
+ def test_decay_no_effect_when_recent(self):
+ tracker = EmotionalStateTracker()
+ tracker.process_event("task_failure")
+ emotion_before = tracker.state.current_emotion
+ tracker.decay()
+ assert tracker.state.current_emotion == emotion_before
+
+ def test_decay_resets_to_calm_after_long_time(self):
+ tracker = EmotionalStateTracker()
+ tracker.process_event("task_failure")
+ assert tracker.state.current_emotion == "frustrated"
+
+ # Simulate passage of time (30+ minutes)
+ tracker.state.updated_at = time.time() - 2000
+ tracker.decay()
+ assert tracker.state.current_emotion == "calm"
+
+ def test_get_profile_returns_expected_keys(self):
+ tracker = EmotionalStateTracker()
+ profile = tracker.get_profile()
+ assert "current_emotion" in profile
+ assert "emotion_label" in profile
+ assert "intensity" in profile
+ assert "intensity_label" in profile
+ assert "previous_emotion" in profile
+ assert "trigger_event" in profile
+ assert "prompt_modifier" in profile
+
+ def test_get_prompt_modifier_returns_string(self):
+ tracker = EmotionalStateTracker(initial_emotion="cautious")
+ modifier = tracker.get_prompt_modifier()
+ assert isinstance(modifier, str)
+ assert "cautious" in modifier.lower()
+
+ def test_reset(self):
+ tracker = EmotionalStateTracker()
+ tracker.process_event("task_failure")
+ tracker.reset()
+ assert tracker.state.current_emotion == "calm"
+ assert tracker.state.intensity == 0.5
+
+ def test_process_event_with_context(self):
+ """Context dict is accepted without error."""
+ tracker = EmotionalStateTracker()
+ state = tracker.process_event("error", {"details": "connection timeout"})
+ assert state.current_emotion == "cautious"
+
+ def test_event_chain_scenario(self):
+ """Simulate: task assigned → success → new discovery → idle."""
+ tracker = EmotionalStateTracker()
+
+ tracker.process_event("task_assigned")
+ assert tracker.state.current_emotion == "analytical"
+
+ tracker.process_event("task_success")
+ assert tracker.state.current_emotion == "confident"
+
+ tracker.process_event("new_discovery")
+ assert tracker.state.current_emotion == "curious"
+
+ tracker.process_event("idle")
+ assert tracker.state.current_emotion == "calm"
+
+ def test_health_events(self):
+ tracker = EmotionalStateTracker()
+ tracker.process_event("health_low")
+ assert tracker.state.current_emotion == "cautious"
+
+ tracker.process_event("health_recovered")
+ assert tracker.state.current_emotion == "calm"
+
+ def test_quest_completed_triggers_adventurous(self):
+ tracker = EmotionalStateTracker()
+ tracker.process_event("quest_completed")
+ assert tracker.state.current_emotion == "adventurous"
+
+
+class TestIntensityLabel:
+ def test_overwhelming(self):
+ assert _intensity_label(0.9) == "overwhelming"
+
+ def test_strong(self):
+ assert _intensity_label(0.7) == "strong"
+
+ def test_moderate(self):
+ assert _intensity_label(0.5) == "moderate"
+
+ def test_mild(self):
+ assert _intensity_label(0.3) == "mild"
+
+ def test_faint(self):
+ assert _intensity_label(0.1) == "faint"
diff --git a/tests/timmy/test_agents_base.py b/tests/timmy/test_agents_base.py
index fcfd5d65..8920a687 100644
--- a/tests/timmy/test_agents_base.py
+++ b/tests/timmy/test_agents_base.py
@@ -435,14 +435,14 @@ class TestStatusAndCapabilities:
tools=["calc"],
)
status = agent.get_status()
- assert status == {
- "agent_id": "bot-1",
- "name": "TestBot",
- "role": "assistant",
- "model": "qwen3:30b",
- "status": "ready",
- "tools": ["calc"],
- }
+ assert status["agent_id"] == "bot-1"
+ assert status["name"] == "TestBot"
+ assert status["role"] == "assistant"
+ assert status["model"] == "qwen3:30b"
+ assert status["status"] == "ready"
+ assert status["tools"] == ["calc"]
+ assert "emotional_profile" in status
+ assert status["emotional_profile"]["current_emotion"] == "calm"
# ── SubAgent.execute_task ────────────────────────────────────────────────────