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_200_OK, HTTP_202_ACCEPTED, HTTP_503_SERVICE_UNAVAILABLE 

20from pydantic import ValidationError 

21 

22from lilbee.app.services import request_server_exit 

23from lilbee.app.settings import config_write_failure_message 

24from lilbee.server import handlers 

25from lilbee.server.content_disposition import CONTENT_DISPOSITION, attachment_disposition 

26from lilbee.server.handlers.sse import SSE_MEDIA_TYPE 

27from lilbee.server.models import ( 

28 ConfigResetRequest, 

29 ConfigResponse, 

30 ConfigSchemaResponse, 

31 ConfigSourcesResponse, 

32 ConfigUpdateResponse, 

33 HealthResponse, 

34 ShutdownResponse, 

35 SourceContentResponse, 

36 StatusResponse, 

37) 

38 

39 

40@get("/api/health") 

41async def health_route() -> HealthResponse: 

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

43 return await handlers.health() 

44 

45 

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

47async def warm_stream_route() -> Stream: 

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

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

50 

51 

52@get("/api/status") 

53async def status_route() -> StatusResponse: 

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

55 return await handlers.status() 

56 

57 

58async def _stop_server() -> None: 

59 """Stop the serving loop when one runs; otherwise raise SIGTERM.""" 

60 if not request_server_exit(): 

61 signal.raise_signal(signal.SIGTERM) 

62 

63 

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

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

66 """Gracefully stop the server through its serving loop. 

67 

68 The stop rides a background task so it runs after the response has 

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

70 slow flush could lose. 

71 """ 

72 return Response( 

73 await handlers.shutdown(), 

74 status_code=HTTP_202_ACCEPTED, 

75 background=BackgroundTask(_stop_server), 

76 ) 

77 

78 

79@get("/api/config") 

80async def config_route() -> ConfigResponse: 

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

82 return await handlers.get_config() 

83 

84 

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

86async def config_defaults_route() -> ConfigResponse: 

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

88 return await handlers.get_config_defaults() 

89 

90 

91@post("/api/config/reset", status_code=HTTP_200_OK) 

92async def config_reset_route(data: ConfigResetRequest) -> ConfigUpdateResponse: 

93 """Remove the listed keys from config.toml; each falls back to the next source.""" 

94 try: 

95 return await handlers.reset_config(data.keys) 

96 except ValueError as exc: 

97 raise ValidationException(str(exc)) from exc 

98 except OSError as exc: 

99 raise HTTPException( 

100 status_code=HTTP_503_SERVICE_UNAVAILABLE, detail=config_write_failure_message(exc) 

101 ) from exc 

102 

103 

104@get("/api/config/schema") 

105async def config_schema_route() -> ConfigSchemaResponse: 

106 """Return type, choices, writability and reindex metadata for every public field.""" 

107 return await handlers.get_config_schema() 

108 

109 

110@get("/api/config/sources") 

111async def config_sources_route() -> ConfigSourcesResponse: 

112 """Return the layer that supplies each configuration value.""" 

113 return await handlers.get_config_sources() 

114 

115 

116@patch("/api/config") 

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

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

119 try: 

120 return await handlers.update_config(data) 

121 except (ValueError, ValidationError) as exc: 

122 raise ValidationException(str(exc)) from exc 

123 except OSError as exc: 

124 raise HTTPException( 

125 status_code=HTTP_503_SERVICE_UNAVAILABLE, detail=config_write_failure_message(exc) 

126 ) from exc 

127 

128 

129@get("/api/source") 

130async def source_content_route( 

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

132) -> SourceContentResponse | Response[bytes]: 

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

134 try: 

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

136 except FileNotFoundError as exc: 

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

138 except ValueError as exc: 

139 raise ValidationException(str(exc)) from exc 

140 

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

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

143 if isinstance(result, tuple): 

144 body, content_type = result 

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

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

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

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

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

150 headers[CONTENT_DISPOSITION] = attachment_disposition(Path(source).name) 

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

152 return result