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
« 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).
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`).
17Pure functions (no Qt).
18"""
20from __future__ import annotations
22import re
23from dataclasses import dataclass, field
25from .ai_edits import Edit, report
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
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)
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
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
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)
66def placeholder(attachment: Attachment) -> str:
67 return f"[{attachment.name} as it was then: replaced by the newer version below]"
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)
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
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
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))
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
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)
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
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