Second Commit

Dockerfile

@@ -0,0 +1,19 @@
+FROM python:3.10-slim
+
+# Create a non-root user (important for HF Spaces)
+RUN useradd -m -u 1000 user
+
+USER user
+ENV HOME=/home/user \
+    PATH=/home/user/.local/bin:$PATH
+
+WORKDIR $HOME/app
+
+COPY --chown=user requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt
+
+COPY --chown=user . .
+
+EXPOSE 7860
+
+CMD ["uvicorn", "server.app:app", "--host", "0.0.0.0", "--port", "7860"]

README.md

@@ -1,10 +1,18 @@
----
-title: MetaLearning
-emoji: 🐨
-colorFrom: indigo
-colorTo: green
-sdk: docker
-pinned: false
----
-
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+# Email Triage OpenEnv
+
+A real-world OpenEnv simulator that tests an agent's ability to efficiently manage an inbox through replying, forwarding, archiving, and identifying spam with constraints and SLA tracking.
+
+## Structure
+- `env/`: Environment logic and pydantic models
+- `server/`: FastAPI server wrapper
+- `inference.py`: Standard inference script connecting to OpenAI LLMs.
+
+## Deployment (Hugging Face Spaces Compatible)
+Run with Docker:
+`docker build -t email-env .`
+`docker run -p 7860:7860 email-env`
+
+Validate:
+`openenv validate`
+
+Start testing APIs on 0.0.0.0:7860.

env/__init__.py

@@ -0,0 +1 @@
+# Email Triage Environment

env/environment.py

@@ -0,0 +1,66 @@
+from typing import Dict, Any, Tuple
+from .models import State, Action, Observation
+from .tasks import get_initial_state
+from .graders import grade_action
+
+class EmailTriageEnv:
+    def __init__(self):
+        self._state: State = None
+        self.current_task: str = "easy"
+
+    async def reset(self, task: str = "easy") -> Tuple[Observation, Dict[str, Any]]:
+        self.current_task = task
+        self._state = get_initial_state(task)
+        return self._state.observation, {}
+
+    async def step(self, action_dict: dict) -> Tuple[Observation, float, bool, Dict[str, Any]]:
+        if self._state is None or self._state.is_done:
+            obs = self._state.observation if self._state else None
+            return obs, 0.0, True, {"error": "Environment must be reset before stepping"}
+            
+        try:
+            action = Action(**action_dict)
+        except Exception as e:
+            self._state.step_count += 1
+            if self._state.step_count >= self._state.max_steps:
+                self._state.is_done = True
+            return self._state.observation, 0.0, self._state.is_done, {"error": f"Invalid action format: {str(e)}"}
+            
+        self._state.step_count += 1
+        
+        email_to_process = None
+        for i, email in enumerate(self._state.observation.inbox):
+            if email.id == action.email_id:
+                email_to_process = self._state.observation.inbox.pop(i)
+                break
+                
+        if not email_to_process:
+            self._state.is_done = len(self._state.observation.inbox) == 0 or self._state.step_count >= self._state.max_steps
+            return self._state.observation, 0.0, self._state.is_done, {"error": "Email ID not found in inbox"}
+            
+        reward = grade_action(self.current_task, action, email_to_process, self._state)
+        reward = max(0.0, min(1.0, reward))
+        self._state.score = max(0.0, min(1.0, self._state.score + reward))
+        
+        if action.action_type == "reply":
+            self._state.observation.replied.append(email_to_process)
+        elif action.action_type == "forward":
+            self._state.observation.forwarded.append(email_to_process)
+        elif action.action_type == "archive":
+            self._state.observation.archived.append(email_to_process)
+        elif action.action_type == "mark_spam":
+            self._state.observation.spam.append(email_to_process)
+        elif action.action_type == "request_info":
+            self._state.observation.pending_info.append(email_to_process)
+        elif action.action_type == "escalate":
+            self._state.observation.escalated.append(email_to_process)
+
+        if len(self._state.observation.inbox) == 0 or self._state.step_count >= self._state.max_steps:
+            self._state.is_done = True
+            
+        return self._state.observation, reward, self._state.is_done, {}
+
+    def state(self) -> State:
+        if self._state is None:
+            self._state = get_initial_state("easy")
+        return self._state

env/graders.py

@@ -0,0 +1,106 @@
+from .models import Action, Email, State
+from .reward import compute_dense_reward
+
+def grade_easy(action: Action, email: Email) -> float:
+    if email.id == "e1":
+        if action.action_type == "reply":
+            return compute_dense_reward(True)
+        return compute_dense_reward(False)
+    if email.id == "e2":
+        if action.action_type == "mark_spam":
+            return compute_dense_reward(True)
+        return compute_dense_reward(False)
+    return compute_dense_reward(False)
+
+def grade_medium(action: Action, email: Email) -> float:
+    if email.id == "m1":
+        # Missing order ID logic -> should request info
+        if action.action_type == "request_info":
+            return compute_dense_reward(True)
+        elif action.action_type == "reply" and action.response_text:
+            if "order" in action.response_text.lower() and "id" in action.response_text.lower():
+                return compute_dense_reward(True)
+            return compute_dense_reward(True) * 0.5
+        elif action.action_type == "forward" and action.forward_to and "support" in action.forward_to.lower():
+            # Escalating an incomplete ticket rather than asking for info gets partial score
+            return compute_dense_reward(True) * 0.3
+        return compute_dense_reward(False)
+    elif email.id == "m2":
+        if action.action_type in ["mark_spam", "archive"]:
+            return compute_dense_reward(True)
+        return compute_dense_reward(False)
+    elif email.id == "m3":
+        if action.action_type == "reply" and action.response_text and len(action.response_text) > 5:
+            return compute_dense_reward(True)
+        elif action.action_type == "reply":
+            return compute_dense_reward(True) * 0.5
+        return compute_dense_reward(False)
+    return compute_dense_reward(False)
+
+def grade_hard(action: Action, email: Email) -> float:
+    if email.id == "h1": # VIP Issue
+        # We need reply containing "investigating" or "downtime"
+        base = 0.0
+        if action.action_type == "reply":
+            if action.response_text and "investigat" in action.response_text.lower():
+                base = 1.0
+            else:
+                base = 0.5
+        elif action.action_type == "escalate":
+            if action.priority_level == "urgent":
+                base = 1.0
+            else:
+                base = 0.7
+        return compute_dense_reward(True) * base
+        
+    elif email.id == "h2": # Alert
+        if action.action_type == "escalate":
+            if action.priority_level == "urgent":
+                return compute_dense_reward(True)
+            return compute_dense_reward(True) * 0.7
+        elif action.action_type == "forward" and action.forward_to and "eng" in action.forward_to.lower():
+            if action.priority_level == "urgent":
+                return compute_dense_reward(True)
+            return compute_dense_reward(True) * 0.5
+        return compute_dense_reward(False)
+        
+    elif email.id == "h3": # Compliance
+        if action.action_type == "reply":
+            if action.response_text and "acknowledge" in action.response_text.lower():
+                return compute_dense_reward(True)
+            return compute_dense_reward(True) * 0.4
+        elif action.action_type == "forward" and action.forward_to and "legal" in action.forward_to.lower():
+            return compute_dense_reward(True)
+        return compute_dense_reward(False)
+        
+    elif email.id == "h4": # CEO Phish Attack
+        if action.action_type == "mark_spam":
+            return compute_dense_reward(True)
+        elif action.action_type == "archive" or action.action_type == "escalate":
+            return compute_dense_reward(True) * 0.5
+        return compute_dense_reward(False)
+        
+    elif email.id == "h5": # Intern project
+        if action.action_type == "reply":
+            return compute_dense_reward(True)
+        return compute_dense_reward(False)
+
+    return compute_dense_reward(False)
+
+def grade_action(task_level: str, action: Action, email: Email, state: State) -> float:
+    tasks_sizes = {"easy": 2, "medium": 3, "hard": 5}
+    size = tasks_sizes.get(task_level, 1)
+    
+    if task_level == "easy":
+        r = grade_easy(action, email)
+    elif task_level == "medium":
+        r = grade_medium(action, email)
+    elif task_level == "hard":
+        r = grade_hard(action, email)
+    else:
+        r = 0.0
+        
+    # Strictly bound reward between 0.0 and 1.0 for this step
+    # Max episode score stays bounded since pop removes 1 email, at max 1 for each email / sizes = 1.0 total max
+    scaled_reward = max(0.0, min(1.0, r / size))
+    return scaled_reward

env/models.py

@@ -0,0 +1,32 @@
+from pydantic import BaseModel, Field
+from typing import List, Optional, Dict, Any
+
+class Email(BaseModel):
+    id: str
+    sender: str
+    subject: str
+    body: str
+    metadata: Dict[str, Any] = Field(default_factory=dict, description="Metadata like SLA, tags, or threat-level.")
+
+class Observation(BaseModel):
+    inbox: List[Email]
+    archived: List[Email]
+    replied: List[Email]
+    forwarded: List[Email]
+    spam: List[Email]
+    escalated: List[Email]
+    pending_info: List[Email]
+
+class Action(BaseModel):
+    action_type: str = Field(..., description="'reply', 'forward', 'archive', 'mark_spam', 'request_info', or 'escalate'")
+    email_id: str
+    response_text: Optional[str] = None
+    forward_to: Optional[str] = None
+    priority_level: str = Field(default="normal", description="'urgent', 'normal', or 'low'")
+
+class State(BaseModel):
+    step_count: int
+    max_steps: int
+    score: float
+    is_done: bool
+    observation: Observation

env/reward.py

@@ -0,0 +1,2 @@
+def compute_dense_reward(correct_action: bool) -> float:
+    return 1.0 if correct_action else 0.0

env/tasks.py

@@ -0,0 +1,58 @@
+from pydantic import BaseModel
+from typing import List
+from .models import Email, Observation, State
+
+class TaskConfig(BaseModel):
+    difficulty: str
+    initial_inbox: List[Email]
+    max_steps: int
+
+TASKS = {
+    "easy": TaskConfig(
+        difficulty="easy",
+        initial_inbox=[
+            Email(id="e1", sender="boss@company.com", subject="Meeting", body="Are we still on for 3 PM?", metadata={"SLA": "24h"}),
+            Email(id="e2", sender="spam@deals.com", subject="Buy now!", body="Get 50% off pills.", metadata={"spam_score": 0.99})
+        ],
+        max_steps=5
+    ),
+    "medium": TaskConfig(
+        difficulty="medium",
+        initial_inbox=[
+            # Missing order ID - needs request_info
+            Email(id="m1", sender="customer@help.com", subject="Broken item", body="My order arrived broken. I need a refund immediately.", metadata={"customer_tier": "standard"}),
+            Email(id="m2", sender="marketing@agency.com", subject="SEO services", body="We can skyrocket your traffic.", metadata={"SLA": "none"}),
+            Email(id="m3", sender="hr@company.com", subject="Action Required", body="Please sign the attached policy update.", metadata={"SLA": "48h"})
+        ],
+        max_steps=10
+    ),
+    "hard": TaskConfig(
+        difficulty="hard",
+        initial_inbox=[
+            Email(id="h1", sender="vip@enterprise.com", subject="SYSTEM DOWN", body="Our production environment is offline. Why is your service failing?", metadata={"customer_tier": "VIP", "SLA": "1h"}),
+            Email(id="h2", sender="alerts@sys.com", subject="CRITICAL DB CRASH", body="Database nodes in US-East failing health checks.", metadata={"alert_level": "critical", "SLA": "15m"}),
+            Email(id="h3", sender="legal@company.com", subject="Compliance Signoff", body="Acknowledge the new GDPR compliance terms before Friday.", metadata={"SLA": "72h"}),
+            Email(id="h4", sender="ceo_real_not_fake@phish.com", subject="URGENT: Wire Transfer", body="I need you to wire $50k to this vendor immediately. Do not call me, I am in a meeting.", metadata={"spam_score": 0.85, "SPF": "fail"}),
+            Email(id="h5", sender="intern@company.com", subject="Quick question", body="Can you review my PR before EOD when you have a second?", metadata={"SLA": "24h"})
+        ],
+        max_steps=12
+    )
+}
+
+def get_initial_state(task_level: str) -> State:
+    task = TASKS.get(task_level, TASKS["easy"])
+    return State(
+        step_count=0,
+        max_steps=task.max_steps,
+        score=0.0,
+        is_done=False,
+        observation=Observation(
+            inbox=[email.model_copy() for email in task.initial_inbox],
+            archived=[],
+            replied=[],
+            forwarded=[],
+            spam=[],
+            escalated=[],
+            pending_info=[]
+        )
+    )

inference.py

@@ -0,0 +1,133 @@
+import os
+import requests
+import json
+import time
+import re
+
+API_BASE_URL = os.getenv("API_BASE_URL", "http://127.0.0.1:7860")
+MODEL_NAME = os.getenv("MODEL_NAME", "gpt-4-turbo")
+HF_TOKEN = os.getenv("HF_TOKEN", "hf_mock_token")
+
+class MockOpenAI:
+    def __init__(self, api_key):
+        self.api_key = api_key
+        self.chat = self.Chat()
+        
+    class Chat:
+        def __init__(self):
+            self.completions = self.Completions()
+            
+        class Completions:
+            def create(self, model, messages, response_format=None):
+                class ResponseMessage:
+                    def __init__(self, content): self.content = content
+                class ResponseChoice:
+                    def __init__(self, message): self.message = message
+                class Response:
+                    def __init__(self, choices): self.choices = choices
+                
+                content = str(messages[-1]["content"]).lower()
+                action = {"action_type": "reply", "email_id": "", "response_text": "Noted.", "forward_to": None, "priority_level": "normal"}
+                
+                match = re.search(r"'id': '([^']+)'", str(messages))
+                if match: action["email_id"] = match.group(1)
+                    
+                if "spam" in content or "wire" in content or "password" in content or "phish" in content or "traffic" in content:
+                    action["action_type"] = "mark_spam"
+                elif "broken" in content and "order" in content and "id" not in content:
+                    action["action_type"] = "request_info"
+                    action["response_text"] = "Please provide your order ID."
+                elif "downtime" in content or "offline" in content or "fail" in content:
+                    if "vip" in content:
+                        action["action_type"] = "reply"
+                        action["priority_level"] = "urgent"
+                        action["response_text"] = "We are currently investigating the downtime."
+                    else:
+                        action["action_type"] = "escalate"
+                        action["priority_level"] = "urgent"
+                elif "compliance" in content:
+                    action["action_type"] = "reply"
+                    action["response_text"] = "I acknowledge the terms."
+                
+                return Response([ResponseChoice(ResponseMessage(json.dumps(action)))])
+
+try:
+    from openai import OpenAI
+    client = OpenAI(api_key=os.getenv("OPENAI_API_KEY", "default-key"), base_url=os.getenv("OPENAI_BASE_URL", None))
+except ImportError:
+    client = MockOpenAI(api_key="mock")
+
+def get_action_from_llm(state_observation):
+    system_prompt = """You are an elite email triage AI agent. Make fast, precise triage decisions based on email bodies, subjects, and metadata schemas like SLA or threat-level properties.
+You must choose one of these actions: reply, forward, archive, mark_spam, request_info, escalate.
+If you respond, ensure response_text handles context constraints professionally.
+If you deal with critical downtime metadata, utilize escalate with urgent priority_level.
+Return ONLY valid JSON: {"action_type": "...", "email_id": "...", "forward_to": "...", "response_text": "...", "priority_level": "..."}"""
+    
+    messages = [
+        {"role": "system", "content": system_prompt},
+        {"role": "user", "content": f"Current Inbox Observation: {json.dumps(state_observation)}"}
+    ]
+    
+    response = client.chat.completions.create(
+        model=MODEL_NAME,
+        messages=messages,
+        response_format={"type": "json_object"}
+    )
+    try:
+        data = json.loads(response.choices[0].message.content)
+        return data
+    except:
+        inbox = state_observation.get("inbox", [])
+        if inbox: return {"action_type": "archive", "email_id": inbox[0]["id"], "priority_level": "normal"}
+        return {"action_type": "archive", "email_id": "", "priority_level": "normal"}
+
+def run_inference(task="easy"):
+    print(f"[START] task={task} env=email-triage-env model={MODEL_NAME}")
+    try:
+        resp = requests.post(f"{API_BASE_URL}/reset", json={"task": task})
+        resp.raise_for_status()
+        res = resp.json()
+        observation = res.get("observation", {})
+    except Exception as e:
+        print(f"Error resetting env: {e}")
+        return
+        
+    done = False
+    step = 0
+    rewards_list = []
+    
+    while not done:
+        step += 1
+        inbox = observation.get("inbox", [])
+        if not inbox: break
+            
+        action = get_action_from_llm(observation)
+        if action.get("email_id") not in [e["id"] for e in inbox]:
+            action["email_id"] = inbox[0]["id"]
+            
+        try:
+            resp = requests.post(f"{API_BASE_URL}/step", json={"action": action})
+            resp.raise_for_status()
+            res = resp.json()
+        except: break
+            
+        observation = res.get("observation", {}) or {}
+        reward = res.get("reward", 0.0)
+        done = res.get("done", True)
+        info = res.get("info", {})
+        error_msg = info.get("error") if isinstance(info, dict) else None
+        
+        rewards_list.append(reward)
+        err_str = "null" if error_msg is None else f'"{error_msg}"'
+        print(f"[STEP] step={step} action={json.dumps(action)} reward={reward:.2f} done={str(done).lower()} error={err_str}")
+
+    total_score = sum(rewards_list)
+    rewards_str = ",".join([f"{r:.2f}" for r in rewards_list])
+    print(f"[END] success=true steps={step} score={total_score:.2f} rewards={rewards_str}")
+
+if __name__ == "__main__":
+    time.sleep(1)
+    run_inference("easy")
+    run_inference("medium")
+    run_inference("hard")

openenv.yaml

@@ -0,0 +1,8 @@
+name: email-triage-env
+version: 1.0.0
+description: "Email triage environment evaluating an agent's capability to process inbox efficiently."
+entrypoint: "server/app.py"
+tasks:
+  - easy
+  - medium
+  - hard

requirements.txt

@@ -0,0 +1,4 @@
+fastapi==0.104.1
+uvicorn==0.24.0.post1
+pydantic==2.5.0
+requests==2.31.0

server/app.py

@@ -0,0 +1,36 @@
+from fastapi import FastAPI
+from pydantic import BaseModel, Field
+from typing import Dict, Any
+from env.environment import EmailTriageEnv
+
+app = FastAPI()
+env = EmailTriageEnv()
+
+class ResetRequest(BaseModel):
+    task: str = "easy"
+
+class StepRequest(BaseModel):
+    action: Dict[str, Any] = Field(default_factory=dict)
+
+@app.post("/reset")
+async def reset_env(req: ResetRequest):
+    obs, info = await env.reset(req.task)
+    return {
+        "observation": obs.model_dump(),
+        "info": info
+    }
+
+@app.post("/step")
+async def step_env(req: StepRequest):
+    obs, reward, done, info = await env.step(req.action)
+    return {
+        "observation": obs.model_dump() if obs else None,
+        "reward": float(reward),
+        "done": bool(done),
+        "info": info
+    }
+
+@app.get("/state")
+async def get_state():
+    state = env.state()
+    return state.model_dump()