import time from typing import Any, Dict, Optional from uuid import uuid4 from fastapi import HTTPException from fastapi.responses import HTMLResponse from pydantic import BaseModel, Field try: from ..models import CounselAction from .counsel_env_environment import CounselEnvironment except (ImportError, ModuleNotFoundError): # pragma: no cover - supports Docker-style imports from models import CounselAction from server.counsel_env_environment import CounselEnvironment SESSION_TTL_SECONDS = 60 * 60 DEFAULT_SEED = 20260425 BENCHMARK_ROWS = [ { "agent": "random", "episodes": 150, "avg_reward": 0.000, "primary_reward": 0.000, "trigger_rate": 0.000, "surface_rate": 0.000, "takeaway": "Vague questions and premature evidence do not score.", }, { "agent": "keyword_spam", "episodes": 150, "avg_reward": 0.066, "primary_reward": 0.000, "trigger_rate": 0.650, "surface_rate": 0.000, "takeaway": "Trigger words alone get only tiny shaping reward.", }, { "agent": "present_all", "episodes": 150, "avg_reward": 0.000, "primary_reward": 0.000, "trigger_rate": 0.000, "surface_rate": 0.000, "takeaway": "Blindly dumping exhibits fails because timing matters.", }, { "agent": "trained_qwen3_8b_qlora_sft_run4b_eval150", "episodes": 150, "avg_reward": 0.864, "primary_reward": 0.943, "trigger_rate": 0.943, "surface_rate": 0.943, "takeaway": "Qwen3-8B QLoRA SFT learns the trigger-then-evidence loop reliably.", }, { "agent": "scripted_oracle", "episodes": 150, "avg_reward": 0.901, "primary_reward": 0.957, "trigger_rate": 0.957, "surface_rate": 0.957, "takeaway": "The target behavior is trigger first, evidence second.", }, ] _SESSIONS: Dict[str, tuple[float, CounselEnvironment]] = {} class DemoResetRequest(BaseModel): seed: Optional[int] = Field(default=DEFAULT_SEED) difficulty: str = Field(default="easy") curriculum_stage: str = Field(default="easy") class DemoStepRequest(BaseModel): session_id: str tool: str text: Optional[str] = None exhibit_id: Optional[str] = None reason: Optional[str] = None def register_demo_routes(app: Any) -> None: @app.get("/", response_class=HTMLResponse, include_in_schema=False) def landing_page() -> HTMLResponse: return HTMLResponse(_landing_html()) @app.get("/demo", response_class=HTMLResponse, include_in_schema=False) def demo_page() -> HTMLResponse: return HTMLResponse(_demo_html()) @app.get("/demo/api/benchmarks") def demo_benchmarks() -> Dict[str, Any]: return {"benchmarks": BENCHMARK_ROWS} @app.post("/demo/api/reset") def demo_reset(request: DemoResetRequest) -> Dict[str, Any]: _prune_sessions() env = CounselEnvironment() observation = env.reset( seed=request.seed, difficulty=request.difficulty, curriculum_stage=request.curriculum_stage, episode_id=f"space_demo_{request.seed or 'random'}", ) session_id = uuid4().hex _SESSIONS[session_id] = (time.time(), env) return _payload(session_id, env, observation) @app.post("/demo/api/step") def demo_step(request: DemoStepRequest) -> Dict[str, Any]: env = _get_session(request.session_id) action = CounselAction( tool=request.tool, text=request.text, exhibit_id=request.exhibit_id, reason=request.reason, ) observation = env.step(action) _SESSIONS[request.session_id] = (time.time(), env) return _payload(request.session_id, env, observation) def _get_session(session_id: str) -> CounselEnvironment: _prune_sessions() entry = _SESSIONS.get(session_id) if entry is None: raise HTTPException(status_code=404, detail="Demo session expired. Reset the case.") return entry[1] def _prune_sessions() -> None: now = time.time() expired = [ session_id for session_id, (last_seen, _env) in _SESSIONS.items() if now - last_seen > SESSION_TTL_SECONDS ] for session_id in expired: _SESSIONS.pop(session_id, None) def _dump_model(model: Any) -> Dict[str, Any]: if hasattr(model, "model_dump"): return model.model_dump() return model.dict() def _payload(session_id: str, env: CounselEnvironment, observation: Any) -> Dict[str, Any]: return { "session_id": session_id, "observation": _dump_model(observation), "state": _dump_model(env.state), "oracle_hint": _oracle_hint(env), "benchmarks": BENCHMARK_ROWS, } def _oracle_hint(env: CounselEnvironment) -> Dict[str, str]: if env.witness is None: return {"label": "Reset the case", "detail": "Start an episode to see the target mechanic."} for contradiction in env.witness.contradictions: if not contradiction.triggered: question = f"{contradiction.trigger_keywords[0]}?" return { "label": "Trigger a sealed claim", "detail": f'Ask: "{question}"', } if not contradiction.surfaced: return { "label": "Present the matching exhibit", "detail": f"Use exhibit: {contradiction.disprover_evidence_id}", } if not env.done: return {"label": "Rest the case", "detail": "All known contradictions are surfaced."} return {"label": "Episode complete", "detail": "Reset to generate a new case."} def _landing_html() -> str: return """ Counsel-Env
OpenEnv hackathon environment

Train LLMs to catch lies under pressure.

Counsel-Env is a cross-examination arena. The agent must make a deterministic witness commit to a claim, then present the one exhibit that proves the claim false.

Open the live demo
0.902scripted oracle reward
0.000random baseline reward
30held-out seeded cases
""" def _demo_html() -> str: return """ Counsel-Env Live Demo

Counsel-Env Live Demo

Make the witness commit, then present the exhibit that exposes the contradiction. This is the target behavior for post-training.

1. Start a Case

Ready.

2. Ask the Witness

3. Evidence

Reset a case to load exhibits.

Case Brief

No active case.

Oracle Hint for Demo

Reset a case to see the target sequence.

Reward and State

Transcript

No transcript yet.

Held-Out Baselines

"""