Coverage for vibefpmgui/ai_context.py: 93%

101 statements  

« prev     ^ index     » next       coverage.py v7.16.2, created at 2026-11-01 20:51 +0000

1"""What a chat with the AI sends the model each time, so a conversation 

2fits in its context window (Ollama doesn't fail when it doesn't: it drops 

3the start, system prompt included). 

4 

5- **The file once per version.** A question carries the open file only if 

6 it changed since it was last sent; earlier copies in the history are 

7 replaced by a placeholder. 

8- **What became of the edits** proposed in an answer is said at the start 

9 of the next question (see `vibefpmgui.ai_edits.report`); answers are sent 

10 back as the model wrote them. 

11- **The oldest turns go first** when the estimate is over the budget (75% 

12 of the context, the rest left for the answer); the system prompt, the 

13 current file and the new question always stay. 

14- **Big files are sent in part:** the selection, or the procedure around 

15 the cursor, in full, plus an outline of the rest (`file_attachment`). 

16 

17Pure functions (no Qt). 

18""" 

19 

20from __future__ import annotations 

21 

22import re 

23from dataclasses import dataclass, field 

24 

25from .ai_edits import Edit, report 

26 

27BUDGET = 0.75 # of the context window, for what's sent (the rest: the answer) 

28BIG_FILE_LINES = 600 # bigger files are sent in part 

29BIG_FILE_SHARE = 0.4 # …and so are files taking more than this share of the budget 

30WINDOW_PADDING = 20 # lines shown around the selection or procedure 

31WINDOW_MAX_LINES = 400 

32 

33PROCEDURE_RE = re.compile(r"^(?!\s*end\b)\s*(?:[\w(),=*\s]*?\s)?(subroutine|function)\s+\w+", re.IGNORECASE) 

34PROCEDURE_END_RE = re.compile(r"^\s*end\s*(subroutine|function)\b|^\s*end\s*(!.*)?$", re.IGNORECASE) 

35OUTLINE_RE = re.compile( 

36 r"^\s*(module|submodule|program|use|implicit|contains|interface|abstract\s+interface|" 

37 r"end\s*(module|submodule|program|interface|type|subroutine|function)\b|" 

38 r"type\s*(,|::|\s\w)|(?!end\b)(?:[\w(),=*\s]*?\s)?(subroutine|function)\s+\w+)", 

39 re.IGNORECASE) 

40 

41 

42def estimate_tokens(text: str, ratio: float = 1.0) -> int: 

43 """A rough token count (about 3 characters each for code), scaled by 

44 ``ratio``: measured counts over estimated ones, from earlier answers.""" 

45 return int(len(text) / 3 * ratio) + 4 

46 

47 

48@dataclass 

49class Attachment: 

50 """The open file as sent with a question: all of it, or part of it.""" 

51 name: str 

52 file_text: str # the whole file when it was sent 

53 shown: str # what the prompt contains 

54 partial: bool = False 

55 

56 

57@dataclass 

58class Turn: 

59 question: str # as the user typed it 

60 request: str # what's sent besides the file (the question, plus notes) 

61 attachment: Attachment | None = None # the file wanted with it (sent if it changed) 

62 answer: str = "" 

63 edits: list[Edit] = field(default_factory=list) 

64 

65 

66def placeholder(attachment: Attachment) -> str: 

67 return f"[{attachment.name} as it was then: replaced by the newer version below]" 

68 

69 

70def compose(past: list[Turn], turn: Turn, system: str, context: int, 

71 ratio: float = 1.0) -> tuple[list[dict], int, int]: 

72 """The messages for ``turn`` after the ``past`` ones: (messages, the 

73 estimated tokens, how many of the oldest turns were left out).""" 

74 kept = list(past) 

75 budget = int(context * BUDGET) 

76 while True: 

77 messages = _render(kept, turn, system) 

78 tokens = sum(estimate_tokens(m["content"], ratio) for m in messages) 

79 if tokens <= budget or not kept: 

80 return messages, tokens, len(past) - len(kept) 

81 kept.pop(0) 

82 

83 

84def _render(kept: list[Turn], turn: Turn, system: str) -> list[dict]: 

85 last = next((t.attachment for t in reversed(kept) if t.attachment is not None), None) 

86 send = turn.attachment is not None and turn.attachment != last 

87 full = None if send else next((i for i in range(len(kept) - 1, -1, -1) if kept[i].attachment), None) 

88 messages = [{"role": "system", "content": system}] 

89 previous: Turn | None = None 

90 for i, t in enumerate(kept): 

91 user = _with_report(previous, t.request) 

92 if t.attachment is not None: 

93 user += "\n\n" + (t.attachment.shown if i == full else placeholder(t.attachment)) 

94 messages.append({"role": "user", "content": user}) 

95 messages.append({"role": "assistant", "content": t.answer}) 

96 previous = t 

97 request = _with_report(previous, turn.request) 

98 if send: 

99 request += "\n\n" + turn.attachment.shown 

100 elif turn.attachment is not None: 

101 request += f"\n\n(My file, {turn.attachment.name}, is as I sent it above.)" 

102 messages.append({"role": "user", "content": request}) 

103 return messages 

104 

105 

106def _with_report(previous: Turn | None, request: str) -> str: 

107 note = report(previous.edits) if previous is not None else "" 

108 return f"{note}\n\n{request}" if note else request 

109 

110 

111# ------------------------------------------------------------- the file 

112def numbered(lines: list[str], first: int = 1) -> str: 

113 width = len(str(first + len(lines) - 1)) 

114 return "\n".join(f"{n:>{width}}| {line}" for n, line in enumerate(lines, first)) 

115 

116 

117def file_lines(text: str) -> list[str]: 

118 lines = text.split("\n") 

119 if len(lines) > 1 and lines[-1] == "": 119 ↛ 121line 119 didn't jump to line 121 because the condition on line 119 was always true

120 lines.pop() # (a final newline doesn't start another line) 

121 return lines 

122 

123 

124def file_attachment(name: str, text: str, numbers: bool, context: int, ratio: float = 1.0, 

125 selection: tuple[int, int] | None = None, cursor_line: int = 1) -> Attachment: 

126 """The file as sent: with line numbers (for answers that point at lines) 

127 or as it is (for edits, whose old text must be copied exactly). A big 

128 file is sent in part: the ``selection`` (first, last line) or the 

129 procedure around ``cursor_line``, plus an outline of the rest.""" 

130 lines = file_lines(text) 

131 whole = numbered(lines) if numbers else "\n".join(lines) 

132 big = len(lines) > BIG_FILE_LINES or estimate_tokens(whole, ratio) > context * BUDGET * BIG_FILE_SHARE 

133 if not big: 

134 how = "line numbers on the left" if numbers else "exactly as it is" 

135 return Attachment(name, text, f"My file, {name} ({how}):\n\n```fortran\n{whole}\n```") 

136 first, last = window(lines, selection, cursor_line) 

137 part = lines[first - 1:last] 

138 shown = numbered(part, first) if numbers else "\n".join(part) 

139 outline = [(n, line) for n, line in enumerate(lines, 1) 

140 if not first <= n <= last and OUTLINE_RE.match(line)] 

141 outline_text = "\n".join(f"line {n}: {line.strip()}" for n, line in outline) or "(nothing)" 

142 return Attachment(name, text, ( 

143 f"My file, {name}, has {len(lines)} lines: too long to send all of it. Only lines " 

144 f"{first}–{last} are shown in full; for the rest, only an outline. Base changes on the " 

145 f"lines shown in full.\n\nOutline of the rest:\n{outline_text}\n\n" 

146 f"Lines {first}–{last}{' (line numbers on the left)' if numbers else ''}:\n\n" 

147 f"```fortran\n{shown}\n```"), partial=True) 

148 

149 

150def window(lines: list[str], selection: tuple[int, int] | None, cursor_line: int) -> tuple[int, int]: 

151 """The lines to send in full (1-based, inclusive): the selection, or the 

152 procedure around the cursor, with some lines around it.""" 

153 if selection is not None: 153 ↛ 154line 153 didn't jump to line 154 because the condition on line 153 was never true

154 first, last = selection 

155 else: 

156 first, last = enclosing_procedure(lines, cursor_line) 

157 first, last = max(1, first - WINDOW_PADDING), min(len(lines), last + WINDOW_PADDING) 

158 if last - first + 1 > WINDOW_MAX_LINES: # (a huge procedure: around the cursor in it) 158 ↛ 159line 158 didn't jump to line 159 because the condition on line 158 was never true

159 center = cursor_line if selection is None else selection[0] 

160 first = max(1, min(center - WINDOW_MAX_LINES // 2, len(lines) - WINDOW_MAX_LINES + 1)) 

161 last = min(len(lines), first + WINDOW_MAX_LINES - 1) 

162 return first, last 

163 

164 

165def enclosing_procedure(lines: list[str], line: int) -> tuple[int, int]: 

166 """The subroutine or function around ``line`` (first, last), or the line 

167 itself when it's in none.""" 

168 line = min(max(1, line), max(1, len(lines))) 

169 start = next((n for n in range(line, 0, -1) if PROCEDURE_RE.match(lines[n - 1])), None) 

170 if start is None or any(PROCEDURE_END_RE.match(lines[n - 1]) for n in range(start + 1, line)): 

171 return line, line # (in none, or between two) 

172 end = next((n for n in range(max(line, start + 1), len(lines) + 1) if PROCEDURE_END_RE.match(lines[n - 1])), 

173 None) 

174 if end is None or (end < line): 174 ↛ 175line 174 didn't jump to line 175 because the condition on line 174 was never true

175 return line, line 

176 return start, end