feat: Workshop Phase 1 — State Schema v1 (#404)
All checks were successful
Tests / lint (push) Successful in 3s
Tests / test (push) Successful in 57s

Co-authored-by: Kimi Agent <kimi@timmy.local>
Co-committed-by: Kimi Agent <kimi@timmy.local>
This commit was merged in pull request #404.
This commit is contained in:
2026-03-19 02:24:13 -04:00
committed by hermes
parent ab3546ae4b
commit 3571d528ad
4 changed files with 134 additions and 20 deletions

View File

@@ -65,7 +65,7 @@ def _build_world_state(presence: dict) -> dict:
"""Transform presence dict into the world/state API response."""
return {
"timmyState": {
"mood": presence.get("mood", "focused"),
"mood": presence.get("mood", "calm"),
"activity": presence.get("current_focus", "idle"),
"energy": presence.get("energy", 0.5),
"confidence": presence.get("confidence", 0.7),
@@ -93,7 +93,7 @@ def _get_current_state() -> dict:
presence = {
"version": 1,
"liveness": datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%SZ"),
"mood": "idle",
"mood": "calm",
"current_focus": "",
"active_threads": [],
"recent_events": [],

View File

@@ -4,13 +4,14 @@ Maintains Timmy's observable presence state for the Workshop 3D renderer.
Writes the presence file every 30 seconds (or on cognitive state change),
skipping writes when state is unchanged.
See ADR-023 for the schema contract.
See ADR-023 for the schema contract and issue #360 for the full v1 schema.
"""
import asyncio
import hashlib
import json
import logging
import time
from collections.abc import Awaitable, Callable
from datetime import UTC, datetime
from pathlib import Path
@@ -20,42 +21,125 @@ logger = logging.getLogger(__name__)
PRESENCE_FILE = Path.home() / ".timmy" / "presence.json"
HEARTBEAT_INTERVAL = 30 # seconds
# Cognitive mood → presence mood mapping
_MOOD_MAP = {
"curious": "exploring",
"settled": "focused",
# Cognitive mood → presence mood mapping (issue #360 enum values)
_MOOD_MAP: dict[str, str] = {
"curious": "contemplative",
"settled": "calm",
"hesitant": "uncertain",
"energized": "excited",
}
# Activity mapping from cognitive engagement
_ACTIVITY_MAP: dict[str, str] = {
"idle": "idle",
"surface": "thinking",
"deep": "thinking",
}
# Module-level energy tracker — decays over time, resets on interaction
_energy_state: dict[str, float] = {"value": 0.8, "last_interaction": time.monotonic()}
# Startup timestamp for uptime calculation
_start_time = time.monotonic()
# Energy decay: 0.01 per minute without interaction (per issue #360)
_ENERGY_DECAY_PER_SECOND = 0.01 / 60.0
_ENERGY_MIN = 0.1
def _time_of_day(hour: int) -> str:
"""Map hour (0-23) to a time-of-day label."""
if 5 <= hour < 12:
return "morning"
if 12 <= hour < 17:
return "afternoon"
if 17 <= hour < 21:
return "evening"
if 21 <= hour or hour < 2:
return "night"
return "deep-night"
def reset_energy() -> None:
"""Reset energy to full (called on interaction)."""
_energy_state["value"] = 0.8
_energy_state["last_interaction"] = time.monotonic()
def _current_energy() -> float:
"""Compute current energy with time-based decay."""
elapsed = time.monotonic() - _energy_state["last_interaction"]
decayed = _energy_state["value"] - (elapsed * _ENERGY_DECAY_PER_SECOND)
return max(_ENERGY_MIN, min(1.0, decayed))
def get_state_dict() -> dict:
"""Build presence state dict from current cognitive state.
Returns a v1 presence schema dict suitable for JSON serialisation.
Includes the full schema from issue #360: identity, mood, activity,
attention, interaction, environment, and meta sections.
"""
from timmy.cognitive_state import cognitive_tracker
state = cognitive_tracker.get_state()
now = datetime.now(UTC)
# Map cognitive engagement to presence mood fallback
mood = _MOOD_MAP.get(state.mood, "focused")
# Map cognitive mood to presence mood
mood = _MOOD_MAP.get(state.mood, "calm")
if state.engagement == "idle" and state.mood == "settled":
mood = "idle"
mood = "calm"
# Confidence from cognitive tracker
if state._confidence_count > 0:
confidence = state._confidence_sum / state._confidence_count
else:
confidence = 0.7
# Build active threads from commitments
threads = []
for commitment in state.active_commitments[:10]:
threads.append({"type": "thinking", "ref": commitment[:80], "status": "active"})
# Activity
activity = _ACTIVITY_MAP.get(state.engagement, "idle")
# Environment
local_now = datetime.now()
return {
"version": 1,
"liveness": datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%SZ"),
"liveness": now.strftime("%Y-%m-%dT%H:%M:%SZ"),
"current_focus": state.focus_topic or "",
"active_threads": threads,
"recent_events": [],
"concerns": [],
"mood": mood,
"confidence": round(max(0.0, min(1.0, confidence)), 2),
"energy": round(_current_energy(), 2),
"identity": {
"name": "Timmy",
"title": "The Workshop Wizard",
"uptime_seconds": int(time.monotonic() - _start_time),
},
"activity": {
"current": activity,
"detail": state.focus_topic or "",
},
"interaction": {
"visitor_present": False,
"conversation_turns": state.conversation_depth,
},
"environment": {
"time_of_day": _time_of_day(local_now.hour),
"local_time": local_now.strftime("%-I:%M %p"),
"day_of_week": local_now.strftime("%A"),
},
"meta": {
"schema_version": 1,
"updated_at": now.strftime("%Y-%m-%dT%H:%M:%SZ"),
"writer": "timmy-loop",
},
}
@@ -75,8 +159,8 @@ def write_state(state_dict: dict | None = None, path: Path | None = None) -> Non
def _state_hash(state_dict: dict) -> str:
"""Compute hash of state dict, ignoring liveness timestamp."""
stable = {k: v for k, v in state_dict.items() if k != "liveness"}
"""Compute hash of state dict, ignoring volatile timestamps."""
stable = {k: v for k, v in state_dict.items() if k not in ("liveness", "meta")}
return hashlib.md5(json.dumps(stable, sort_keys=True).encode()).hexdigest()

View File

@@ -52,7 +52,7 @@ def test_build_world_state_maps_fields():
def test_build_world_state_defaults():
"""Missing fields get safe defaults."""
result = _build_world_state({})
assert result["timmyState"]["mood"] == "focused"
assert result["timmyState"]["mood"] == "calm"
assert result["timmyState"]["energy"] == 0.5
assert result["version"] == 1
@@ -147,7 +147,7 @@ def test_world_state_endpoint_fallback(client, tmp_path):
mock_get.return_value = {
"version": 1,
"liveness": "2026-03-19T02:00:00Z",
"mood": "idle",
"mood": "calm",
"current_focus": "",
"active_threads": [],
"recent_events": [],
@@ -156,7 +156,7 @@ def test_world_state_endpoint_fallback(client, tmp_path):
resp = client.get("/api/world/state")
assert resp.status_code == 200
assert resp.json()["timmyState"]["mood"] == "idle"
assert resp.json()["timmyState"]["mood"] == "calm"
def test_world_state_endpoint_full_fallback(client, tmp_path):
@@ -172,7 +172,7 @@ def test_world_state_endpoint_full_fallback(client, tmp_path):
assert resp.status_code == 200
data = resp.json()
assert data["timmyState"]["mood"] == "idle"
assert data["timmyState"]["mood"] == "calm"
assert data["version"] == 1

View File

@@ -26,17 +26,47 @@ def test_get_state_dict_returns_v1_schema():
assert isinstance(state["active_threads"], list)
assert isinstance(state["recent_events"], list)
assert isinstance(state["concerns"], list)
# Issue #360 enriched fields
assert isinstance(state["confidence"], float)
assert 0.0 <= state["confidence"] <= 1.0
assert isinstance(state["energy"], float)
assert 0.0 <= state["energy"] <= 1.0
assert state["identity"]["name"] == "Timmy"
assert state["identity"]["title"] == "The Workshop Wizard"
assert isinstance(state["identity"]["uptime_seconds"], int)
assert state["activity"]["current"] in ("idle", "thinking")
assert state["environment"]["time_of_day"] in (
"morning",
"afternoon",
"evening",
"night",
"deep-night",
)
assert state["environment"]["day_of_week"] in (
"Monday",
"Tuesday",
"Wednesday",
"Thursday",
"Friday",
"Saturday",
"Sunday",
)
assert state["interaction"]["visitor_present"] is False
assert isinstance(state["interaction"]["conversation_turns"], int)
assert state["meta"]["schema_version"] == 1
assert state["meta"]["writer"] == "timmy-loop"
assert "updated_at" in state["meta"]
def test_get_state_dict_idle_mood():
"""Idle engagement + settled mood → 'idle' presence mood."""
"""Idle engagement + settled mood → 'calm' presence mood."""
from timmy.cognitive_state import CognitiveState, CognitiveTracker
tracker = CognitiveTracker.__new__(CognitiveTracker)
tracker.state = CognitiveState(engagement="idle", mood="settled")
with patch("timmy.cognitive_state.cognitive_tracker", tracker):
state = get_state_dict()
assert state["mood"] == "idle"
assert state["mood"] == "calm"
def test_get_state_dict_maps_mood():
@@ -44,7 +74,7 @@ def test_get_state_dict_maps_mood():
from timmy.cognitive_state import CognitiveState, CognitiveTracker
for cog_mood, expected in [
("curious", "exploring"),
("curious", "contemplative"),
("hesitant", "uncertain"),
("energized", "excited"),
]: