Coverage for src/lilbee/server/routes/general.py: 100%

1"""General routes: health, status, config, source, warm. 

2 

3Every route needs the token, ``/api/health`` included: it reports the chat 

4engine's last error, which carries model paths and loader failures. A local 

5probe reads the token from server.json like every other local client. 

6""" 

7 

8from __future__ import annotations 

9 

10import signal 

11from pathlib import Path 

12from typing import Any 

13 

14from litestar import Response, get, patch, post 

15from litestar.background_tasks import BackgroundTask 

16from litestar.exceptions import HTTPException, NotFoundException, ValidationException 

17from litestar.params import FromQuery 

18from litestar.response import Stream 

19from litestar.status_codes import HTTP_202_ACCEPTED, HTTP_503_SERVICE_UNAVAILABLE 

20from pydantic import ValidationError 

21 

22from lilbee.app.settings import config_write_failure_message 

23from lilbee.server import handlers 

24from lilbee.server.handlers.sse import SSE_MEDIA_TYPE 

25from lilbee.server.models import ( 

26 ConfigResponse, 

27 ConfigUpdateResponse, 

28 HealthResponse, 

29 ShutdownResponse, 

30 SourceContentResponse, 

31 StatusResponse, 

32) 

33 

34 

35@get("/api/health") 

36async def health_route() -> HealthResponse: 

37 """Service health check returning server version and uptime status.""" 

38 return await handlers.health() 

39 

40 

41@get("/api/warm/stream", media_type=SSE_MEDIA_TYPE) 

42async def warm_stream_route() -> Stream: 

43 """Stream chat-model cold-load progress as SSE for a launcher's warm indicator.""" 

44 return Stream(handlers.warm_stream(), media_type=SSE_MEDIA_TYPE) 

45 

46 

47@get("/api/status") 

48async def status_route() -> StatusResponse: 

49 """Current configuration, indexed document sources, and chunk counts.""" 

50 return await handlers.status() 

51 

52 

53@post("/api/shutdown", status_code=HTTP_202_ACCEPTED) 

54async def shutdown_route() -> Response[ShutdownResponse]: 

55 """Gracefully stop the server, exactly as an external SIGTERM would. 

56 

57 The signal rides a background task so it is raised after the response has 

58 been handed to the transport, rather than after a guessed delay that a 

59 slow flush could lose. 

60 """ 

61 return Response( 

62 await handlers.shutdown(), 

63 status_code=HTTP_202_ACCEPTED, 

64 background=BackgroundTask(signal.raise_signal, signal.SIGTERM), 

65 ) 

66 

67 

68@get("/api/config") 

69async def config_route() -> ConfigResponse: 

70 """Return all user-facing configuration values.""" 

71 return await handlers.get_config() 

72 

73 

74@get("/api/config/defaults") 

75async def config_defaults_route() -> ConfigResponse: 

76 """Return canonical defaults for every writable, public configuration field.""" 

77 return await handlers.get_config_defaults() 

78 

79 

80@patch("/api/config") 

81async def config_update_route(data: dict[str, Any]) -> ConfigUpdateResponse: 

82 """Partial update of writable configuration fields.""" 

83 try: 

84 return await handlers.update_config(data) 

85 except (ValueError, ValidationError) as exc: 

86 raise ValidationException(str(exc)) from exc 

87 except OSError as exc: 

88 raise HTTPException( 

89 status_code=HTTP_503_SERVICE_UNAVAILABLE, detail=config_write_failure_message(exc) 

90 ) from exc 

91 

92 

93@get("/api/source") 

94async def source_content_route( 

95 source: FromQuery[str], raw: FromQuery[bool] = False 

96) -> SourceContentResponse | Response[bytes]: 

97 """Return stored source file as JSON (``raw=0``) or raw bytes (``raw=1``).""" 

98 try: 

99 result = await handlers.get_source_content(source, raw=raw) 

100 except FileNotFoundError as exc: 

101 raise NotFoundException(f"source not found: {source}") from exc 

102 except ValueError as exc: 

103 raise ValidationException(str(exc)) from exc 

104 

105 # ``raw=True`` returns ``(bytes, content_type)``; narrow via ``isinstance`` 

106 # so mypy sees the tuple branch without leaning on ``type: ignore``. 

107 if isinstance(result, tuple): 

108 body, content_type = result 

109 # nosniff blocks browser MIME-sniffing fallbacks; attachment forces a 

110 # download for any type the handler degraded to octet-stream so 

111 # attacker-named files don't render inline anywhere. 

112 headers = {"X-Content-Type-Options": "nosniff"} 

113 if content_type == "application/octet-stream": 

114 headers["Content-Disposition"] = f'attachment; filename="{Path(source).name}"' 

115 return Response(content=body, media_type=content_type, status_code=200, headers=headers) 

116 return result