Delete server

server/Dockerfile

@@ -1,80 +0,0 @@
-# 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=gov_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
-# 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 --frozen --no-install-project --no-editable; \
-    else \
-        uv sync --no-install-project --no-editable; \
-    fi
-
-RUN --mount=type=cache,target=/root/.cache/uv \
-    if [ -f uv.lock ]; then \
-        uv sync --frozen --no-editable; \
-    else \
-        uv sync --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: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 8000"]

server/__init__.py

@@ -1,11 +0,0 @@
-# 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.
-
-"""Gov Env environment server components."""
-
-from .gov_env_environment import GovEnvironment
-
-__all__ = ["GovEnvironment"]

server/app.py

@@ -1,21 +0,0 @@
-try:
-    from openenv.core.env_server.http_server import create_app
-except Exception as e:
-    raise ImportError("openenv is required. Install: pip install openenv-core") from e
-
-from models import GovAction, GovObservation
-from server.gov_env_environment import GovEnvironment
-
-app = create_app(
-    GovEnvironment,
-    GovAction,
-    GovObservation,
-    env_name="gov_env",
-    max_concurrent_envs=1,
-)
-
-if __name__ == "__main__":
-    import uvicorn
-    import os
-    port = int(os.environ.get("PORT", 7860))
-    uvicorn.run(app, host="0.0.0.0", port=port)

server/gov_env_environment.py

@@ -1,400 +0,0 @@
-# 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.
-
-"""
-Government Service Application Assistant Environment Implementation.
-
-A simple implementation that simulates government service document assistance.
-"""
-
-from uuid import uuid4
-from typing import Dict, Any, Optional
-
-from openenv.core.env_server.interfaces import Environment
-from openenv.core.env_server.types import State
-
-from models import GovAction, GovObservation
-
-
-class GovEnvironment(Environment):
-    """
-    Government Service Application Assistant Environment.
-
-    This environment simulates an AI assistant helping users with government service applications.
-    """
-
-    SUPPORTS_CONCURRENT_SESSIONS: bool = True
-
-    def __init__(self):
-        """Initialize the government service environment."""
-        self._state = State(episode_id=str(uuid4()), step_count=0)
-        self._service_type: Optional[str] = None
-        self._current_stage = "service_selection"
-        self._required_documents: list[Dict[str, Any]] = []
-        self._submitted_documents: list[Dict[str, Any]] = []
-        self._validation_results: Optional[Dict[str, Any]] = None
-        self._correction_suggestions: list[Dict[str, Any]] = []
-        self._application_id: Optional[str] = None
-        self._done = False
-
-    def reset(self) -> GovObservation:
-        """
-        Reset the environment to initial state.
-
-        Returns:
-            GovObservation with welcome message
-        """
-        self._state = State(episode_id=str(uuid4()), step_count=0)
-        self._service_type = None
-        self._current_stage = "service_selection"
-        self._required_documents = []
-        self._submitted_documents = []
-        self._validation_results = None
-        self._correction_suggestions = []
-        self._application_id = None
-        self._done = False
-
-        return GovObservation(
-            message="Government Service Assistant ready. Please select a service to begin.",
-            current_stage=self._current_stage,
-            reward=0.0,
-            done=False,
-        )
-
-    def step(self, action: GovAction) -> GovObservation:  # type: ignore[override]
-        """
-        Execute a step in the environment.
-
-        Args:
-            action: GovAction containing the action to perform
-
-        Returns:
-            GovObservation with updated state and feedback
-        """
-        self._state.step_count += 1
-
-        # Handle the action
-        observation = self._handle_action(action)
-
-        # Check if episode should end
-        observation.done = self._done or self._state.step_count >= 10
-
-        return observation
-
-    def _handle_action(self, action: GovAction) -> GovObservation:
-        """Handle the specific action type."""
-        # For backward compatibility, if no action_type, treat as echo
-        if not action.action_type:
-            return GovObservation(
-                echoed_message=action.message,
-                message_length=len(action.message),
-                message=f"Echo: {action.message}",
-                current_stage=self._current_stage,
-                reward=min(len(action.message) * 0.1, 1.0),
-                done=False,
-            )
-
-        if action.action_type == "select_service":
-            return self._handle_select_service(action)
-        elif action.action_type == "list_required_documents":
-            return self._handle_list_required_documents(action)
-        elif action.action_type == "validate_documents":
-            return self._handle_validate_documents(action)
-        elif action.action_type == "suggest_corrections":
-            return self._handle_suggest_corrections(action)
-        elif action.action_type == "submit_application":
-            return self._handle_submit_application(action)
-        else:
-            return GovObservation(
-                message=f"Unknown action type: {action.action_type}",
-                current_stage=self._current_stage,
-                reward=-0.1,
-                done=self._done,
-            )
-
-    def _handle_select_service(self, action: GovAction) -> GovObservation:
-        """Handle service selection action."""
-        if not action.service_type:
-            return GovObservation(
-                message="Please specify a service type to select.",
-                current_stage=self._current_stage,
-                reward=-0.05,
-                done=False,
-            )
-
-        service_type = action.service_type
-        available_services = ["passport_new", "passport_renewal", "pan_new", "aadhaar_update", "driving_license_new"]
-
-        if service_type not in available_services:
-            return GovObservation(
-                message=f"Unknown service type: {service_type}. Available: {available_services}",
-                current_stage=self._current_stage,
-                reward=-0.1,
-                done=False,
-            )
-
-        # Update state
-        self._service_type = service_type
-        self._required_documents = self._get_required_documents(service_type)
-        self._current_stage = "document_identification"
-
-        return GovObservation(
-            message=f"Selected service: {service_type}. {len(self._required_documents)} documents required.",
-            current_stage=self._current_stage,
-            service_type=self._service_type,
-            required_documents=self._required_documents,
-            reward=0.1,
-            done=False,
-        )
-
-    def _handle_list_required_documents(self, action: GovAction) -> GovObservation:
-        """Handle listing required documents."""
-        if not self._service_type:
-            return GovObservation(
-                message="Please select a service first.",
-                current_stage=self._current_stage,
-                reward=-0.05,
-                done=False,
-            )
-
-        return GovObservation(
-            message=f"Required documents for {self._service_type}:",
-            current_stage=self._current_stage,
-            service_type=self._service_type,
-            required_documents=self._required_documents,
-            reward=0.05,
-            done=False,
-        )
-
-    def _handle_validate_documents(self, action: GovAction) -> GovObservation:
-        """Handle document validation."""
-        if not action.documents:
-            return GovObservation(
-                message="Please provide documents to validate.",
-                current_stage=self._current_stage,
-                reward=-0.05,
-                done=False,
-            )
-
-        # Update submitted documents
-        self._submitted_documents = [doc.model_dump() for doc in action.documents]
-
-        # Perform validation
-        validation_results = self._validate_documents(action.documents)
-        self._validation_results = validation_results
-
-        # Update stage
-        self._current_stage = "validation"
-
-        # Generate message
-        if validation_results["is_complete"] and validation_results["is_valid"]:
-            message = "All documents are valid and complete. You can now submit the application."
-            self._current_stage = "submission"
-        elif validation_results["is_complete"] and not validation_results["is_valid"]:
-            message = f"Documents are complete but some are invalid. Found {len(validation_results['invalid_documents'])} invalid documents."
-            self._current_stage = "correction"
-        else:
-            message = f"Documents are incomplete. Missing {len(validation_results['missing_documents'])} required documents."
-            self._current_stage = "correction"
-
-        return GovObservation(
-            message=message,
-            current_stage=self._current_stage,
-            service_type=self._service_type,
-            submitted_documents=self._submitted_documents,
-            validation_results=validation_results,
-            reward=0.0,
-            done=False,
-        )
-
-    def _handle_suggest_corrections(self, action: GovAction) -> GovObservation:
-        """Handle suggesting corrections."""
-        if not self._validation_results:
-            return GovObservation(
-                message="Please validate documents first.",
-                current_stage=self._current_stage,
-                reward=-0.05,
-                done=False,
-            )
-
-        corrections = self._generate_corrections()
-        self._correction_suggestions = corrections
-
-        if corrections:
-            message = f"Generated {len(corrections)} correction suggestions."
-        else:
-            message = "No corrections needed - all documents appear valid."
-
-        return GovObservation(
-            message=message,
-            current_stage=self._current_stage,
-            service_type=self._service_type,
-            submitted_documents=self._submitted_documents,
-            validation_results=self._validation_results,
-            correction_suggestions=corrections,
-            reward=0.0,
-            done=False,
-        )
-
-    def _handle_submit_application(self, action: GovAction) -> GovObservation:
-        """Handle application submission."""
-        if not self._validation_results:
-            return GovObservation(
-                message="Please validate documents before submitting.",
-                current_stage=self._current_stage,
-                reward=-0.1,
-                done=False,
-            )
-
-        validation = self._validation_results
-        if not validation["is_complete"] or not validation["is_valid"]:
-            return GovObservation(
-                message="Cannot submit - documents are incomplete or invalid.",
-                current_stage=self._current_stage,
-                reward=-0.1,
-                done=False,
-            )
-
-        # Generate application ID and complete
-        application_id = f"APP-{str(uuid4())[:8].upper()}"
-        self._application_id = application_id
-        self._current_stage = "completed"
-        self._done = True
-
-        return GovObservation(
-            message=f"Application submitted successfully! ID: {application_id}",
-            current_stage=self._current_stage,
-            service_type=self._service_type,
-            application_id=application_id,
-            is_complete=True,
-            reward=0.5,
-            done=True,
-        )
-
-    def _get_required_documents(self, service_type: str) -> list[Dict[str, Any]]:
-        """Get required documents for a service type."""
-        if service_type == "passport_new":
-            return [
-                {"type": "Proof of Address", "description": "Aadhaar/Voter ID/Utility bill", "mandatory": True},
-                {"type": "Proof of Date of Birth", "description": "Birth certificate/PAN/Aadhaar", "mandatory": True},
-                {"type": "Photograph", "description": "Recent passport-size photo", "mandatory": True}
-            ]
-        elif service_type == "passport_renewal":
-            return [
-                {"type": "Old Passport", "description": "Original passport to renew", "mandatory": True},
-                {"type": "Proof of Address", "description": "Aadhaar/Voter ID/Utility bill", "mandatory": True},
-                {"type": "Photograph", "description": "Recent passport-size photo", "mandatory": True}
-            ]
-        elif service_type == "pan_new":
-            return [
-                {"type": "Proof of Identity", "description": "Aadhaar/Voter ID/Driving license", "mandatory": True},
-                {"type": "Proof of Address", "description": "Aadhaar/Voter ID/Utility bill", "mandatory": True},
-                {"type": "Proof of Date of Birth", "description": "Birth certificate/PAN/Aadhaar", "mandatory": True}
-            ]
-        elif service_type == "aadhaar_update":
-            return [
-                {"type": "Proof of Address", "description": "Utility bill/Bank statement/Rental agreement", "mandatory": True},
-                {"type": "Proof of Identity", "description": "PAN/Passport/Voter ID", "mandatory": True}
-            ]
-        elif service_type == "driving_license_new":
-            return [
-                {"type": "Proof of Age", "description": "Birth certificate/School certificate", "mandatory": True},
-                {"type": "Proof of Address", "description": "Aadhaar/Voter ID/Utility bill", "mandatory": True},
-                {"type": "Application Form", "description": "Form 2 or Form 4", "mandatory": True}
-            ]
-        return []
-
-    def _validate_documents(self, documents: list) -> Dict[str, Any]:
-        """Validate submitted documents."""
-        result = {
-            "is_complete": False,
-            "is_valid": False,
-            "missing_documents": [],
-            "invalid_documents": [],
-            "valid_documents": []
-        }
-
-        required_types = {doc["type"] for doc in self._required_documents if doc["mandatory"]}
-        submitted_types = {doc.type for doc in documents}
-
-        # Check missing
-        result["missing_documents"] = list(required_types - submitted_types)
-
-        # Check invalid
-        for doc in documents:
-            is_valid = True
-            reason = ""
-
-            details_lower = doc.details.lower()
-
-            if "expired" in details_lower or "2023" in details_lower:
-                is_valid = False
-                reason = "Document appears to be expired"
-            elif doc.type == "Proof of Address" and "spouse" in details_lower:
-                is_valid = False
-                reason = "Must be in applicant's name"
-            elif doc.type == "Proof of Date of Birth" and "10th marksheet" in details_lower:
-                is_valid = False
-                reason = "Not sufficient as standalone DOB proof"
-
-            if is_valid:
-                result["valid_documents"].append(doc.type)
-            else:
-                result["invalid_documents"].append({"type": doc.type, "reason": reason})
-
-        result["is_complete"] = len(result["missing_documents"]) == 0
-        result["is_valid"] = len(result["invalid_documents"]) == 0
-
-        return result
-
-    def _generate_corrections(self) -> list[Dict[str, Any]]:
-        """Generate correction suggestions."""
-        corrections = []
-
-        if not self._validation_results:
-            return corrections
-
-        # Handle missing documents
-        for missing in self._validation_results["missing_documents"]:
-            if missing == "Proof of Address":
-                corrections.append({
-                    "issue_type": "missing",
-                    "suggested_action": "Submit a valid address proof",
-                    "alternative_documents": ["Aadhaar Card", "Voter ID", "Utility Bill"]
-                })
-            elif missing == "Proof of Date of Birth":
-                corrections.append({
-                    "issue_type": "missing",
-                    "suggested_action": "Submit a valid DOB proof",
-                    "alternative_documents": ["Birth Certificate", "PAN Card", "Aadhaar Card"]
-                })
-
-        # Handle invalid documents
-        for invalid in self._validation_results["invalid_documents"]:
-            doc_type = invalid["type"]
-            reason = invalid["reason"]
-
-            if reason == "Document appears to be expired":
-                corrections.append({
-                    "issue_type": "expired",
-                    "current_document": doc_type,
-                    "suggested_action": f"Replace expired {doc_type} with current version",
-                    "alternative_documents": ["Updated version of same document"]
-                })
-            elif "applicant's name" in reason:
-                corrections.append({
-                    "issue_type": "name_mismatch",
-                    "current_document": doc_type,
-                    "suggested_action": f"Submit {doc_type} in applicant's name",
-                    "alternative_documents": ["Alternative address proof in applicant's name"]
-                })
-
-        return corrections
-
-    @property
-    def state(self) -> State:
-        """Get the current environment state."""
-        return self._state

server/gov_env_environment.py.backup

@@ -1,104 +0,0 @@
-# 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.
-
-"""
-Gov Env Environment Implementation.
-
-A simple test environment that echoes back messages sent to it.
-Perfect for testing HTTP server infrastructure.
-"""
-
-from uuid import uuid4
-
-from openenv.core.env_server.interfaces import Environment
-from openenv.core.env_server.types import State
-
-try:
-    from ..models import GovAction, GovObservation
-except ImportError:
-    from models import GovAction, GovObservation
-
-
-class GovEnvironment(Environment):
-    """
-    A simple echo environment that echoes back messages.
-
-    This environment is designed for testing the HTTP server infrastructure.
-    It maintains minimal state and simply echoes back whatever message it receives.
-
-    Example:
-        >>> env = GovEnvironment()
-        >>> obs = env.reset()
-        >>> print(obs.echoed_message)  # "Gov Env environment ready!"
-        >>>
-        >>> obs = env.step(GovAction(message="Hello"))
-        >>> print(obs.echoed_message)  # "Hello"
-        >>> print(obs.message_length)  # 5
-    """
-
-    # Enable concurrent WebSocket sessions.
-    # Set to True if your environment isolates state between instances.
-    # When True, multiple WebSocket clients can connect simultaneously, each
-    # getting their own environment instance (when using factory mode in app.py).
-    SUPPORTS_CONCURRENT_SESSIONS: bool = True
-
-    def __init__(self):
-        """Initialize the gov_env environment."""
-        self._state = State(episode_id=str(uuid4()), step_count=0)
-        self._reset_count = 0
-
-    def reset(self) -> GovObservation:
-        """
-        Reset the environment.
-
-        Returns:
-            GovObservation with a ready message
-        """
-        self._state = State(episode_id=str(uuid4()), step_count=0)
-        self._reset_count += 1
-
-        return GovObservation(
-            echoed_message="Gov Env environment ready!",
-            message_length=0,
-            done=False,
-            reward=0.0,
-        )
-
-    def step(self, action: GovAction) -> GovObservation:  # type: ignore[override]
-        """
-        Execute a step in the environment by echoing the message.
-
-        Args:
-            action: GovAction containing the message to echo
-
-        Returns:
-            GovObservation with the echoed message and its length
-        """
-        self._state.step_count += 1
-
-        message = action.message
-        length = len(message)
-
-        # Simple reward: longer messages get higher rewards
-        reward = length * 0.1
-
-        return GovObservation(
-            echoed_message=message,
-            message_length=length,
-            done=False,
-            reward=reward,
-            metadata={"original_message": message, "step": self._state.step_count},
-        )
-
-    @property
-    def state(self) -> State:
-        """
-        Get the current environment state.
-
-        Returns:
-            Current State with episode_id and step_count
-        """
-        return self._state

server/requirements.txt

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