Add "read only" run mode for codegate

src/codegate/cli.py

@@ -146,6 +146,12 @@ def show_prompts(prompts: Optional[Path]) -> None:
     default=None,
     help="Host to bind to (default: localhost)",
 )
+@click.option(
+    "--read-only",
+    is_flag=True,
+    default=False,
+    help="Run in read-only mode (v1 API will not be served)",
+)
 @click.option(
     "--log-level",
     type=click.Choice([level.value for level in LogLevel]),
@@ -258,6 +264,7 @@ def serve(  # noqa: C901
     port: Optional[int],
     proxy_port: Optional[int],
     host: Optional[str],
+    read_only: bool,
     log_level: Optional[str],
     log_format: Optional[str],
     config: Optional[Path],
@@ -332,9 +339,12 @@ def serve(  # noqa: C901
 
         # Initialize secrets manager and pipeline factory
         secrets_manager = SecretsManager()
-        pipeline_factory = PipelineFactory(secrets_manager)
+        pipeline_factory = PipelineFactory(secrets_manager, read_only=read_only)
+
+        app = init_app(pipeline_factory, read_only=read_only)
 
-        app = init_app(pipeline_factory)
+        if read_only:
+            logger.info("Starting server in read-only mode (v1 API disabled)")
 
         # Set up event loop
         loop = asyncio.new_event_loop()
@@ -382,6 +392,7 @@ async def run_servers(cfg: Config, app) -> None:
                 "certs_dir": cfg.certs_dir,
                 "db_path": cfg.db_path,
                 "vec_db_path": cfg.vec_db_path,
+                "read_only": app.read_only,
             },
         )
 

src/codegate/pipeline/cli/cli.py

@@ -98,6 +98,15 @@ def _get_cli_from_continue(last_user_message_str: str) -> Optional[re.Match[str]
 class CodegateCli(PipelineStep):
     """Pipeline step that handles codegate cli."""
 
+    def __init__(self, read_only: bool = False):
+        """
+        Initialize the CodegateCli step.
+
+        Args:
+            read_only: If True, the CLI commands will be disabled
+        """
+        self.read_only = read_only
+
     @property
     def name(self) -> str:
         """
@@ -123,6 +132,10 @@ async def process(
             PipelineResult: Contains the response if triggered, otherwise continues
             pipeline
         """
+        # If in read-only mode, don't process CLI commands and just fall through
+        if self.read_only:
+            return PipelineResult(request=request, context=context)
+
         last_user_message = self.get_last_user_message(request)
 
         if last_user_message is not None:

src/codegate/pipeline/factory.py

@@ -22,8 +22,9 @@
 
 
 class PipelineFactory:
-    def __init__(self, secrets_manager: SecretsManager):
+    def __init__(self, secrets_manager: SecretsManager, read_only: bool = False):
         self.secrets_manager = secrets_manager
+        self.read_only = read_only
 
     def create_input_pipeline(self, client_type: ClientType) -> SequentialPipelineProcessor:
         input_steps: List[PipelineStep] = [
@@ -33,7 +34,7 @@ def create_input_pipeline(self, client_type: ClientType) -> SequentialPipelinePr
             # later steps
             CodegateSecrets(),
             CodegatePii(),
-            CodegateCli(),
+            CodegateCli(read_only=self.read_only),
             CodegateContextRetriever(),
             SystemPrompt(
                 Config.get_config().prompts.default_chat, Config.get_config().prompts.client_prompts

src/codegate/server.py

@@ -36,18 +36,25 @@ async def custom_error_handler(request, exc: Exception):
 
 class CodeGateServer(FastAPI):
     provider_registry: ProviderRegistry = None
+    read_only: bool = False
 
     def set_provider_registry(self, registry: ProviderRegistry):
         self.provider_registry = registry
 
 
-def init_app(pipeline_factory: PipelineFactory) -> CodeGateServer:
-    """Create the FastAPI application."""
+def init_app(pipeline_factory: PipelineFactory, read_only: bool = False) -> CodeGateServer:
+    """Create the FastAPI application.
+
+    Args:
+        pipeline_factory: The pipeline factory to use
+        read_only: If True, the v1 API will not be served (read-only mode)
+    """
     app = CodeGateServer(
         title="CodeGate",
         description=__description__,
         version=__version__,
     )
+    app.read_only = read_only
 
     @app.middleware("http")
     async def log_user_agent(request: Request, call_next):
@@ -123,14 +130,18 @@ async def health_check():
 
     app.include_router(system_router)
 
-    # CodeGate API
-    app.include_router(v1, prefix="/api/v1", tags=["CodeGate API"])
+    # CodeGate API - only include if not in read-only mode
+    if not read_only:
+        app.include_router(v1, prefix="/api/v1", tags=["CodeGate API"])
+        logger.info("V1 API enabled")
+    else:
+        logger.info("Running in read-only mode - V1 API disabled")
 
     return app
 
 
 def generate_openapi():
-    app = init_app(Mock(spec=PipelineFactory))
+    app = init_app(Mock(spec=PipelineFactory), read_only=False)
 
     # Generate OpenAPI JSON
     openapi_schema = app.openapi()

tests/test_server.py

@@ -48,7 +48,14 @@ def mock_pipeline_factory():
 @pytest.fixture
 def test_client(mock_pipeline_factory) -> TestClient:
     """Create a test client for the FastAPI application."""
-    app = init_app(mock_pipeline_factory)
+    app = init_app(mock_pipeline_factory, read_only=False)
+    return TestClient(app)
+
+
+@pytest.fixture
+def test_client_read_only(mock_pipeline_factory) -> TestClient:
+    """Create a test client for the FastAPI application in read-only mode."""
+    app = init_app(mock_pipeline_factory, read_only=True)
     return TestClient(app)
 
 
@@ -58,6 +65,12 @@ def test_app_initialization(mock_pipeline_factory) -> None:
     assert app is not None
     assert app.title == "CodeGate"
     assert app.version == __version__
+    assert hasattr(app, "read_only")
+    assert app.read_only is False
+
+    # Test with read_only=True
+    app_read_only = init_app(mock_pipeline_factory, read_only=True)
+    assert app_read_only.read_only is True
 
 
 def test_cors_middleware(mock_pipeline_factory) -> None:
@@ -96,6 +109,12 @@ def test_version_endpoint(mock_fetch_latest_version, test_client: TestClient) ->
     assert response_data["is_latest"] is False
 
 
+def test_version_endpoint_read_only(test_client_read_only: TestClient) -> None:
+    """Test the version endpoint is not available in read-only mode."""
+    response = test_client_read_only.get("/api/v1/version")
+    assert response.status_code == 404  # Should return 404 Not Found in read-only mode
+
+
 @patch("codegate.pipeline.secrets.manager.SecretsManager")
 @patch("codegate.server.get_provider_registry")
 def test_provider_registration(mock_registry, mock_secrets_mgr, mock_pipeline_factory) -> None:
@@ -145,6 +164,21 @@ def test_workspaces_routes(mock_pipeline_factory) -> None:
     assert len(dashboard_routes) > 0
 
 
+def test_read_only_mode(mock_pipeline_factory) -> None:
+    """Test that v1 API is not included in read-only mode."""
+    # Test with read_only=False (default)
+    app = init_app(mock_pipeline_factory, read_only=False)
+    routes = [route.path for route in app.routes]
+    v1_routes = [route for route in routes if route.startswith("/api/v1")]
+    assert len(v1_routes) > 0
+
+    # Test with read_only=True
+    app_read_only = init_app(mock_pipeline_factory, read_only=True)
+    routes_read_only = [route.path for route in app_read_only.routes]
+    v1_routes_read_only = [route for route in routes_read_only if route.startswith("/api/v1")]
+    assert len(v1_routes_read_only) == 0
+
+
 def test_system_routes(mock_pipeline_factory) -> None:
     """Test that system routes are included."""
     app = init_app(mock_pipeline_factory)
@@ -469,6 +503,37 @@ def test_serve_certificate_options(cli_runner: CliRunner) -> None:
             ), f"{key} does not match expected value"
 
 
+def test_serve_read_only_mode(cli_runner: CliRunner) -> None:
+    """Test serve command with read-only mode."""
+    with (
+        patch("src.codegate.cli.setup_logging") as mock_setup_logging,
+        patch("src.codegate.cli.init_app") as mock_init_app,
+    ):
+        # Mock the init_app function to return a MagicMock
+        mock_app = MagicMock()
+        mock_init_app.return_value = mock_app
+
+        # Execute CLI command with read-only flag
+        result = cli_runner.invoke(
+            cli,
+            [
+                "serve",
+                "--read-only",
+            ],
+        )
+
+        # Check the result of the command
+        assert result.exit_code == 0
+
+        # Ensure logging setup was called with expected arguments
+        mock_setup_logging.assert_called_once_with("INFO", "JSON")
+
+        # Verify that init_app was called with read_only=True
+        mock_init_app.assert_called_once()
+        _, kwargs = mock_init_app.call_args
+        assert kwargs.get("read_only") is True
+
+
 def test_main_function() -> None:
     """Test main function."""
     with patch("sys.argv", ["cli"]), patch("codegate.cli.cli") as mock_cli: