persona_env OpenEnv Docker Space

Dockerfile

@@ -0,0 +1,80 @@
+# Copyright (c) Meta Platforms, Inc. and affiliates.
+# All rights reserved.
+#
+# This source code is licensed under the BSD-style license found in the
+# LICENSE file in the root directory of this source tree.
+
+# Multi-stage build using openenv-base
+# This Dockerfile is flexible and works for both:
+# - In-repo environments (with local OpenEnv sources)
+# - Standalone environments (with openenv from PyPI/Git)
+# The build script (openenv build) handles context detection and sets appropriate build args.
+
+ARG BASE_IMAGE=ghcr.io/meta-pytorch/openenv-base:latest
+FROM ${BASE_IMAGE} AS builder
+
+WORKDIR /app
+
+# Ensure git is available (required for installing dependencies from VCS)
+RUN apt-get update && \
+    apt-get install -y --no-install-recommends git && \
+    rm -rf /var/lib/apt/lists/*
+
+# Build argument to control whether we're building standalone or in-repo
+ARG BUILD_MODE=in-repo
+ARG ENV_NAME=persona_env
+
+# Copy environment code (always at root of build context)
+COPY . /app/env
+
+# For in-repo builds, openenv is already vendored in the build context
+# For standalone builds, openenv will be installed via pyproject.toml
+WORKDIR /app/env
+
+# Ensure uv is available (for local builds where base image lacks it)
+RUN if ! command -v uv >/dev/null 2>&1; then \
+        curl -LsSf https://astral.sh/uv/install.sh | sh && \
+        mv /root/.local/bin/uv /usr/local/bin/uv && \
+        mv /root/.local/bin/uvx /usr/local/bin/uvx; \
+    fi
+    
+# Install dependencies using uv sync --no-dev
+# If uv.lock exists, use it; otherwise resolve on the fly
+RUN --mount=type=cache,target=/root/.cache/uv \
+    if [ -f uv.lock ]; then \
+        uv sync --no-dev --frozen --no-install-project --no-editable; \
+    else \
+        uv sync --no-dev --no-install-project --no-editable; \
+    fi
+
+RUN --mount=type=cache,target=/root/.cache/uv \
+    if [ -f uv.lock ]; then \
+        uv sync --no-dev --frozen --no-editable; \
+    else \
+        uv sync --no-dev --no-editable; \
+    fi
+
+# Final runtime stage
+FROM ${BASE_IMAGE}
+
+WORKDIR /app
+
+# Copy the virtual environment from builder
+COPY --from=builder /app/env/.venv /app/.venv
+
+# Copy the environment code
+COPY --from=builder /app/env /app/env
+
+# Set PATH to use the virtual environment
+ENV PATH="/app/.venv/bin:$PATH"
+
+# Set PYTHONPATH so imports work correctly
+ENV PYTHONPATH="/app/env:$PYTHONPATH"
+
+# Health check
+HEALTHCHECK --interval=30s --timeout=3s --start-period=5s --retries=3 \
+    CMD curl -f http://localhost:${PORT:-8000}/health || exit 1
+
+# Run the FastAPI server
+# The module path is constructed to work with the /app/env structure
+CMD ["sh", "-c", "cd /app/env && uvicorn server.app:app --host 0.0.0.0 --port ${PORT:-8000}"]

README.md

@@ -0,0 +1,255 @@
+---
+title: Persona Env Environment Server
+emoji: 🎾
+colorFrom: red
+colorTo: purple
+sdk: docker
+pinned: false
+app_port: 8000
+base_path: /web
+tags:
+  - openenv
+---
+
+# Persona Env Environment
+
+A simple test environment that echoes back messages. Perfect for testing the env APIs as well as demonstrating environment usage patterns.
+
+## Quick Start
+
+The simplest way to use the Persona Env environment is through the `PersonaEnv` class:
+
+```python
+from persona_env import PersonaAction, PersonaEnv
+
+try:
+    # Create environment from Docker image
+    persona_envenv = PersonaEnv.from_docker_image("persona_env-env:latest")
+
+    # Reset
+    result = persona_envenv.reset()
+    print(f"Reset: {result.observation.echoed_message}")
+
+    # Send multiple messages
+    messages = ["Hello, World!", "Testing echo", "Final message"]
+
+    for msg in messages:
+        result = persona_envenv.step(PersonaAction(message=msg))
+        print(f"Sent: '{msg}'")
+        print(f"  → Echoed: '{result.observation.echoed_message}'")
+        print(f"  → Length: {result.observation.message_length}")
+        print(f"  → Reward: {result.reward}")
+
+finally:
+    # Always clean up
+    persona_envenv.close()
+```
+
+That's it! The `PersonaEnv.from_docker_image()` method handles:
+- Starting the Docker container
+- Waiting for the server to be ready
+- Connecting to the environment
+- Container cleanup when you call `close()`
+
+## Building the Docker Image
+
+Before using the environment, you need to build the Docker image:
+
+```bash
+# From project root
+docker build -t persona_env-env:latest -f server/Dockerfile .
+```
+
+## Deploying to Hugging Face Spaces
+
+You can easily deploy your OpenEnv environment to Hugging Face Spaces using the `openenv push` command:
+
+```bash
+# From the environment directory (where openenv.yaml is located)
+openenv push
+
+# Or specify options
+openenv push --namespace my-org --private
+```
+
+The `openenv push` command will:
+1. Validate that the directory is an OpenEnv environment (checks for `openenv.yaml`)
+2. Prepare a custom build for Hugging Face Docker space (enables web interface)
+3. Upload to Hugging Face (ensuring you're logged in)
+
+### Prerequisites
+
+- Authenticate with Hugging Face: The command will prompt for login if not already authenticated
+
+### Options
+
+- `--directory`, `-d`: Directory containing the OpenEnv environment (defaults to current directory)
+- `--repo-id`, `-r`: Repository ID in format 'username/repo-name' (defaults to 'username/env-name' from openenv.yaml)
+- `--base-image`, `-b`: Base Docker image to use (overrides Dockerfile FROM)
+- `--private`: Deploy the space as private (default: public)
+
+### Examples
+
+```bash
+# Push to your personal namespace (defaults to username/env-name from openenv.yaml)
+openenv push
+
+# Push to a specific repository
+openenv push --repo-id my-org/my-env
+
+# Push with a custom base image
+openenv push --base-image ghcr.io/meta-pytorch/openenv-base:latest
+
+# Push as a private space
+openenv push --private
+
+# Combine options
+openenv push --repo-id my-org/my-env --base-image custom-base:latest --private
+```
+
+After deployment, your space will be available at:
+`https://huggingface.co/spaces/<repo-id>`
+
+The deployed space includes:
+- **Web Interface** at `/web` - Interactive UI for exploring the environment
+- **API Documentation** at `/docs` - Full OpenAPI/Swagger interface
+- **Health Check** at `/health` - Container health monitoring
+- **WebSocket** at `/ws` - Persistent session endpoint for low-latency interactions
+
+## Environment Details
+
+### Action
+**PersonaAction**: Contains a single field
+- `message` (str) - The message to echo back
+
+### Observation
+**PersonaObservation**: Contains the echo response and metadata
+- `echoed_message` (str) - The message echoed back
+- `message_length` (int) - Length of the message
+- `reward` (float) - Reward based on message length (length × 0.1)
+- `done` (bool) - Always False for echo environment
+- `metadata` (dict) - Additional info like step count
+
+### Reward
+The reward is calculated as: `message_length × 0.1`
+- "Hi" → reward: 0.2
+- "Hello, World!" → reward: 1.3
+- Empty message → reward: 0.0
+
+## Advanced Usage
+
+### Connecting to an Existing Server
+
+If you already have a Persona Env environment server running, you can connect directly:
+
+```python
+from persona_env import PersonaEnv
+
+# Connect to existing server
+persona_envenv = PersonaEnv(base_url="<ENV_HTTP_URL_HERE>")
+
+# Use as normal
+result = persona_envenv.reset()
+result = persona_envenv.step(PersonaAction(message="Hello!"))
+```
+
+Note: When connecting to an existing server, `persona_envenv.close()` will NOT stop the server.
+
+### Using the Context Manager
+
+The client supports context manager usage for automatic connection management:
+
+```python
+from persona_env import PersonaAction, PersonaEnv
+
+# Connect with context manager (auto-connects and closes)
+with PersonaEnv(base_url="http://localhost:8000") as env:
+    result = env.reset()
+    print(f"Reset: {result.observation.echoed_message}")
+    # Multiple steps with low latency
+    for msg in ["Hello", "World", "!"]:
+        result = env.step(PersonaAction(message=msg))
+        print(f"Echoed: {result.observation.echoed_message}")
+```
+
+The client uses WebSocket connections for:
+- **Lower latency**: No HTTP connection overhead per request
+- **Persistent session**: Server maintains your environment state
+- **Efficient for episodes**: Better for many sequential steps
+
+### Concurrent WebSocket Sessions
+
+The server supports multiple concurrent WebSocket connections. To enable this,
+modify `server/app.py` to use factory mode:
+
+```python
+# In server/app.py - use factory mode for concurrent sessions
+app = create_app(
+    PersonaEnvironment,  # Pass class, not instance
+    PersonaAction,
+    PersonaObservation,
+    max_concurrent_envs=4,  # Allow 4 concurrent sessions
+)
+```
+
+Then multiple clients can connect simultaneously:
+
+```python
+from persona_env import PersonaAction, PersonaEnv
+from concurrent.futures import ThreadPoolExecutor
+
+def run_episode(client_id: int):
+    with PersonaEnv(base_url="http://localhost:8000") as env:
+        result = env.reset()
+        for i in range(10):
+            result = env.step(PersonaAction(message=f"Client {client_id}, step {i}"))
+        return client_id, result.observation.message_length
+
+# Run 4 episodes concurrently
+with ThreadPoolExecutor(max_workers=4) as executor:
+    results = list(executor.map(run_episode, range(4)))
+```
+
+## Development & Testing
+
+### Direct Environment Testing
+
+Test the environment logic directly without starting the HTTP server:
+
+```bash
+# From the server directory
+python3 server/persona_env_environment.py
+```
+
+This verifies that:
+- Environment resets correctly
+- Step executes actions properly
+- State tracking works
+- Rewards are calculated correctly
+
+### Running Locally
+
+Run the server locally for development:
+
+```bash
+uvicorn server.app:app --reload
+```
+
+## Project Structure
+
+```
+persona_env/
+├── .dockerignore         # Docker build exclusions
+├── __init__.py            # Module exports
+├── README.md              # This file
+├── openenv.yaml           # OpenEnv manifest
+├── pyproject.toml         # Project metadata and dependencies
+├── uv.lock                # Locked dependencies (generated)
+├── client.py              # PersonaEnv client
+├── models.py              # Action and Observation models
+└── server/
+    ├── __init__.py        # Server module exports
+    ├── persona_env_environment.py  # Core environment logic
+    ├── app.py             # FastAPI application (HTTP + WebSocket endpoints)
+    └── Dockerfile         # Container image definition
+```

__init__.py

@@ -0,0 +1,16 @@
+# Copyright (c) Meta Platforms, Inc. and affiliates.
+# All rights reserved.
+#
+# This source code is licensed under the BSD-style license found in the
+# LICENSE file in the root directory of this source tree.
+
+"""Persona Env Environment."""
+
+from .client import PersonaEnv
+from .models import PersonaAction, PersonaObservation
+
+__all__ = [
+    "PersonaAction",
+    "PersonaObservation",
+    "PersonaEnv",
+]

client.py

@@ -0,0 +1,99 @@
+# Copyright (c) Meta Platforms, Inc. and affiliates.
+# All rights reserved.
+#
+# This source code is licensed under the BSD-style license found in the
+# LICENSE file in the root directory of this source tree.
+
+"""Persona Env Environment Client."""
+
+from typing import Dict
+
+from openenv.core.client_types import StepResult
+from openenv.core.env_server.types import State
+from openenv.core import EnvClient
+
+from .models import PersonaAction, PersonaObservation
+
+
+class PersonaEnv(
+    EnvClient[PersonaAction, PersonaObservation]
+):
+    """
+    Client for the Persona Env Environment.
+
+    This client maintains a persistent WebSocket connection to the environment server,
+    enabling efficient multi-step interactions with lower latency.
+    Each client instance has its own dedicated environment session on the server.
+
+    Example:
+        >>> # Connect to a running server
+        >>> with PersonaEnv(base_url="http://localhost:8000") as client:
+        ...     result = client.reset()
+        ...     print(result.observation.echoed_message)
+        ...
+        ...     result = client.step(PersonaAction(message="Hello!"))
+        ...     print(result.observation.echoed_message)
+
+    Example with Docker:
+        >>> # Automatically start container and connect
+        >>> client = PersonaEnv.from_docker_image("persona_env-env:latest")
+        >>> try:
+        ...     result = client.reset()
+        ...     result = client.step(PersonaAction(message="Test"))
+        ... finally:
+        ...     client.close()
+    """
+
+    def _step_payload(self, action: PersonaAction) -> Dict:
+        """
+        Convert PersonaAction to JSON payload for step message.
+
+        Args:
+            action: PersonaAction instance
+
+        Returns:
+            Dictionary representation suitable for JSON encoding
+        """
+        return {
+            "message": action.message,
+        }
+
+    def _parse_result(self, payload: Dict) -> StepResult[PersonaObservation]:
+        """
+        Parse server response into StepResult[PersonaObservation].
+
+        Args:
+            payload: JSON response data from server
+
+        Returns:
+            StepResult with PersonaObservation
+        """
+        obs_data = payload.get("observation", {})
+        observation = PersonaObservation(
+            echoed_message=obs_data.get("echoed_message", ""),
+            message_length=obs_data.get("message_length", 0),
+            done=payload.get("done", False),
+            reward=payload.get("reward"),
+            metadata=obs_data.get("metadata", {}),
+        )
+
+        return StepResult(
+            observation=observation,
+            reward=payload.get("reward"),
+            done=payload.get("done", False),
+        )
+
+    def _parse_state(self, payload: Dict) -> State:
+        """
+        Parse server response into State object.
+
+        Args:
+            payload: JSON response from state request
+
+        Returns:
+            State object with episode_id and step_count
+        """
+        return State(
+            episode_id=payload.get("episode_id"),
+            step_count=payload.get("step_count", 0),
+        )

models.py

@@ -0,0 +1,29 @@
+from __future__ import annotations
+
+from typing import Dict, Literal, Optional
+from pydantic import Field
+from openenv.core.env_server.types import Action, Observation
+
+
+class PersonaAction(Action):
+    # = one of three actions
+    kind: Literal["show_content", "ask_question", "advance_time"] = Field(
+        ..., description="Which action to apply"
+    )
+
+    # = show_content
+    topic: Optional[str] = None
+    source: Optional[str] = None
+    valence: Optional[Literal["positive", "neutral", "negative"]] = None
+
+    # = ask_question
+    question: Optional[str] = None
+
+    # = advance_time
+    hours: Optional[int] = None
+
+
+class PersonaObservation(Observation):
+    reaction_text: str
+    mood: float
+    interests: Dict[str, float]

openenv.yaml

@@ -0,0 +1,7 @@
+spec_version: 1
+name: persona_env
+type: space
+runtime: fastapi
+app: server.app:app
+port: 8000
+

pyproject.toml

@@ -0,0 +1,45 @@
+# Copyright (c) Meta Platforms, Inc. and affiliates.
+# All rights reserved.
+#
+# This source code is licensed under the BSD-style license found in the
+# LICENSE file in the root directory of this source tree.
+
+[build-system]
+requires = ["setuptools>=45", "wheel"]
+build-backend = "setuptools.build_meta"
+
+[project]
+name = "openenv-persona_env"
+version = "0.1.0"
+description = "Persona Env environment for OpenEnv"
+requires-python = ">=3.10"
+dependencies = [
+    # Core OpenEnv runtime (provides FastAPI server + HTTP client types)
+    # install from github
+    # "openenv-core[core] @ git+https://github.com/meta-pytorch/OpenEnv.git",
+    "openenv-core[core]>=0.2.0",
+    # Environment-specific dependencies
+    # Add all dependencies needed for your environment here
+    # Examples:
+    # "numpy>=1.19.0",
+    # "torch>=2.0.0",
+    # "gymnasium>=0.29.0",
+    # "openspiel>=1.0.0",
+    # "smolagents>=1.22.0,<2",
+]
+
+[project.optional-dependencies]
+dev = [
+    "pytest>=8.0.0",
+    "pytest-cov>=4.0.0",
+]
+
+[project.scripts]
+# Server entry point - enables running via: uv run --project . server
+# or: python -m persona_env.server.app
+server = "persona_env.server.app:main"
+
+[tool.setuptools]
+include-package-data = true
+packages = ["persona_env", "persona_env.server"]
+package-dir = { "persona_env" = ".", "persona_env.server" = "server" }

server/Dockerfile

@@ -0,0 +1,80 @@
+# Copyright (c) Meta Platforms, Inc. and affiliates.
+# All rights reserved.
+#
+# This source code is licensed under the BSD-style license found in the
+# LICENSE file in the root directory of this source tree.
+
+# Multi-stage build using openenv-base
+# This Dockerfile is flexible and works for both:
+# - In-repo environments (with local OpenEnv sources)
+# - Standalone environments (with openenv from PyPI/Git)
+# The build script (openenv build) handles context detection and sets appropriate build args.
+
+ARG BASE_IMAGE=ghcr.io/meta-pytorch/openenv-base:latest
+FROM ${BASE_IMAGE} AS builder
+
+WORKDIR /app
+
+# Ensure git is available (required for installing dependencies from VCS)
+RUN apt-get update && \
+    apt-get install -y --no-install-recommends git && \
+    rm -rf /var/lib/apt/lists/*
+
+# Build argument to control whether we're building standalone or in-repo
+ARG BUILD_MODE=in-repo
+ARG ENV_NAME=persona_env
+
+# Copy environment code (always at root of build context)
+COPY . /app/env
+
+# For in-repo builds, openenv is already vendored in the build context
+# For standalone builds, openenv will be installed via pyproject.toml
+WORKDIR /app/env
+
+# Ensure uv is available (for local builds where base image lacks it)
+RUN if ! command -v uv >/dev/null 2>&1; then \
+        curl -LsSf https://astral.sh/uv/install.sh | sh && \
+        mv /root/.local/bin/uv /usr/local/bin/uv && \
+        mv /root/.local/bin/uvx /usr/local/bin/uvx; \
+    fi
+    
+# Install dependencies using uv sync --no-dev
+# If uv.lock exists, use it; otherwise resolve on the fly
+RUN --mount=type=cache,target=/root/.cache/uv \
+    if [ -f uv.lock ]; then \
+        uv sync --no-dev --frozen --no-install-project --no-editable; \
+    else \
+        uv sync --no-dev --no-install-project --no-editable; \
+    fi
+
+RUN --mount=type=cache,target=/root/.cache/uv \
+    if [ -f uv.lock ]; then \
+        uv sync --no-dev --frozen --no-editable; \
+    else \
+        uv sync --no-dev --no-editable; \
+    fi
+
+# Final runtime stage
+FROM ${BASE_IMAGE}
+
+WORKDIR /app
+
+# Copy the virtual environment from builder
+COPY --from=builder /app/env/.venv /app/.venv
+
+# Copy the environment code
+COPY --from=builder /app/env /app/env
+
+# Set PATH to use the virtual environment
+ENV PATH="/app/.venv/bin:$PATH"
+
+# Set PYTHONPATH so imports work correctly
+ENV PYTHONPATH="/app/env:$PYTHONPATH"
+
+# Health check
+HEALTHCHECK --interval=30s --timeout=3s --start-period=5s --retries=3 \
+    CMD curl -f http://localhost:${PORT:-8000}/health || exit 1
+
+# Run the FastAPI server
+# The module path is constructed to work with the /app/env structure
+CMD ["sh", "-c", "cd /app/env && uvicorn server.app:app --host 0.0.0.0 --port ${PORT:-8000}"]

server/__init__.py

@@ -0,0 +1,11 @@
+# Copyright (c) Meta Platforms, Inc. and affiliates.
+# All rights reserved.
+#
+# This source code is licensed under the BSD-style license found in the
+# LICENSE file in the root directory of this source tree.
+
+"""Persona Env environment server components."""
+
+from .persona_env_environment import PersonaEnvironment
+
+__all__ = ["PersonaEnvironment"]

server/app.py

@@ -0,0 +1,37 @@
+from openenv.core.env_server import create_app
+
+from models import PersonaAction, PersonaObservation
+from .persona_env_environment import PersonaEnvironment
+
+
+# = create one env instance we control (for debug visibility)
+_env = PersonaEnvironment()
+
+# = create the OpenEnv app (keeps /reset, /step, /schema, etc working)
+app = create_app(
+    PersonaEnvironment,
+    PersonaAction,
+    PersonaObservation,
+    env_name="persona_env",
+)
+
+# = add a debug endpoint that reflects the live env state we control
+@app.get("/debug_state")
+def debug_state():
+    return _env.state
+
+
+# Optional: also step/reset the debug env using the same action schema
+@app.post("/debug_reset")
+def debug_reset():
+    obs = _env.reset()
+    return {"observation": obs, "reward": getattr(obs, "reward", 0.0), "done": getattr(obs, "done", False)}
+
+
+@app.post("/debug_step")
+def debug_step(payload: dict):
+    # payload format matches your /step envelope: {"action": {...}}
+    action_dict = payload.get("action", {})
+    action = PersonaAction(**action_dict)
+    obs = _env.step(action)
+    return {"observation": obs, "reward": getattr(obs, "reward", 0.0), "done": getattr(obs, "done", False)}

server/k_persona_env_environment.py

@@ -0,0 +1,116 @@
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Dict
+from uuid import uuid4
+
+from openenv.core.env_server.interfaces import Environment
+from openenv.core.env_server.types import State
+
+from models import PersonaAction, PersonaObservation
+
+
+@dataclass
+class PersonaInternal:
+    # = stable attributes (placeholders)
+    star_sign: str = "Taurus"
+    character_type: str = "Sanguine-melancholic"
+    background: str = "Arts-adjacent, lower-middle class"
+
+    # = evolving state
+    mood: float = 0.1  # -> [-1, 1]
+    interests: Dict[str, float] = field(default_factory=lambda: {
+        "animal_welfare": 0.7,
+        "interior_design": 0.5,
+        "politics": 0.3,
+    })
+
+    def clamp(self) -> None:
+        self.mood = max(-1.0, min(1.0, self.mood))
+        for k, v in list(self.interests.items()):
+            self.interests[k] = max(0.0, min(1.0, v))
+
+
+class PersonaEnvironment(Environment):
+    def __init__(self):
+        self._state = State(episode_id=str(uuid4()), step_count=0)
+        self._p = PersonaInternal()
+
+    def reset(self) -> PersonaObservation:
+        self._state = State(episode_id=str(uuid4()), step_count=0)
+        self._p = PersonaInternal()
+        return PersonaObservation(
+            reaction_text="Persona initialised.",
+            mood=self._p.mood,
+            interests=dict(self._p.interests),
+            done=False,
+            reward=0.0,
+        )
+
+    def step(self, action: PersonaAction) -> PersonaObservation:
+        self._state.step_count += 1
+
+        if action.kind == "show_content":
+            reaction = self._apply_content(
+                topic=action.topic or "unknown",
+                source=action.source or "unknown",
+                valence=action.valence or "neutral",
+            )
+        elif action.kind == "ask_question":
+            reaction = self._answer_question(action.question or "")
+        elif action.kind == "advance_time":
+            reaction = self._advance_time(action.hours or 0)
+        else:
+            reaction = "Action rejected."
+
+        self._p.clamp()
+
+        return PersonaObservation(
+            reaction_text=reaction,
+            mood=self._p.mood,
+            interests=dict(self._p.interests),
+            done=False,
+            reward=0.0,
+        )
+
+    @property
+    def state(self) -> State:
+        return self._state
+
+    # = internal logic (simple, deterministic)
+
+    def _apply_content(self, topic: str, source: str, valence: str) -> str:
+        base = 0.02
+        if valence == "positive":
+            mood_delta = +0.05
+            interest_delta = +base
+        elif valence == "negative":
+            mood_delta = -0.05
+            interest_delta = +base / 2
+        else:
+            mood_delta = 0.0
+            interest_delta = +base / 4
+
+        if source in {"tabloid", "ragebait"}:
+            mood_delta -= 0.03
+        elif source in {"charity", "trusted"}:
+            mood_delta += 0.02
+
+        self._p.mood += mood_delta
+        self._p.interests[topic] = self._p.interests.get(topic, 0.2) + interest_delta
+
+        return f"Consumed {valence} content on {topic} from {source}. Mood {mood_delta:+.2f}."
+
+    def _answer_question(self, question: str) -> str:
+        if not question.strip():
+            return "No reaction."
+        top_interest = max(self._p.interests.items(), key=lambda kv: kv[1])[0]
+        return f"Answers via {top_interest}: '{question.strip()}'"
+
+    def _advance_time(self, hours: int) -> str:
+        hours = max(0, hours)
+        decay = min(0.2, hours / 240.0)
+        self._p.mood *= (1.0 - decay)
+        for k in list(self._p.interests.keys()):
+            self._p.interests[k] *= (1.0 - decay / 5.0)
+        return f"Advanced time by {hours}h."

server/persona_env_environment.py

@@ -0,0 +1,174 @@
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Dict, Optional, Tuple
+from uuid import uuid4
+
+from openenv.core.env_server.interfaces import Environment
+from openenv.core.env_server.types import State
+
+from models import PersonaAction, PersonaObservation
+
+
+@dataclass
+class PersonaInternal:
+    # = stable attributes (placeholders)
+    star_sign: str = "Taurus"
+    character_type: str = "Sanguine-melancholic"
+    background: str = "Arts-adjacent, lower-middle class"
+
+    # = evolving state
+    mood: float = 0.1  # -> [-1, 1]
+    interests: Dict[str, float] = field(default_factory=lambda: {
+        "animal_welfare": 0.7,
+        "interior_design": 0.5,
+        "politics": 0.3,
+    })
+
+    # = tiny memory (for continuity)
+    last_question: Optional[str] = None
+    last_topic: Optional[str] = None
+
+    def clamp(self) -> None:
+        self.mood = max(-1.0, min(1.0, self.mood))
+        for k, v in list(self.interests.items()):
+            self.interests[k] = max(0.0, min(1.0, v))
+
+    def bump_interest(self, topic: str, delta: float) -> None:
+        self.interests[topic] = self.interests.get(topic, 0.2) + delta
+
+
+class PersonaEnvironment(Environment):
+    def __init__(self):
+        self._state = State(episode_id=str(uuid4()), step_count=0)
+        self._p = PersonaInternal()
+
+    def reset(self) -> PersonaObservation:
+        self._state = State(episode_id=str(uuid4()), step_count=0)
+        self._p = PersonaInternal()
+        return PersonaObservation(
+            reaction_text="Persona initialised.",
+            mood=self._p.mood,
+            interests=dict(self._p.interests),
+            done=False,
+            reward=0.0,
+        )
+
+    def step(self, action: PersonaAction) -> PersonaObservation:
+        self._state.step_count += 1
+
+        if action.kind == "show_content":
+            reaction = self._apply_content(
+                topic=action.topic or "unknown",
+                source=action.source or "unknown",
+                valence=action.valence or "neutral",
+            )
+        elif action.kind == "ask_question":
+            reaction = self._answer_question(action.question or "")
+        elif action.kind == "advance_time":
+            reaction = self._advance_time(action.hours or 0)
+        else:
+            reaction = "Action rejected."
+
+        self._p.clamp()
+
+        return PersonaObservation(
+            reaction_text=reaction,
+            mood=self._p.mood,
+            interests=dict(self._p.interests),
+            done=False,
+            reward=0.0,
+        )
+
+    @property
+    def state(self) -> State:
+        return self._state
+
+    # = internal logic (simple, deterministic)
+
+    def _apply_content(self, topic: str, source: str, valence: str) -> str:
+        base = 0.02
+        if valence == "positive":
+            mood_delta = +0.05
+            interest_delta = +base
+        elif valence == "negative":
+            mood_delta = -0.05
+            interest_delta = +base / 2
+        else:
+            mood_delta = 0.0
+            interest_delta = +base / 4
+
+        if source in {"tabloid", "ragebait"}:
+            mood_delta -= 0.03
+        elif source in {"charity", "trusted"}:
+            mood_delta += 0.02
+
+        self._p.mood += mood_delta
+        self._p.bump_interest(topic, interest_delta)
+        self._p.last_topic = topic
+
+        return f"Consumed {valence} content on {topic} from {source}. Mood {mood_delta:+.2f}."
+
+    def _answer_question(self, question: str) -> str:
+        q = question.strip()
+        if not q:
+            return "No reaction."
+
+        self._p.last_question = q
+
+        # = keyword -> topic mapping
+        # -> This gives you a controllable, inspectable way to make questions influence state.
+        topic, mood_delta = self._infer_topic_and_mood_from_question(q)
+
+        if topic is not None:
+            # -> Questions increase attention to a topic a little.
+            self._p.bump_interest(topic, 0.015)
+            self._p.last_topic = topic
+
+        self._p.mood += mood_delta
+
+        top_interest = max(self._p.interests.items(), key=lambda kv: kv[1])[0]
+
+        # -> Mild continuity: reference last topic if available
+        if self._p.last_topic:
+            continuity = f" (recently thinking about {self._p.last_topic})"
+        else:
+            continuity = ""
+
+        return f"Answers via {top_interest}{continuity}: '{q}'"
+
+    def _infer_topic_and_mood_from_question(self, q: str) -> Tuple[Optional[str], float]:
+        ql = q.lower()
+
+        # -> Default: neutral mood change from being asked something
+        mood_delta = 0.0
+
+        # -> A few simple triggers
+        if any(w in ql for w in ["ethical", "cruelty", "welfare", "rescue", "animal"]):
+            return "animal_welfare", +0.01
+
+        if any(w in ql for w in ["decor", "interior", "furniture", "colour", "paint", "design"]):
+            return "interior_design", +0.01
+
+        if any(w in ql for w in ["election", "immigration", "tax", "government", "policy", "minister", "party"]):
+            # -> Politics questions tend to stress this persona slightly
+            return "politics", -0.01
+
+        return None, mood_delta
+
+    def _advance_time(self, hours: int) -> str:
+        hours = max(0, hours)
+        decay = min(0.2, hours / 240.0)
+
+        # -> mood drifts towards 0 with time
+        self._p.mood *= (1.0 - decay)
+
+        # -> interests slowly decay with time
+        for k in list(self._p.interests.keys()):
+            self._p.interests[k] *= (1.0 - decay / 5.0)
+
+        # -> very light memory fade
+        if hours >= 24:
+            self._p.last_question = None
+
+        return f"Advanced time by {hours}h."

server/requirements.txt

@@ -0,0 +1,6 @@
+openenv[core]>=0.2.0
+fastapi>=0.115.0
+uvicorn>=0.24.0
+
+
+