Pass proxy auth token to Telegram CLI sessions

api/runtime.py

@@ -255,6 +255,7 @@ class AppRuntime:
             allowed_dirs=allowed_dirs,
             plans_directory=plans_directory,
             claude_bin=self.settings.claude_cli_bin,
+            auth_token=getattr(self.settings, "anthropic_auth_token", ""),
             log_raw_cli_diagnostics=self.settings.log_raw_cli_diagnostics,
             log_messaging_error_details=self.settings.log_messaging_error_details,
         )

cli/manager.py

@@ -29,6 +29,7 @@ class CLISessionManager:
         allowed_dirs: list[str] | None = None,
         plans_directory: str | None = None,
         claude_bin: str = "claude",
+        auth_token: str = "",
         *,
         log_raw_cli_diagnostics: bool = False,
         log_messaging_error_details: bool = False,
@@ -47,6 +48,7 @@ class CLISessionManager:
         self.allowed_dirs = allowed_dirs or []
         self.plans_directory = plans_directory
         self.claude_bin = claude_bin
+        self.auth_token = auth_token
         self._log_raw_cli_diagnostics = log_raw_cli_diagnostics
         self._log_messaging_error_details = log_messaging_error_details
 
@@ -82,6 +84,7 @@ class CLISessionManager:
                 allowed_dirs=self.allowed_dirs,
                 plans_directory=self.plans_directory,
                 claude_bin=self.claude_bin,
+                auth_token=self.auth_token,
                 log_raw_cli_diagnostics=self._log_raw_cli_diagnostics,
             )
             self._pending_sessions[temp_id] = new_session

cli/session.py

@@ -26,6 +26,7 @@ class ClaudeCliConfig:
     allowed_dirs: list[str] = field(default_factory=list)
     plans_directory: str | None = None
     claude_bin: str = "claude"
+    auth_token: str = ""
 
 
 class CLISession:
@@ -38,6 +39,7 @@ class CLISession:
         allowed_dirs: list[str] | None = None,
         plans_directory: str | None = None,
         claude_bin: str = "claude",
+        auth_token: str = "",
         *,
         log_raw_cli_diagnostics: bool = False,
     ):
@@ -47,12 +49,14 @@ class CLISession:
             allowed_dirs=[os.path.normpath(d) for d in (allowed_dirs or [])],
             plans_directory=plans_directory,
             claude_bin=claude_bin,
+            auth_token=auth_token,
         )
         self.workspace = self.config.workspace_path
         self.api_url = self.config.api_url
         self.allowed_dirs = self.config.allowed_dirs
         self.plans_directory = self.config.plans_directory
         self.claude_bin = self.config.claude_bin
+        self.auth_token = self.config.auth_token
         self._log_raw_cli_diagnostics = log_raw_cli_diagnostics
         self.process: asyncio.subprocess.Process | None = None
         self.current_session_id: str | None = None
@@ -117,6 +121,10 @@ class CLISession:
                 env["ANTHROPIC_BASE_URL"] = self.api_url[:-3]
             else:
                 env["ANTHROPIC_BASE_URL"] = self.api_url
+            if token := self.auth_token.strip():
+                env["ANTHROPIC_AUTH_TOKEN"] = token
+            else:
+                env.pop("ANTHROPIC_AUTH_TOKEN", None)
 
             env["TERM"] = "dumb"
             env["PYTHONIOENCODING"] = "utf-8"

tests/cli/test_cli.py

@@ -506,6 +506,55 @@ class TestCLISession:
             env = kwargs["env"]
             assert env["ANTHROPIC_BASE_URL"] == "http://localhost:8082"
 
+    @pytest.mark.asyncio
+    async def test_start_task_sets_proxy_auth_token(self):
+        """Test start_task forwards configured proxy auth to Claude Code."""
+        from cli.session import CLISession
+
+        session = CLISession(
+            "/tmp", "http://localhost:8082/v1", auth_token="proxy-token"
+        )
+
+        mock_process = AsyncMock()
+        mock_process.stdout.read.side_effect = [b""]
+        mock_process.stderr.read.return_value = b""
+        mock_process.wait.return_value = 0
+
+        with patch(
+            "asyncio.create_subprocess_exec", new_callable=AsyncMock
+        ) as mock_exec:
+            mock_exec.return_value = mock_process
+            async for _ in session.start_task("test"):
+                pass
+
+            env = mock_exec.call_args.kwargs["env"]
+            assert env["ANTHROPIC_AUTH_TOKEN"] == "proxy-token"
+
+    @pytest.mark.asyncio
+    async def test_start_task_removes_stale_auth_token_when_proxy_auth_blank(self):
+        """Test start_task does not leak inherited Claude auth into proxy calls."""
+        from cli.session import CLISession
+
+        session = CLISession("/tmp", "http://localhost:8082/v1", auth_token="")
+
+        mock_process = AsyncMock()
+        mock_process.stdout.read.side_effect = [b""]
+        mock_process.stderr.read.return_value = b""
+        mock_process.wait.return_value = 0
+
+        with (
+            patch.dict(os.environ, {"ANTHROPIC_AUTH_TOKEN": "stale"}, clear=False),
+            patch(
+                "asyncio.create_subprocess_exec", new_callable=AsyncMock
+            ) as mock_exec,
+        ):
+            mock_exec.return_value = mock_process
+            async for _ in session.start_task("test"):
+                pass
+
+            env = mock_exec.call_args.kwargs["env"]
+            assert "ANTHROPIC_AUTH_TOKEN" not in env
+
     @pytest.mark.asyncio
     async def test_start_task_allowed_dirs(self):
         """Test start_task includes allowed dirs in command."""

tests/cli/test_cli_manager_edge_cases.py

@@ -13,10 +13,14 @@ async def test_register_real_session_id_moves_pending_to_active_and_maps():
         mock_session.stop = AsyncMock(return_value=True)
         mock_session_cls.return_value = mock_session
 
-        manager = CLISessionManager(workspace_path="/tmp", api_url="http://x/v1")
+        manager = CLISessionManager(
+            workspace_path="/tmp", api_url="http://x/v1", auth_token="proxy-token"
+        )
         session, temp_id, is_new = await manager.get_or_create_session()
         assert session is mock_session
         assert is_new is True
+        mock_session_cls.assert_called_once()
+        assert mock_session_cls.call_args.kwargs["auth_token"] == "proxy-token"
 
         ok = await manager.register_real_session_id(temp_id, "real_1")
         assert ok is True