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.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 ConfigResponse, 

29 ConfigSchemaResponse, 

30 ConfigUpdateResponse, 

31 HealthResponse, 

32 ShutdownResponse, 

33 SourceContentResponse, 

34 StatusResponse, 

35) 

36 

37 

38@get("/api/health") 

39async def health_route() -> HealthResponse: 

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

41 return await handlers.health() 

42 

43 

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

45async def warm_stream_route() -> Stream: 

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

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

48 

49 

50@get("/api/status") 

51async def status_route() -> StatusResponse: 

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

53 return await handlers.status() 

54 

55 

56async def _stop_server() -> None: 

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

58 if not request_server_exit(): 

59 signal.raise_signal(signal.SIGTERM) 

60 

61 

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

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

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

65 

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

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

68 slow flush could lose. 

69 """ 

70 return Response( 

71 await handlers.shutdown(), 

72 status_code=HTTP_202_ACCEPTED, 

73 background=BackgroundTask(_stop_server), 

74 ) 

75 

76 

77@get("/api/config") 

78async def config_route() -> ConfigResponse: 

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

80 return await handlers.get_config() 

81 

82 

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

84async def config_defaults_route() -> ConfigResponse: 

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

86 return await handlers.get_config_defaults() 

87 

88 

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

90async def config_schema_route() -> ConfigSchemaResponse: 

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

92 return await handlers.get_config_schema() 

93 

94 

95@patch("/api/config") 

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

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

98 try: 

99 return await handlers.update_config(data) 

100 except (ValueError, ValidationError) as exc: 

101 raise ValidationException(str(exc)) from exc 

102 except OSError as exc: 

103 raise HTTPException( 

104 status_code=HTTP_503_SERVICE_UNAVAILABLE, detail=config_write_failure_message(exc) 

105 ) from exc 

106 

107 

108@get("/api/source") 

109async def source_content_route( 

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

111) -> SourceContentResponse | Response[bytes]: 

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

113 try: 

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

115 except FileNotFoundError as exc: 

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

117 except ValueError as exc: 

118 raise ValidationException(str(exc)) from exc 

119 

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

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

122 if isinstance(result, tuple): 

123 body, content_type = result 

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

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

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

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

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

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

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

131 return result