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 NotFoundException, ValidationException 

17from litestar.params import FromQuery 

18from litestar.response import Stream 

19from litestar.status_codes import HTTP_202_ACCEPTED 

20from pydantic import ValidationError 

21 

22from lilbee.server import handlers 

23from lilbee.server.handlers.sse import SSE_MEDIA_TYPE 

24from lilbee.server.models import ( 

25 ConfigResponse, 

26 ConfigUpdateResponse, 

27 HealthResponse, 

28 ShutdownResponse, 

29 SourceContentResponse, 

30 StatusResponse, 

31) 

32 

33 

34@get("/api/health") 

35async def health_route() -> HealthResponse: 

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

37 return await handlers.health() 

38 

39 

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

41async def warm_stream_route() -> Stream: 

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

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

44 

45 

46@get("/api/status") 

47async def status_route() -> StatusResponse: 

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

49 return await handlers.status() 

50 

51 

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

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

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

55 

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

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

58 slow flush could lose. 

59 """ 

60 return Response( 

61 await handlers.shutdown(), 

62 status_code=HTTP_202_ACCEPTED, 

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

64 ) 

65 

66 

67@get("/api/config") 

68async def config_route() -> ConfigResponse: 

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

70 return await handlers.get_config() 

71 

72 

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

74async def config_defaults_route() -> ConfigResponse: 

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

76 return await handlers.get_config_defaults() 

77 

78 

79@patch("/api/config") 

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

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

82 try: 

83 return await handlers.update_config(data) 

84 except (ValueError, ValidationError) as exc: 

85 raise ValidationException(str(exc)) from exc 

86 

87 

88@get("/api/source") 

89async def source_content_route( 

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

91) -> SourceContentResponse | Response[bytes]: 

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

93 try: 

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

95 except FileNotFoundError as exc: 

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

97 except ValueError as exc: 

98 raise ValidationException(str(exc)) from exc 

99 

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

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

102 if isinstance(result, tuple): 

103 body, content_type = result 

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

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

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

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

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

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

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

111 return result