Coverage for src / lexigram / ai / relay / finish_reasons.py: 100%

1"""Unified finish-reason and Responses-status mapping tables. 

2 

3Every canonical ``finish_reason`` -> wire translation for the OpenAI chat, 

4OpenAI Responses, Claude, and Gemini wire formats lives here so mappers and 

5emitters never duplicate or drift. Unknown canonical values fall back to a 

6safe terminal value rather than raising ``KeyError``; callers that can record 

7diagnostics attach a ``finish_reason_adapted`` loss around the fallback. 

8""" 

9 

10from __future__ import annotations 

11 

12from collections.abc import Mapping 

13 

14from lexigram.contracts.ai.relay.ir import normalize_finish_reason 

15from lexigram.contracts.ai.relay.types import RelayFormat 

16 

17__all__ = [ 

18 "FINISH_REASON_TO_WIRE", 

19 "RESPONSES_STATUS_FROM_FINISH", 

20 "finish_reason_to_wire", 

21 "normalize_finish_reason", 

22 "responses_incomplete_from_finish", 

23 "responses_status_from_finish", 

24] 

25 

26#: Canonical finish reason -> wire value per target format. 

27FINISH_REASON_TO_WIRE: Mapping[str, Mapping[RelayFormat, str]] = { 

28 "stop": { 

29 RelayFormat.OPENAI_CHAT: "stop", 

30 RelayFormat.CLAUDE: "end_turn", 

31 RelayFormat.GEMINI: "STOP", 

32 }, 

33 "length": { 

34 RelayFormat.OPENAI_CHAT: "length", 

35 RelayFormat.CLAUDE: "max_tokens", 

36 RelayFormat.GEMINI: "MAX_TOKENS", 

37 }, 

38 "tool_calls": { 

39 RelayFormat.OPENAI_CHAT: "tool_calls", 

40 RelayFormat.CLAUDE: "tool_use", 

41 RelayFormat.GEMINI: "STOP", 

42 }, 

43 "function_call": { 

44 RelayFormat.OPENAI_CHAT: "function_call", 

45 RelayFormat.CLAUDE: "tool_use", 

46 RelayFormat.GEMINI: "STOP", 

47 }, 

48 "content_filter": { 

49 RelayFormat.OPENAI_CHAT: "content_filter", 

50 RelayFormat.CLAUDE: "end_turn", 

51 RelayFormat.GEMINI: "SAFETY", 

52 }, 

53 "other": { 

54 RelayFormat.OPENAI_CHAT: "other", 

55 RelayFormat.CLAUDE: "end_turn", 

56 RelayFormat.GEMINI: "OTHER", 

57 }, 

58} 

59 

60#: Safe terminal value per format for canonical values absent from the table. 

61_WIRE_FALLBACK: Mapping[RelayFormat, str] = { 

62 RelayFormat.OPENAI_CHAT: "stop", 

63 RelayFormat.CLAUDE: "end_turn", 

64 RelayFormat.GEMINI: "OTHER", 

65} 

66 

67#: Canonical finish reason -> (Responses status, incomplete-detail reason). 

68RESPONSES_STATUS_FROM_FINISH: Mapping[str, tuple[str, str | None]] = { 

69 "stop": ("completed", None), 

70 "tool_calls": ("completed", None), 

71 "function_call": ("completed", None), 

72 "length": ("incomplete", "max_output_tokens"), 

73 "content_filter": ("incomplete", "content_filter"), 

74 "other": ("incomplete", "other"), 

75} 

76 

77 

78def finish_reason_to_wire(finish_reason: str | None, target: RelayFormat) -> str: 

79 """Map a canonical finish reason onto ``target``'s wire value. 

80 

81 Unknown or missing values fall back to the target's terminal value 

82 rather than raising ``KeyError``. 

83 """ 

84 if finish_reason is None: 

85 return _WIRE_FALLBACK[target] 

86 row = FINISH_REASON_TO_WIRE.get(finish_reason) 

87 if row is None: 

88 return _WIRE_FALLBACK[target] 

89 return row.get(target, _WIRE_FALLBACK[target]) 

90 

91 

92def responses_status_from_finish( 

93 finish_reason: str | None, 

94) -> tuple[str, str | None]: 

95 """Derive a Responses status and incomplete-detail reason.""" 

96 if finish_reason is None: 

97 return "completed", None 

98 return RESPONSES_STATUS_FROM_FINISH.get(finish_reason, ("completed", None)) 

99 

100 

101def responses_incomplete_from_finish(finish_reason: str | None) -> str | None: 

102 """Return the Responses incomplete-detail reason, or ``None``.""" 

103 if finish_reason is None: 

104 return None 

105 _status, detail = RESPONSES_STATUS_FROM_FINISH.get( 

106 finish_reason, ("completed", None) 

107 ) 

108 return detail