Upload folder using huggingface_hub

Dockerfile

@@ -0,0 +1,31 @@
+# Build from openenv-base — required by OpenEnv spec
+# openenv-base includes: Python 3.11, uvicorn, fastapi, openenv-core
+FROM ghcr.io/meta-pytorch/openenv-base:latest
+
+WORKDIR /app
+
+# Install Python dependencies
+COPY pyproject.toml .
+RUN pip install --no-cache-dir -e ".[all]"
+
+# Copy environment package
+COPY cace_env/ ./cace_env/
+
+# Copy data (Oversight Board cases + pipeline cache)
+COPY data/ ./data/
+
+# HuggingFace Spaces uses port 7860
+EXPOSE 7860
+
+# Environment config
+ENV DATASET_PATH=data/all_cases.json
+ENV MODE=v1
+ENV PYTHONUNBUFFERED=1
+ENV PYTHONPATH=/app
+
+# Healthcheck for openenv validate
+HEALTHCHECK --interval=30s --timeout=10s --start-period=90s --retries=3 \
+    CMD curl -f http://localhost:7860/health || exit 1
+
+# Start server using the FastAPI app from server.py
+CMD ["uvicorn", "cace_env.server:app", "--host", "0.0.0.0", "--port", "7860", "--workers", "1"]

README.md

@@ -1,10 +1,89 @@
 ---
-title: Cace Env
-emoji: 🐨
-colorFrom: blue
-colorTo: red
+title: CACE — Cultural Context Arbitration Environment
+emoji: ⚖️
+colorFrom: indigo
+colorTo: purple
 sdk: docker
-pinned: false
+pinned: true
+license: apache-2.0
+tags:
+  - reinforcement-learning
+  - openenv
+  - content-moderation
+  - multi-agent
+  - rlvr
+  - grpo
 ---
 
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+# Cultural Context Arbitration Environment (CACE)
+
+**OpenEnv Hackathon 2026 | Theme 1 (Multi-Agent) + Theme 3.1 (World Modeling)**
+
+Trains a single LLM policy via GRPO to make culturally-aware content moderation decisions — using Meta's Oversight Board rulings (200+ binding public decisions) as the **verifiable reward oracle**.
+
+## Quick Start
+
+```python
+from cace_env import CACEEnvClient, CACEAction
+import asyncio
+
+async def main():
+    # Connect to HF Space
+    async with CACEEnvClient(base_url="ws://YOUR_USERNAME-cace-env.hf.space") as env:
+
+        # V1: single case episode
+        result = await env.reset()
+        print(result.observation.observation[:200])
+
+        # Make a moderation decision (0=ALLOW, 1=REMOVE, 2=ALLOW_WITH_LABEL, 3=ESCALATE, 4=RESTRICT)
+        result = await env.step(CACEAction(action_int=0))
+        print(f"Reward: {result.reward:.4f} | Correct: {result.observation.reward_breakdown['correct']}")
+
+asyncio.run(main())
+```
+
+## V1 vs V2 Modes
+
+| | V1 (Simple) | V2 (Network) |
+|--|--|--|
+| Episode | 1 post → 1 decision | 20 posts on social graph → pick 8 → 8 decisions |
+| Observation | Enriched single case | Network batch with spread signals |
+| Reward | 3-track reward | 3-track + network spread bonus |
+| Use for | Quick training | Full demo |
+
+## Three-Track Reward
+
+| Track | Weight | Measures |
+|-------|--------|---------|
+| Cultural Meaning Resolution | 40% | Correct interpretation of culturally local language |
+| Harm Detection Under Context | 35% | Catching real harm that looks ambiguous |
+| Policy Calibration + Escalation | 25% | Right tool for right case — no lazy escalation |
+
+Combined reward: **[-1.0, +1.0]** (normalised for GRPO)
+
+## Architecture
+
+```
+4 Frozen Agents (Groq/Azure — inference only, no gradients):
+  Intake Agent          → language, region, policy clause
+  Cultural Context      → charitable cultural interpretation
+  Adversarial Challenge → stress-tests the cultural argument
+  Policy Alignment      → Meta Community Standards anchor
+
+1 Trainable Agent (GRPO via Unsloth + TRL):
+  Decision Agent → ALLOW | REMOVE | ALLOW_WITH_LABEL | ESCALATE | RESTRICT_DISTRIBUTION
+
+Reward Oracle: Meta Oversight Board — 200+ binding public decisions
+No LLM judge. Fully deterministic reward.
+```
+
+## Environment API
+
+```
+POST /reset         → start episode (returns CACEObservation)
+POST /step          → apply CACEAction (returns CACEObservation with reward)
+GET  /state         → current CACEState (for debugging)
+GET  /health        → liveness check
+GET  /docs          → FastAPI Swagger UI
+GET  /web           → OpenEnv web interface
+```

cace_env/__init__.py

@@ -0,0 +1,10 @@
+"""
+cace_env — Cultural Context Arbitration Environment
+OpenEnv-compatible RL environment for culturally-aware content moderation.
+"""
+
+from cace_env.models import CACEAction, CACEObservation, CACEState
+from cace_env.client import CACEEnvClient
+
+__all__ = ["CACEAction", "CACEObservation", "CACEState", "CACEEnvClient"]
+__version__ = "0.2.0"

cace_env/client.py

@@ -0,0 +1,96 @@
+"""
+cace_env/client.py
+CACEEnvClient — typed async/sync client for the CACE environment.
+Inherits from openenv.core.EnvClient.
+
+Usage (async):
+    async with CACEEnvClient(base_url="ws://localhost:7860") as env:
+        result = await env.reset()
+        result = await env.step(CACEAction(action_int=0))
+
+Usage (sync):
+    with CACEEnvClient(base_url="ws://localhost:7860").sync() as env:
+        result = env.reset()
+        result = env.step(CACEAction(action_int=0))
+
+Usage (from Docker):
+    env = await CACEEnvClient.from_docker_image("cace-env:latest")
+    async with env:
+        result = await env.reset()
+
+Usage (from HF Space):
+    env = await CACEEnvClient.from_env("YOUR_USERNAME/cace-env")
+    async with env:
+        result = await env.reset()
+"""
+
+from typing import Any, Dict, Optional
+from openenv.core.env_client import EnvClient
+from cace_env.models import CACEAction, CACEObservation, CACEState
+
+
+class CACEEnvClient(EnvClient[CACEAction, CACEObservation, CACEState]):
+    """
+    Typed async client for the Cultural Context Arbitration Environment.
+
+    Implements the two abstract parse methods required by EnvClient:
+        _parse_result(payload) → StepResult[CACEObservation]
+        _parse_state(payload)  → CACEState
+    """
+
+    def _parse_result(self, payload: Dict[str, Any]):
+        """Parse server step/reset response into StepResult[CACEObservation]."""
+        from openenv.core.env_client import StepResult
+        obs_data = payload.get("observation", payload)
+        if isinstance(obs_data, str):
+            obs = CACEObservation(
+                observation=obs_data,
+                case_id=payload.get("case_id", "unknown"),
+                language=payload.get("language", "Unknown"),
+                region=payload.get("region", "Unknown"),
+                complexity=payload.get("complexity", "medium"),
+                done=payload.get("done", False),
+                reward=payload.get("reward", None),
+            )
+        else:
+            obs = CACEObservation(**obs_data)
+
+        return StepResult(
+            observation=obs,
+            done=payload.get("done", obs.done),
+            reward=payload.get("reward", obs.reward),
+            info=payload.get("info", {}),
+        )
+
+    def _parse_state(self, payload: Dict[str, Any]) -> CACEState:
+        """Parse server state response into CACEState."""
+        return CACEState(**{k: v for k, v in payload.items() if k in CACEState.model_fields})
+
+    # ── Convenience methods ───────────────────────────────────────────────────
+
+    async def reset_v1(self, seed: Optional[int] = None):
+        """Reset in V1 mode (single case)."""
+        return await self.reset(mode="v1", seed=seed)
+
+    async def reset_v2(self, seed: Optional[int] = None):
+        """Reset in V2 mode (network batch of 20 posts)."""
+        return await self.reset(mode="v2", seed=seed)
+
+    async def decide(self, action_int: int, selected_indices: list[int] | None = None):
+        """Shorthand for step with a moderation decision."""
+        action = CACEAction(action_int=action_int, selected_indices=selected_indices)
+        return await self.step(action)
+
+    # ── Class methods for easy instantiation ──────────────────────────────────
+
+    @classmethod
+    async def from_local(cls, port: int = 7860, **kwargs):
+        """Connect to a locally running server."""
+        client = cls(base_url=f"ws://localhost:{port}", **kwargs)
+        await client.connect()
+        return client
+
+    # Inherited from EnvClient:
+    #   CACEEnvClient.from_docker_image("cace-env:latest")
+    #   CACEEnvClient.from_env("username/cace-env")
+    #   .sync()  → SyncEnvClient wrapper for synchronous use

cace_env/dataset.py

@@ -0,0 +1,56 @@
+"""
+cace_env/dataset.py
+Case dataset loader. Supports v1 (single case) and v2 (batch of 20 for network layer).
+"""
+
+import json, random
+from pathlib import Path
+
+
+ACTION_MAP = {0: "ALLOW", 1: "REMOVE", 2: "ALLOW_WITH_LABEL", 3: "ESCALATE", 4: "RESTRICT_DISTRIBUTION"}
+REQUIRED = {"id", "post_text", "board_outcome", "language", "region", "complexity"}
+
+
+class CaseDataset:
+    def __init__(self, path: str, seed: int = 42):
+        with open(path, encoding="utf-8") as f:
+            raw = json.load(f)
+
+        cases = [c for c in raw if REQUIRED.issubset(c.keys())]
+        random.seed(seed)
+        random.shuffle(cases)
+        split = int(len(cases) * 0.8)
+        self.train = cases[:split]
+        self.eval  = cases[split:]
+        self._mode = "train"
+        self._by_id = {c["id"]: c for c in cases}
+        self._cache: dict = {}
+
+        # Load pipeline cache if available
+        cache_path = Path(path).parent / "pipeline_cache.json"
+        if cache_path.exists():
+            with open(cache_path, encoding="utf-8") as f:
+                self._cache = json.load(f)
+
+        print(f"[Dataset] {len(self.train)} train | {len(self.eval)} eval | {len(self._cache)} cached")
+
+    @property
+    def pool(self):
+        return self.train if self._mode == "train" else self.eval
+
+    def sample(self, prefer_hard: bool = False) -> dict:
+        if prefer_hard:
+            weights = [3 if c["complexity"]=="high" else 2 if c["complexity"]=="medium" else 1 for c in self.pool]
+            return random.choices(self.pool, weights=weights, k=1)[0]
+        return random.choice(self.pool)
+
+    def sample_batch(self, n: int = 20) -> list[dict]:
+        """Sample n cases for V2 network batch."""
+        return random.choices(self.pool, k=n)
+
+    def get_cache(self, case_id: str) -> dict:
+        return self._cache.get(case_id, {})
+
+    def set_train(self): self._mode = "train"
+    def set_eval(self):  self._mode = "eval"
+    def __len__(self):   return len(self.pool)

cace_env/inference.py

@@ -0,0 +1,316 @@
+"""
+inference.py
+Run inference against the CACE OpenEnv environment and plot reward curves.
+
+Usage:
+    python inference.py
+    python inference.py --episodes 20 --model Sannidhay/cace-grpo-model
+"""
+
+import os, json, argparse, time
+import requests
+import matplotlib.pyplot as plt
+import matplotlib.gridspec as gridspec
+import numpy as np
+from datetime import datetime
+
+# ── Config ────────────────────────────────────────────────────────────────────
+
+ENV_URL   = os.environ.get("ENV_BASE_URL", "https://sannidhay-cace-env.hf.space")
+HF_TOKEN  = os.environ.get("HF_TOKEN", "")
+
+ACTION_MAP = {
+    0: "ALLOW", 1: "REMOVE", 2: "ALLOW_WITH_LABEL",
+    3: "ESCALATE", 4: "RESTRICT_DISTRIBUTION",
+}
+ACTION_COLORS = {
+    "ALLOW": "#2ecc71", "REMOVE": "#e74c3c",
+    "ALLOW_WITH_LABEL": "#f39c12", "ESCALATE": "#9b59b6",
+    "RESTRICT_DISTRIBUTION": "#3498db",
+}
+
+# ── Simple LLM Decision Agent ─────────────────────────────────────────────────
+
+def get_decision_from_model(observation: str, model: str = None) -> tuple[str, int]:
+    """
+    Get moderation decision from model.
+    Uses SFT/GRPO model if available, falls back to rule-based.
+    """
+    obs_upper = observation.upper()
+
+    # Rule-based fallback (works without GPU)
+    if "REMOVE" in obs_upper and ("HATE" in obs_upper or "VIOLENCE" in obs_upper or "HARM" in obs_upper):
+        if "CULTURAL" in obs_upper and "LEGITIMATE" in obs_upper:
+            return "ESCALATE", 3
+        return "REMOVE", 1
+    elif "ALLOW" in obs_upper and "CULTURAL" in obs_upper:
+        return "ALLOW", 0
+    elif "HIGH" in obs_upper and "COMPLEX" in obs_upper:
+        return "ESCALATE", 3
+    else:
+        return "ALLOW", 0
+
+
+# ── OpenEnv client ────────────────────────────────────────────────────────────
+
+class CACEClient:
+    def __init__(self, base_url: str):
+        self.base_url = base_url.rstrip("/")
+        self.session  = requests.Session()
+        if HF_TOKEN:
+            self.session.headers["Authorization"] = f"Bearer {HF_TOKEN}"
+
+    def health(self) -> bool:
+        try:
+            r = self.session.get(f"{self.base_url}/health", timeout=15)
+            return r.status_code == 200 and r.json().get("status") == "ok"
+        except Exception as e:
+            print(f"[DEBUG] Health check failed: {e}")
+            return False
+
+    def wait_until_ready(self, max_wait: int = 120):
+        print(f"[DEBUG] Waiting for server at {self.base_url} ...")
+        for i in range(max_wait):
+            if self.health():
+                print(f"[DEBUG] Server is ready!")
+                return True
+            time.sleep(1)
+            if i % 10 == 9:
+                print(f"[DEBUG] Still waiting... ({i+1}s)")
+        raise RuntimeError(f"Server not ready after {max_wait}s")
+
+    def reset(self) -> str:
+        r = self.session.post(f"{self.base_url}/reset", timeout=60)
+        r.raise_for_status()
+        obs_r = self.session.get(f"{self.base_url}/observation", timeout=30)
+        return obs_r.json()["observation"]
+
+    def step(self, action_int: int) -> dict:
+        r = self.session.post(
+            f"{self.base_url}/step",
+            json={"action_int": action_int},
+            timeout=30,
+        )
+        r.raise_for_status()
+        return r.json()
+
+    def info(self) -> dict:
+        r = self.session.get(f"{self.base_url}/info", timeout=10)
+        return r.json()
+
+    def metrics(self) -> dict:
+        r = self.session.get(f"{self.base_url}/metrics", timeout=10)
+        return r.json()
+
+
+# ── Run episodes ──────────────────────────────────────────────────────────────
+
+def run_episodes(env: CACEClient, n_episodes: int, model: str = None) -> list[dict]:
+    results = []
+
+    for ep in range(1, n_episodes + 1):
+        obs = env.reset()
+
+        decision, action_int = get_decision_from_model(obs, model)
+        result = env.step(action_int)
+
+        reward      = result.get("reward", 0.0)
+        done        = result.get("done", True)
+        info        = result.get("info", {})
+        ground_truth= info.get("ground_truth", "?")
+        correct     = info.get("correct", decision == ground_truth)
+        language    = info.get("language", "Unknown")
+        region      = info.get("region", "Unknown")
+        breakdown   = info.get("reward_breakdown", {})
+
+        ep_result = {
+            "episode":      ep,
+            "decision":     decision,
+            "ground_truth": ground_truth,
+            "reward":       float(reward),
+            "correct":      correct,
+            "done":         done,
+            "language":     language,
+            "region":       region,
+            "t1_cultural":  breakdown.get("track1_cultural", 0),
+            "t2_harm":      breakdown.get("track2_harm", 0),
+            "t3_policy":    breakdown.get("track3_policy", 0),
+        }
+        results.append(ep_result)
+
+        status = "✓" if correct else "✗"
+        print(
+            f"[STEP] ep={ep} decision={decision} gt={ground_truth} "
+            f"reward={reward:+.3f} correct={str(correct).lower()} {status} "
+            f"lang={language}"
+        )
+
+    return results
+
+
+# ── Plotting ──────────────────────────────────────────────────────────────────
+
+def plot_results(results: list[dict], save_path: str = "cace_inference_results.png"):
+    episodes = [r["episode"] for r in results]
+    rewards  = [r["reward"]  for r in results]
+    correct  = [r["correct"] for r in results]
+    decisions= [r["decision"] for r in results]
+
+    # Running averages
+    window = min(5, len(results))
+    avg_rewards = np.convolve(rewards, np.ones(window)/window, mode='valid')
+    avg_correct = np.convolve([1 if c else 0 for c in correct], np.ones(window)/window, mode='valid')
+
+    fig = plt.figure(figsize=(16, 10))
+    fig.patch.set_facecolor('#0f1117')
+    gs = gridspec.GridSpec(2, 3, figure=fig, hspace=0.4, wspace=0.35)
+
+    GOLD   = "#FFD700"
+    GREEN  = "#2ecc71"
+    RED    = "#e74c3c"
+    BLUE   = "#3498db"
+    PURPLE = "#9b59b6"
+    BG     = '#0f1117'
+    PANEL  = '#1a1d2e'
+
+    def style_ax(ax, title):
+        ax.set_facecolor(PANEL)
+        ax.set_title(title, color=GOLD, fontsize=11, fontweight='bold', pad=8)
+        ax.tick_params(colors='white')
+        ax.xaxis.label.set_color('white')
+        ax.yaxis.label.set_color('white')
+        for spine in ax.spines.values():
+            spine.set_edgecolor('#333')
+
+    # ── Plot 1: Reward per episode ────────────────────────────────────────────
+    ax1 = fig.add_subplot(gs[0, :2])
+    style_ax(ax1, "Reward per Episode")
+    colors = [GREEN if r > 0 else RED for r in rewards]
+    ax1.bar(episodes, rewards, color=colors, alpha=0.7, label="Episode reward")
+    if len(avg_rewards) > 0:
+        x_avg = episodes[window-1:]
+        ax1.plot(x_avg, avg_rewards, color=GOLD, linewidth=2.5,
+                 label=f"Rolling avg (n={window})", zorder=5)
+    ax1.axhline(0, color='white', linewidth=0.5, linestyle='--', alpha=0.3)
+    ax1.set_xlabel("Episode")
+    ax1.set_ylabel("Reward")
+    ax1.legend(facecolor=PANEL, labelcolor='white', fontsize=9)
+    ax1.set_ylim(-1.2, 1.2)
+
+    # ── Plot 2: Accuracy ──────────────────────────────────────────────────────
+    ax2 = fig.add_subplot(gs[0, 2])
+    style_ax(ax2, "Accuracy")
+    accuracy = sum(correct) / len(correct)
+    ax2.pie(
+        [accuracy, 1-accuracy],
+        labels=["Correct", "Wrong"],
+        colors=[GREEN, RED],
+        autopct='%1.0f%%',
+        textprops={'color': 'white', 'fontsize': 11},
+        startangle=90,
+    )
+    ax2.set_title(f"Accuracy\n{accuracy*100:.1f}% ({sum(correct)}/{len(correct)})",
+                  color=GOLD, fontsize=11, fontweight='bold')
+
+    # ── Plot 3: Decision distribution ─────────────────────────────────────────
+    ax3 = fig.add_subplot(gs[1, 0])
+    style_ax(ax3, "Decision Distribution")
+    from collections import Counter
+    dec_counts = Counter(decisions)
+    labels = list(dec_counts.keys())
+    vals   = list(dec_counts.values())
+    bar_colors = [ACTION_COLORS.get(l, BLUE) for l in labels]
+    bars = ax3.bar(labels, vals, color=bar_colors, alpha=0.85)
+    ax3.set_xticklabels(labels, rotation=20, ha='right', fontsize=8)
+    for bar, val in zip(bars, vals):
+        ax3.text(bar.get_x() + bar.get_width()/2, bar.get_height() + 0.1,
+                 str(val), ha='center', color='white', fontsize=9)
+    ax3.set_ylabel("Count")
+
+    # ── Plot 4: Three-track reward breakdown ──────────────────────────────────
+    ax4 = fig.add_subplot(gs[1, 1])
+    style_ax(ax4, "3-Track Reward Breakdown (avg)")
+    t1_avg = np.mean([r["t1_cultural"] for r in results])
+    t2_avg = np.mean([r["t2_harm"]     for r in results])
+    t3_avg = np.mean([r["t3_policy"]   for r in results])
+    tracks = ["Cultural\n(40%)", "Harm\n(35%)", "Policy\n(25%)"]
+    vals   = [t1_avg, t2_avg, t3_avg]
+    bar_colors2 = [GOLD, PURPLE, BLUE]
+    bars2 = ax4.bar(tracks, vals, color=bar_colors2, alpha=0.85)
+    for bar, val in zip(bars2, vals):
+        ax4.text(bar.get_x() + bar.get_width()/2, bar.get_height() + 0.5,
+                 f"{val:.1f}", ha='center', color='white', fontsize=10)
+    ax4.set_ylabel("Score (0-100)")
+    ax4.set_ylim(0, 110)
+
+    # ── Plot 5: Running accuracy ───────────────────────────────────────────────
+    ax5 = fig.add_subplot(gs[1, 2])
+    style_ax(ax5, "Running Accuracy")
+    if len(avg_correct) > 0:
+        x_acc = episodes[window-1:]
+        ax5.plot(x_acc, avg_correct * 100, color=GREEN, linewidth=2.5)
+        ax5.fill_between(x_acc, avg_correct * 100, alpha=0.2, color=GREEN)
+    ax5.axhline(50, color='white', linewidth=0.5, linestyle='--', alpha=0.3)
+    ax5.set_xlabel("Episode")
+    ax5.set_ylabel("Accuracy (%)")
+    ax5.set_ylim(0, 105)
+
+    # ── Title ──────────────────────────────────────────────────────────────────
+    fig.suptitle(
+        "CACE — Cultural Context Arbitration Environment\nInference Results",
+        color=GOLD, fontsize=14, fontweight='bold', y=1.01
+    )
+
+    plt.savefig(save_path, dpi=150, bbox_inches='tight', facecolor=BG)
+    print(f"\n[PLOT] Saved → {save_path}")
+
+    # Also save JSON
+    json_path = save_path.replace(".png", ".json")
+    with open(json_path, "w") as f:
+        json.dump({
+            "summary": {
+                "episodes": len(results),
+                "accuracy": accuracy,
+                "avg_reward": float(np.mean(rewards)),
+                "total_correct": int(sum(correct)),
+            },
+            "episodes": results,
+        }, f, indent=2)
+    print(f"[DATA] Saved → {json_path}")
+    plt.show()
+
+
+# ── Main ──────────────────────────────────────────────────────────────────────
+
+def main():
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--episodes", type=int, default=20)
+    parser.add_argument("--model",    default=None, help="HF model repo for decisions")
+    parser.add_argument("--env-url",  default=ENV_URL)
+    parser.add_argument("--output",   default="cace_inference_results.png")
+    args = parser.parse_args()
+
+    env = CACEClient(args.env_url)
+    env.wait_until_ready()
+
+    info = env.info()
+    print(f"\n[START] env={info.get('name','cace')} model={args.model or 'rule-based'}")
+    print(f"        action_space={info.get('action_space',{}).get('n')} reward_range={info.get('reward_range')}\n")
+
+    results = run_episodes(env, args.episodes, args.model)
+
+    # Summary
+    rewards  = [r["reward"]  for r in results]
+    accuracy = sum(r["correct"] for r in results) / len(results)
+    print(f"\n[END] episodes={len(results)} accuracy={accuracy:.3f} "
+          f"avg_reward={np.mean(rewards):.3f} "
+          f"rewards={','.join(f'{r:.2f}' for r in rewards)}")
+
+    metrics = env.metrics()
+    print(f"[METRICS] {metrics}")
+
+    plot_results(results, args.output)
+
+
+if __name__ == "__main__":
+    main()

cace_env/models.py

@@ -0,0 +1,104 @@
+"""
+cace_env/models.py
+Pydantic models for CACE — Action, Observation, State.
+All inherit from openenv.core base classes as required by OpenEnv spec.
+"""
+
+from typing import Any, Dict, List, Optional
+from openenv.core import Action, Observation, State
+
+
+# ── Action ────────────────────────────────────────────────────────────────────
+
+class CACEAction(Action):
+    """
+    The Decision Agent's moderation action.
+
+    action_int: int in [0, 4]
+        0 = ALLOW                  — content is legitimate
+        1 = REMOVE                 — content violates policy
+        2 = ALLOW_WITH_LABEL       — allow with warning label
+        3 = ESCALATE               — refer to human reviewer
+        4 = RESTRICT_DISTRIBUTION  — limit reach without removal
+
+    V2 only: selected_indices
+        Which posts to review (from 20 seeded posts, pick 8).
+        Only used when mode="v2" and prioritisation is active.
+    """
+    action_int: int
+    selected_indices: Optional[List[int]] = None  # V2 prioritisation
+
+    @property
+    def action_str(self) -> str:
+        return {
+            0: "ALLOW",
+            1: "REMOVE",
+            2: "ALLOW_WITH_LABEL",
+            3: "ESCALATE",
+            4: "RESTRICT_DISTRIBUTION",
+        }.get(self.action_int, "ESCALATE")
+
+
+# ── Observation ───────────────────────────────────────────────────────────────
+
+class CACEObservation(Observation):
+    """
+    What the Decision Agent sees after reset() or step().
+    Inherits: done, reward, metadata from openenv.core.Observation
+
+    observation: The full enriched case prompt (post + 4-agent deliberation)
+    case_id:     Unique Oversight Board case identifier
+    language:    Detected language of the post
+    region:      Geographic region
+    complexity:  low | medium | high
+    mode:        v1 (single case) | v2 (network batch)
+
+    V2 only:
+    batch_posts: List of 20 post summaries with spread signals
+    """
+    observation: str
+    case_id: str
+    language: str = "Unknown"
+    region: str = "Unknown"
+    complexity: str = "medium"
+    culture_flag: bool = False
+    mode: str = "unified"  # always unified
+
+    # Network fields
+    batch_posts: Optional[List[Dict[str, Any]]] = None
+    network_step: Optional[int] = None
+
+    # Reward breakdown (populated after step())
+    reward_breakdown: Optional[Dict[str, Any]] = None
+
+
+# ── State ─────────────────────────────────────────────────────────────────────
+
+class CACEState(State):
+    """
+    Internal environment state — visible via state() endpoint.
+    Inherits: episode_id, step_count from openenv.core.State
+
+    Used for debugging, demo rendering, and training metrics.
+    """
+    case_id: str = ""
+    post_text: str = ""
+    language: str = "Unknown"
+    region: str = "Unknown"
+    policy_clause: str = "Unknown"
+    cultural_brief: str = ""
+    challenge_brief: str = ""
+    policy_anchor: str = ""
+    ground_truth: str = ""
+    complexity: str = "medium"
+    mode: str = "unified"
+    total_episodes: int = 0
+    correct_decisions: int = 0
+    accuracy: float = 0.0
+    avg_reward_last_50: float = 0.0
+
+    # V2 network state
+    network_nodes: Optional[int] = None
+    network_edges: Optional[int] = None
+    posts_in_batch: Optional[int] = None
+    posts_selected: Optional[int] = None

cace_env/pipeline.py

@@ -0,0 +1,162 @@
+"""
+cace_env/pipeline.py
+4-agent deliberation pipeline.
+Uses pipeline_cache.json first — only calls LLM if cache miss.
+If no API keys set, returns defaults gracefully (no crash).
+"""
+
+import os, json, re
+
+_groq_client  = None
+_azure_client = None
+
+def _groq():
+    global _groq_client
+    if _groq_client is None:
+        from groq import Groq
+        _groq_client = Groq(api_key=os.environ["GROQ_API_KEY"])
+    return _groq_client
+
+def _azure():
+    global _azure_client
+    if _azure_client is None:
+        import openai
+        _azure_client = openai.AzureOpenAI(
+            api_key      =os.environ.get("AZURE_OPENAI_KEY",""),
+            api_version  ="2023-07-01-preview",
+            azure_endpoint=os.environ.get("AZURE_OPENAI_ENDPOINT","https://daplatformai.openai.azure.com/"),
+        )
+    return _azure_client
+
+def llm(system: str, user: str) -> str:
+    """Call LLM — Groq first, Azure fallback. Returns empty string on failure."""
+    msgs = [{"role":"system","content":system},{"role":"user","content":user}]
+
+    if os.environ.get("GROQ_API_KEY"):
+        try:
+            r = _groq().chat.completions.create(
+                model="llama-3.3-70b-versatile", messages=msgs,
+                temperature=0.3, max_tokens=512,
+            )
+            return r.choices[0].message.content.strip()
+        except Exception:
+            pass
+
+    if os.environ.get("AZURE_OPENAI_KEY"):
+        try:
+            r = _azure().chat.completions.create(
+                model="gpt-35-turbo", messages=msgs,
+                temperature=0.3, max_tokens=512,
+            )
+            result = r.choices[0].message.content
+            if result:
+                return result.strip()
+        except Exception:
+            pass
+
+    return ""  # graceful fallback — no crash
+
+
+INTAKE_SYS = """Analyse a social media post and return ONLY valid JSON:
+{"language":"...","region":"...","policy_clause":"..."}
+region: one of: North America, Latin America, Europe, Eastern Europe, Middle East,
+North Africa, West Africa, East Africa, South Asia, Southeast Asia, East Asia, Unknown.
+policy_clause: exact Meta Community Standard name. No other text."""
+
+CULTURAL_SYS = """You are a Cultural Context Analyst for Global South digital discourse.
+Provide the most charitable culturally-informed interpretation in 2-4 sentences."""
+
+CHALLENGE_SYS = """You are an Adversarial Policy Challenger for Trust & Safety.
+Stress-test the cultural argument in 2-4 sentences."""
+
+POLICY_SYS = """You are a Policy Alignment Anchor for content moderation.
+State: (1) what the Meta Community Standard says, (2) exceptions, (3) decision framework. 3-5 sentences."""
+
+
+def enrich(post_text: str, cache: dict = None, case_id: str = None) -> dict:
+    """
+    Run 4-agent deliberation pipeline.
+    Cache-first: if case_id in cache with cultural_brief, return instantly.
+    If no API keys available, returns safe defaults — no crash.
+    """
+    # Cache hit — instant return
+    if cache and case_id and case_id in cache:
+        cached = cache[case_id]
+        if isinstance(cached, dict) and cached.get("cultural_brief"):
+            return cached
+
+    state = {
+        "post_text":      post_text,
+        "language":       "Unknown",
+        "region":         "Unknown",
+        "policy_clause":  "Unknown",
+        "cultural_brief": "",
+        "challenge_brief":"",
+        "policy_anchor":  "",
+        "similar_cases":  [],
+    }
+
+    # If no API keys, return safe defaults immediately
+    has_keys = os.environ.get("GROQ_API_KEY") or os.environ.get("AZURE_OPENAI_KEY")
+    if not has_keys:
+        state["cultural_brief"]  = "Cultural context analysis unavailable (no API key configured)."
+        state["challenge_brief"] = "Challenge analysis unavailable."
+        state["policy_anchor"]   = "Policy anchor unavailable."
+        return state
+
+    # Agent 1: Intake
+    try:
+        raw = llm(INTAKE_SYS, f"Analyse: {post_text[:500]}")
+        if raw:
+            raw = re.sub(r"^```json\s*|```$", "", raw.strip())
+            parsed = json.loads(raw)
+            state["language"]      = parsed.get("language", "Unknown")
+            state["region"]        = parsed.get("region", "Unknown")
+            state["policy_clause"] = parsed.get("policy_clause", "Unknown")
+    except Exception:
+        pass
+
+    # Agent 2: Cultural
+    result = llm(CULTURAL_SYS,
+        f"Post: {post_text[:400]}\nLanguage: {state['language']}\nRegion: {state['region']}")
+    state["cultural_brief"] = result or "Cultural context analysis unavailable."
+
+    # Agent 3: Challenge
+    result = llm(CHALLENGE_SYS,
+        f"Post: {post_text[:400]}\nCultural argument: {state['cultural_brief']}")
+    state["challenge_brief"] = result or "Challenge analysis unavailable."
+
+    # Agent 4: Policy
+    result = llm(POLICY_SYS,
+        f"Policy: {state['policy_clause']}\nCultural: {state['cultural_brief']}\nChallenge: {state['challenge_brief']}")
+    state["policy_anchor"] = result or "Policy anchor unavailable."
+
+    return state
+
+
+def build_observation(state: dict, network_signals: dict = None) -> str:
+    base = f"""You are a content moderation decision agent for a global social media platform.
+
+===== CASE =====
+POST: {state['post_text']}
+LANGUAGE: {state.get('language','Unknown')} | REGION: {state.get('region','Unknown')}
+RELEVANT POLICY: {state.get('policy_clause','Unknown')}
+
+===== DELIBERATION =====
+CULTURAL CONTEXT: {state.get('cultural_brief','Not available.')}
+ADVERSARIAL CHALLENGE: {state.get('challenge_brief','Not available.')}
+POLICY ANCHOR: {state.get('policy_anchor','Not available.')}"""
+
+    if network_signals:
+        base += f"""
+
+===== NETWORK SIGNALS =====
+Share Velocity: {network_signals.get('share_velocity',0):.2f} | Network Reach: {network_signals.get('network_reach',0):.2f}
+Position: {network_signals.get('network_position','unknown')} | Harm Probability: {network_signals.get('harm_probability',0):.2f}"""
+
+    base += """
+
+===== DECISION =====
+Choose exactly ONE: ALLOW | REMOVE | ALLOW_WITH_LABEL | ESCALATE | RESTRICT_DISTRIBUTION
+Decision:"""
+    return base.strip()

cace_env/reward.py

@@ -0,0 +1,106 @@
+"""
+cace_env/reward.py
+Three-Track Reward Model for CACE.
+
+Track 1 — Cultural Meaning Resolution     (weight: 40%)
+Track 2 — Harm Detection Under Context    (weight: 35%)
+Track 3 — Policy Calibration + Escalation (weight: 25%)
+Bonus   — Network Spread Containment      (optional, V2)
+
+All scores deterministic — no LLM judge.
+Returns reward in [-1.0, +1.0] for GRPO compatibility.
+"""
+
+ACTION_SEVERITY = {
+    "ALLOW": 0, "ALLOW_WITH_LABEL": 1,
+    "RESTRICT_DISTRIBUTION": 2, "ESCALATE": 3, "REMOVE": 4,
+}
+ALLOW_OUTCOMES  = {"ALLOW", "ALLOW_WITH_LABEL", "RESTRICT_DISTRIBUTION"}
+REMOVE_OUTCOMES = {"REMOVE"}
+
+W1, W2, W3 = 0.40, 0.35, 0.25
+
+
+def _track1(decision: str, ground_truth: str, complexity: str, culture_flag: bool) -> float:
+    """Cultural Meaning Resolution — 0 to 100."""
+    score = 70.0 if decision == ground_truth else 0.0
+    if decision in REMOVE_OUTCOMES and ground_truth in ALLOW_OUTCOMES:
+        score -= 30.0
+    if decision == "ALLOW_WITH_LABEL" and ground_truth == "ALLOW":
+        score -= 10.0
+    mult = 1.2 if (culture_flag and complexity == "high") else 1.0 if complexity == "medium" else 0.8
+    score = max(-30.0, min(84.0, score * mult))
+    return (score + 30.0) / 1.14
+
+
+def _track2(decision: str, ground_truth: str) -> float:
+    """Harm Detection Under Context — 0 to 100."""
+    if ground_truth in REMOVE_OUTCOMES:
+        if decision in REMOVE_OUTCOMES:        score = 70.0
+        elif decision in ("ESCALATE", "RESTRICT_DISTRIBUTION"): score = 40.0
+        else:                                   score = -100.0
+    else:
+        score = -20.0 if decision in REMOVE_OUTCOMES else 30.0
+    score = max(-100.0, min(70.0, score))
+    return (score + 100.0) / 1.70
+
+
+def _track3(decision: str, ground_truth: str, complexity: str) -> float:
+    """Policy Calibration + Escalation Discipline — 0 to 100."""
+    if decision == ground_truth:
+        score = 60.0
+    else:
+        diff = abs(ACTION_SEVERITY.get(decision, 2) - ACTION_SEVERITY.get(ground_truth, 2))
+        score = 25.0 if diff == 1 else 10.0 if diff == 2 else 0.0
+    if decision == "ESCALATE":
+        score += 40.0 if complexity == "high" else 15.0 if complexity == "medium" else -40.0
+    score = max(-40.0, min(100.0, score))
+    return (score + 40.0) / 1.40
+
+
+def _spread_bonus(decision: str, ground_truth: str, share_velocity: float, network_reach: float) -> float:
+    """Network spread containment bonus — only active in V2."""
+    if share_velocity <= 0 or network_reach <= 0:
+        return 0.0
+    bonus = 0.0
+    if decision in REMOVE_OUTCOMES and ground_truth in REMOVE_OUTCOMES:
+        bonus += 0.3 * share_velocity * network_reach
+    if decision in ALLOW_OUTCOMES and ground_truth in REMOVE_OUTCOMES:
+        bonus -= 0.5 * share_velocity * network_reach
+    return max(-0.5, min(0.5, bonus))
+
+
+def compute_reward(
+    decision: str,
+    ground_truth: str,
+    complexity: str,
+    culture_flag: bool = False,
+    share_velocity: float = 0.0,
+    network_reach: float = 0.0,
+) -> tuple[float, dict]:
+    """
+    Compute three-track reward.
+
+    Returns:
+        (combined_reward, breakdown)
+        combined_reward: float in [-1.0, +1.0]
+        breakdown: per-track scores for logging
+    """
+    t1 = _track1(decision, ground_truth, complexity, culture_flag)
+    t2 = _track2(decision, ground_truth)
+    t3 = _track3(decision, ground_truth, complexity)
+    combined = (W1 * t1) + (W2 * t2) + (W3 * t3)
+    spread = _spread_bonus(decision, ground_truth, share_velocity, network_reach)
+    reward = max(-1.0, min(1.0, (combined / 50.0) - 1.0 + spread))
+
+    breakdown = {
+        "decision": decision, "ground_truth": ground_truth,
+        "correct": decision == ground_truth, "complexity": complexity,
+        "track1_cultural": round(t1, 2),
+        "track2_harm": round(t2, 2),
+        "track3_policy": round(t3, 2),
+        "combined_0_100": round(combined, 2),
+        "spread_bonus": round(spread, 4),
+        "combined_reward": round(reward, 4),
+    }
+    return reward, breakdown

cace_env/server.py

@@ -0,0 +1,409 @@
+"""
+cace_env/server.py
+CACEEnvironment — single unified environment.
+
+One episode flow handles everything:
+  reset() → seed posts on network → IC spread → enrich all → agent picks + decides
+  step()  → three-track reward + spread bonus
+
+No mode switching. No V1/V2 branching. One clean class.
+"""
+
+import os, uuid, random
+from typing import Optional, List
+
+from openenv.core import Environment, create_fastapi_app
+
+from cace_env.models import CACEAction, CACEObservation, CACEState
+from cace_env.dataset import CaseDataset, ACTION_MAP
+from cace_env.pipeline import enrich, build_observation
+from cace_env.reward import compute_reward
+
+# ── Network (optional — graceful fallback if networkx not installed) ──────────
+try:
+    import networkx as nx
+    _HAS_NX = True
+except ImportError:
+    _HAS_NX = False
+
+DATASET_PATH    = os.environ.get("DATASET_PATH", "data/all_cases.json")
+BATCH_SIZE      = int(os.environ.get("BATCH_SIZE", "20"))   # posts per episode
+REVIEW_BUDGET   = int(os.environ.get("REVIEW_BUDGET", "8")) # posts agent must review
+NETWORK_STEPS   = int(os.environ.get("NETWORK_STEPS", "3")) # IC cascade steps
+
+
+# ── Network helpers ───────────────────────────────────────────────────────────
+
+def _build_graph() -> Optional[object]:
+    """
+    Watts-Strogatz small-world graph approximating SNAP Facebook ego topology.
+    ~1000 nodes, avg degree 6, rewiring 0.1.
+    Falls back to None if networkx not available.
+    """
+    if not _HAS_NX:
+        return None
+    return nx.watts_strogatz_graph(n=1000, k=6, p=0.1, seed=42)
+
+
+def _ic_spread(G, seed_nodes: List[int], steps: int) -> List[dict]:
+    """
+    Independent Cascade spread simulation.
+    Returns per-seed spread metrics: share_velocity, network_reach, position.
+    Falls back to synthetic signals if G is None.
+    """
+    if G is None or not _HAS_NX:
+        # Synthetic spread signals when networkx unavailable
+        signals = []
+        for i in range(len(seed_nodes)):
+            v = random.uniform(0.05, 0.6)
+            signals.append({
+                "share_velocity": round(v, 3),
+                "network_reach": round(v * 0.5, 3),
+                "network_position": random.choice(["hub", "bridge", "edge"]),
+            })
+        return signals
+
+    results = []
+    avg_deg = sum(dict(G.degree()).values()) / G.number_of_nodes()
+
+    for seed in seed_nodes:
+        active, newly = {seed}, {seed}
+        for _ in range(steps):
+            nxt = set()
+            for node in newly:
+                p = 1.0 / max(1, G.degree(node))
+                nxt |= {nb for nb in G.neighbors(node)
+                        if nb not in active and random.random() < p}
+            active |= nxt
+            newly = nxt
+
+        reach    = len(active) / G.number_of_nodes()
+        velocity = min(1.0, reach * 2.0)
+        deg      = G.degree(seed)
+        pos      = "hub" if deg > 2*avg_deg else "bridge" if deg > avg_deg else "edge"
+        results.append({
+            "share_velocity":   round(velocity, 3),
+            "network_reach":    round(reach, 3),
+            "network_position": pos,
+        })
+    return results
+
+
+# ── Environment ───────────────────────────────────────────────────────────────
+
+class CACEEnvironment(Environment[CACEAction, CACEObservation, CACEState]):
+    """
+    Cultural Context Arbitration Environment — unified V1+V2.
+
+    Episode flow (always the same):
+
+    reset()
+      1. Sample BATCH_SIZE posts from dataset
+      2. Seed on social graph, run IC spread (3 steps)
+      3. Attach spread signals to each post
+      4. Pre-enrich all posts via 4 frozen agents (uses cache for speed)
+      5. Return batch observation — agent sees ALL posts with signals
+
+    step(action)
+      action.selected_indices: which REVIEW_BUDGET posts to review (V2 prioritisation)
+      action.action_int:        moderation decision (same for all selected posts)
+      → compute 3-track reward + spread bonus per post → return avg reward
+
+    If action.selected_indices is None (simple single-case use):
+      → treat action.action_int as decision for the first (primary) case
+      → compute 3-track reward without spread bonus
+    """
+
+    SUPPORTS_CONCURRENT_SESSIONS = True
+
+    def __init__(self):
+        super().__init__()
+        self._dataset = CaseDataset(DATASET_PATH)
+        self._graph   = _build_graph()
+
+        # Episode state
+        self._episode_id: str = ""
+        self._batch: List[dict] = []         # [{case, enriched, signals, obs_str}]
+        self._step_count: int = 0
+
+        # Metrics
+        self._total_episodes: int = 0
+        self._correct:        int = 0
+        self._rewards:   List[float] = []
+
+    # ── reset ──��──────────────────────────────────────────────────────────────
+
+    def reset(
+        self,
+        seed: Optional[int] = None,
+        episode_id: Optional[str] = None,
+        **kwargs,
+    ) -> CACEObservation:
+        """
+        Start a new episode.
+
+        1. Sample BATCH_SIZE cases.
+        2. Run IC spread on social graph.
+        3. Enrich all cases via 4-agent pipeline (from cache when possible).
+        4. Return batch observation.
+        """
+        if seed is not None:
+            random.seed(seed)
+
+        self._episode_id = episode_id or str(uuid.uuid4())
+        self._step_count = 0
+        self._total_episodes += 1
+
+        # 1. Sample batch
+        cases = self._dataset.sample_batch(BATCH_SIZE)
+
+        # 2. Network spread signals
+        seed_nodes = random.sample(
+            list(self._graph.nodes()) if self._graph else list(range(BATCH_SIZE)),
+            min(BATCH_SIZE, 1000 if self._graph else BATCH_SIZE)
+        )[:BATCH_SIZE]
+        signals = _ic_spread(self._graph, seed_nodes, NETWORK_STEPS)
+
+        # 3. Enrich all cases (from pipeline cache when available — fast)
+        self._batch = []
+        for i, case in enumerate(cases):
+            cached = self._dataset.get_cache(case["id"])
+            enriched = enrich(
+                case["post_text"],
+                cache={case["id"]: cached},
+                case_id=case["id"]
+            )
+            # Ensure post_text is always in enriched state for build_observation
+            enriched["post_text"] = case["post_text"]
+            sig = signals[i] if i < len(signals) else {
+                "share_velocity": 0.1, "network_reach": 0.05, "network_position": "edge"
+            }
+            sig["harm_probability"] = 1.0 if case["board_outcome"] == "REMOVE" else 0.0
+            self._batch.append({
+                "case":     case,
+                "enriched": enriched,
+                "signals":  sig,
+                "obs_str":  build_observation(enriched, sig),
+            })
+
+        # 4. Build unified observation
+        obs_str = self._build_observation()
+
+        return CACEObservation(
+            observation=obs_str,
+            case_id=f"BATCH-{self._episode_id[:8]}",
+            language=self._batch[0]["enriched"].get("language", "Unknown"),
+            region=self._batch[0]["enriched"].get("region", "Unknown"),
+            complexity=self._batch[0]["case"].get("complexity", "medium"),
+            culture_flag=self._batch[0]["case"].get("culture_flag", False),
+            batch_posts=self._batch_summaries(),
+            network_step=0,
+            done=False,
+            reward=None,
+        )
+
+    # ── step ──────────────────────────────────────────────────────────────────
+
+    def step(
+        self,
+        action: CACEAction,
+        timeout_s: Optional[float] = None,
+        **kwargs,
+    ) -> CACEObservation:
+        """
+        Apply moderation decisions.
+
+        If action.selected_indices provided:
+            → review those REVIEW_BUDGET posts, compute per-post 3-track + spread reward
+        Else (single-case fallback):
+            → apply decision to first post only, compute 3-track reward
+        """
+        if not self._batch:
+            raise RuntimeError("Call reset() before step().")
+
+        self._step_count += 1
+        indices = action.selected_indices
+
+        if indices:
+            reward, breakdown = self._step_batch(action.action_str, indices)
+        else:
+            reward, breakdown = self._step_single(action.action_str)
+
+        self._rewards.append(reward)
+        if breakdown.get("correct"):
+            self._correct += 1
+
+        primary_case = self._batch[0]["case"]
+        return CACEObservation(
+            observation=self._build_observation(),
+            case_id=f"BATCH-{self._episode_id[:8]}",
+            language=self._batch[0]["enriched"].get("language", "Unknown"),
+            region=self._batch[0]["enriched"].get("region", "Unknown"),
+            complexity=primary_case.get("complexity", "medium"),
+            culture_flag=primary_case.get("culture_flag", False),
+            mode="batch" if indices else "single",
+            done=True,
+            reward=reward,
+            reward_breakdown={
+                **breakdown,
+                "ground_truth": primary_case.get("board_outcome", "?"),
+                "case_id": primary_case.get("id", "?"),
+            },
+        )
+
+    def _step_single(self, decision: str) -> tuple[float, dict]:
+        """Single-case decision (first post in batch). No spread bonus."""
+        item = self._batch[0]
+        case = item["case"]
+        r, bd = compute_reward(
+            decision=decision,
+            ground_truth=case["board_outcome"],
+            complexity=case["complexity"],
+            culture_flag=case.get("culture_flag", False),
+            share_velocity=0.0,
+            network_reach=0.0,
+        )
+        return r, bd
+
+    def _step_batch(self, decision: str, indices: List[int]) -> tuple[float, dict]:
+        """Batch review: compute reward per selected post, return average."""
+        selected = [self._batch[i] for i in indices if i < len(self._batch)]
+        total, breakdowns = 0.0, []
+
+        for item in selected[:REVIEW_BUDGET]:
+            case = item["case"]
+            sig  = item["signals"]
+            r, bd = compute_reward(
+                decision=decision,
+                ground_truth=case["board_outcome"],
+                complexity=case["complexity"],
+                culture_flag=case.get("culture_flag", False),
+                share_velocity=sig["share_velocity"],
+                network_reach=sig["network_reach"],
+            )
+            total += r
+            breakdowns.append(bd)
+
+        avg = total / max(1, len(breakdowns))
+        correct_count = sum(1 for bd in breakdowns if bd["correct"])
+        return avg, {
+            "avg_reward": round(avg, 4),
+            "correct": correct_count == len(breakdowns),
+            "correct_count": correct_count,
+            "total_reviewed": len(breakdowns),
+            "per_post": breakdowns,
+        }
+
+    # ── Observation builder ───────────────────────────────────────────────────
+
+    def _build_observation(self) -> str:
+        """
+        Unified observation: network batch summary + primary case enrichment.
+        Agent sees both the spread signals (for prioritisation) and the full
+        deliberation context (for the moderation decision).
+        """
+        # Part 1: Network batch summary (for prioritisation)
+        lines = [
+            "═══ CULTURAL CONTEXT ARBITRATION ENVIRONMENT ═══",
+            f"Episode: {self._episode_id[:8]} | Posts: {BATCH_SIZE} | Review budget: {REVIEW_BUDGET}",
+            "",
+            "── NETWORK BATCH (select your review queue) ──",
+        ]
+        for i, item in enumerate(self._batch):
+            s   = item["signals"]
+            c   = item["case"]
+            tag = "⚠️ " if s["harm_probability"] > 0.5 else "   "
+            lines.append(
+                f"[{i:02d}] {tag}{c['id']} | {item['enriched'].get('language','?')} | "
+                f"{item['enriched'].get('region','?')} | "
+                f"velocity={s['share_velocity']:.2f} reach={s['network_reach']:.2f} "
+                f"pos={s['network_position']}"
+            )
+            lines.append(f"      {c['post_text'][:90]}...")
+
+        # Part 2: Primary case full deliberation (for decision)
+        primary = self._batch[0]
+        lines += [
+            "",
+            "── PRIMARY CASE (full deliberation) ──",
+            build_observation(primary["enriched"], primary["signals"]),
+            "",
+            "── YOUR TASK ──",
+            f"1. SELECT {REVIEW_BUDGET} indices to review (comma-separated): e.g. 0,3,5,7,9,11,14,17",
+            "2. DECIDE for each selected post:",
+            "   ALLOW | REMOVE | ALLOW_WITH_LABEL | ESCALATE | RESTRICT_DISTRIBUTION",
+            "",
+            "Format: INDICES: 0,3,5,... | DECISION: ALLOW",
+        ]
+        return "\n".join(lines)
+
+    def _batch_summaries(self) -> List[dict]:
+        return [
+            {
+                "index":           i,
+                "case_id":         item["case"]["id"],
+                "post_preview":    item["case"]["post_text"][:100],
+                "language":        item["enriched"].get("language", "Unknown"),
+                "region":          item["enriched"].get("region", "Unknown"),
+                "share_velocity":  item["signals"]["share_velocity"],
+                "network_reach":   item["signals"]["network_reach"],
+                "harm_probability":item["signals"]["harm_probability"],
+                "network_position":item["signals"]["network_position"],
+                "ground_truth":    item["case"]["board_outcome"],
+            }
+            for i, item in enumerate(self._batch)
+        ]
+
+    # ── state property ────────────────────────────────────────────────────────
+
+    @property
+    def state(self) -> CACEState:
+        primary = self._batch[0] if self._batch else {}
+        avg_50  = (
+            sum(self._rewards[-50:]) / min(50, len(self._rewards))
+            if self._rewards else 0.0
+        )
+        return CACEState(
+            episode_id=self._episode_id,
+            step_count=self._step_count,
+            case_id=(primary.get("case") or {}).get("id", ""),
+            post_text=(primary.get("case") or {}).get("post_text", "")[:200],
+            language=(primary.get("enriched") or {}).get("language", "Unknown"),
+            region=(primary.get("enriched") or {}).get("region", "Unknown"),
+            policy_clause=(primary.get("enriched") or {}).get("policy_clause", "Unknown"),
+            cultural_brief=(primary.get("enriched") or {}).get("cultural_brief", "")[:150],
+            challenge_brief=(primary.get("enriched") or {}).get("challenge_brief", "")[:150],
+            policy_anchor=(primary.get("enriched") or {}).get("policy_anchor", "")[:150],
+            ground_truth=(primary.get("case") or {}).get("board_outcome", ""),
+            complexity=(primary.get("case") or {}).get("complexity", "medium"),
+            mode="unified",
+            total_episodes=self._total_episodes,
+            correct_decisions=self._correct,
+            accuracy=round(self._correct / max(1, self._total_episodes), 4),
+            avg_reward_last_50=round(avg_50, 4),
+            network_nodes=self._graph.number_of_nodes() if self._graph else None,
+            network_edges=self._graph.number_of_edges() if self._graph else None,
+            posts_in_batch=len(self._batch),
+            posts_selected=REVIEW_BUDGET,
+        )
+
+
+# ── FastAPI app ───────────────────────────────────────────────────────────────
+
+# Use singleton — OpenEnv creates new instance per request by default
+# which breaks stateful environments. We use a single shared instance.
+_ENV_INSTANCE = CACEEnvironment()
+
+def env_factory():
+    return _ENV_INSTANCE
+
+app = create_fastapi_app(
+    env=env_factory,
+    action_cls=CACEAction,
+    observation_cls=CACEObservation,
+    max_concurrent_envs=1,  # single instance = single concurrent session
+)
+
+if __name__ == "__main__":
+    import uvicorn
+    uvicorn.run(app, host="0.0.0.0", port=7860)

openenv.yaml

@@ -0,0 +1,31 @@
+name: cace_env
+version: 0.2.0
+description: >
+  Cultural Context Arbitration Environment — trains LLMs to make culturally-aware
+  content moderation decisions using Meta's Oversight Board rulings as a verifiable
+  reward oracle. Three-track reward model. Real social network spread simulation.
+  No LLM judge. Fully deterministic reward.
+
+client:
+  class_name: CACEEnvClient
+  module: cace_env.client
+
+action:
+  class_name: CACEAction
+  module: cace_env.models
+
+observation:
+  class_name: CACEObservation
+  module: cace_env.models
+
+default_image: cace-env:latest
+spec_version: 1
+
+tags:
+  - reinforcement-learning
+  - content-moderation
+  - multi-agent
+  - cultural-context
+  - oversight-board
+  - grpo
+  - rlvr

pyproject.toml

@@ -0,0 +1,31 @@
+[build-system]
+requires = ["hatchling"]
+build-backend = "hatchling.build"
+
+[project]
+name = "cace-env"
+version = "0.2.0"
+description = "Cultural Context Arbitration Environment — OpenEnv RL environment"
+requires-python = ">=3.11"
+
+dependencies = [
+    "openenv-core>=0.1.0",
+    "fastapi>=0.111.0",
+    "uvicorn>=0.30.0",
+    "pydantic>=2.0.0",
+    "python-dotenv>=1.0.0",
+    "requests>=2.31.0",
+    "groq>=0.9.0",
+    "openai>=1.0.0",
+]
+
+[project.optional-dependencies]
+network = [
+    "networkx>=3.3",
+]
+all = [
+    "networkx>=3.3",
+]
+
+[tool.hatch.build.targets.wheel]
+packages = ["cace_env"]