|
20 | 20 |
|
21 | 21 | * **Marker items track the boundary.** Each compaction writes the summary into |
22 | 22 | the conversation as a recognizable *marker* message (a message whose text |
23 | | - starts with ``MARKER_SENTINEL``). The items after the last marker are the |
24 | | - recent verbatim turns; the marker texts are the additive summaries. This is |
| 23 | + starts with ``MARKER_SENTINEL``) recording how many leading items it covers. |
| 24 | + The items past that count are the recent verbatim turns; the marker texts are |
| 25 | + the additive summaries. This is |
25 | 26 | lightspeed's own bookkeeping — OGX never interprets it (we no longer |
26 | 27 | pass ``conversation`` to inference once a marker exists). |
27 | 28 |
|
|
42 | 43 | """ |
43 | 44 |
|
44 | 45 | import asyncio |
| 46 | +import re |
45 | 47 | from collections.abc import AsyncIterator, Sequence |
46 | 48 | from contextlib import asynccontextmanager |
47 | 49 | from dataclasses import dataclass |
|
92 | 94 | the model (:func:`_summary_input_message`). |
93 | 95 | """ |
94 | 96 |
|
| 97 | +MARKER_COVERS_PREFIX = "[covers:" |
| 98 | +"""Prefix of the item count a marker records, e.g. ``[covers:4]``. |
| 99 | +
|
| 100 | +A marker is appended to the end of the conversation, after the turns the |
| 101 | +compaction kept verbatim, so its position does not mark the boundary between |
| 102 | +summarized history and recent turns. The count does: it is how many leading |
| 103 | +conversation items the summary covers (LCORE-4219). Markers written before |
| 104 | +that fix carry no count and fall back to their position. |
| 105 | +""" |
| 106 | + |
95 | 107 |
|
96 | 108 | # Per-conversation locks (R11). A request that triggers compaction holds the |
97 | 109 | # conversation's lock across the summarization LLM call so concurrent requests |
@@ -197,22 +209,71 @@ def is_marker_item(item: Any) -> bool: |
197 | 209 | return extract_message_text(item).startswith(MARKER_SENTINEL) |
198 | 210 |
|
199 | 211 |
|
| 212 | +def _split_marker_text(item: Any) -> tuple[Optional[int], str]: |
| 213 | + """Split a marker item into its covered-item count and its summary text. |
| 214 | +
|
| 215 | + The count is absent on markers written before LCORE-4219; the caller then |
| 216 | + falls back to the marker's position in the conversation. |
| 217 | + """ |
| 218 | + body = extract_message_text(item)[len(MARKER_SENTINEL) :].strip() |
| 219 | + match = re.match(rf"{re.escape(MARKER_COVERS_PREFIX)}(\d+)\]\s*", body) |
| 220 | + if match is None: |
| 221 | + return None, body |
| 222 | + return int(match.group(1)), body[match.end() :] |
| 223 | + |
| 224 | + |
200 | 225 | def _summary_text_of(item: Any) -> str: |
201 | 226 | """Extract the summary text from a marker item (sentinel stripped).""" |
202 | | - return extract_message_text(item)[len(MARKER_SENTINEL) :].strip() |
| 227 | + return _split_marker_text(item)[1] |
203 | 228 |
|
204 | 229 |
|
205 | 230 | def _items_after_last_marker(items: list[Any]) -> list[Any]: |
206 | | - """Return the conversation items that follow the last summary marker. |
| 231 | + """Return the conversation items that are not covered by any summary. |
| 232 | +
|
| 233 | + These are the recent turns replayed verbatim. When there is no marker the |
| 234 | + whole list is returned (no compaction has happened yet). |
| 235 | +
|
| 236 | + A marker records how many leading items its summary covers, because it is |
| 237 | + appended to the end of the conversation, *after* the turns the compaction |
| 238 | + deliberately kept verbatim (the buffer). Taking the items that follow the |
| 239 | + marker's position would drop those turns from every later request |
| 240 | + (LCORE-4219): they are in no summary and would never reach the model |
| 241 | + again. Markers written before that fix carry no count, so they keep the |
| 242 | + old position-based boundary. |
207 | 243 |
|
208 | | - These are the recent turns kept verbatim. When there is no marker the whole |
209 | | - list is returned (no compaction has happened yet). |
| 244 | + Marker items themselves are never part of the result: their text is |
| 245 | + replayed as a summary, not as a conversation turn. |
210 | 246 | """ |
211 | | - last = -1 |
| 247 | + last_index = -1 |
| 248 | + covered: Optional[int] = None |
212 | 249 | for index, item in enumerate(items): |
213 | 250 | if is_marker_item(item): |
214 | | - last = index |
215 | | - return items[last + 1 :] |
| 251 | + last_index = index |
| 252 | + covered, _ = _split_marker_text(item) |
| 253 | + if last_index < 0: |
| 254 | + return items |
| 255 | + boundary = covered if covered is not None else last_index + 1 |
| 256 | + return [item for item in items[boundary:] if not is_marker_item(item)] |
| 257 | + |
| 258 | + |
| 259 | +def _covered_item_count(items: list[Any], keep_items: list[Any]) -> int: |
| 260 | + """Return how many leading items of *items* a new summary covers. |
| 261 | +
|
| 262 | + That is the position of the first kept turn, so the count is a raw index |
| 263 | + into the stored items, markers included. It cannot be derived by |
| 264 | + subtracting the kept turns from the total: markers are filtered out of the |
| 265 | + recent turns, so an older marker may sit among the items being kept, and |
| 266 | + subtracting would then place the boundary past a turn that is still |
| 267 | + verbatim (which would drop it, the bug LCORE-4219 fixes). An empty buffer |
| 268 | + means the summary covers everything stored so far. |
| 269 | + """ |
| 270 | + if not keep_items: |
| 271 | + return len(items) |
| 272 | + first_kept = keep_items[0] |
| 273 | + for index, item in enumerate(items): |
| 274 | + if item is first_kept: |
| 275 | + return index |
| 276 | + return len(items) - len(keep_items) |
216 | 277 |
|
217 | 278 |
|
218 | 279 | def _marker_summaries(items: list[Any]) -> list[str]: |
@@ -361,16 +422,26 @@ async def _write_summary_marker( |
361 | 422 | client: AsyncOgxClient, |
362 | 423 | conversation_id: str, |
363 | 424 | summary_text: str, |
| 425 | + covered_items: int, |
364 | 426 | ) -> None: |
365 | | - """Write the summary into the conversation as a recognizable marker message.""" |
| 427 | + """Write the summary into the conversation as a recognizable marker message. |
| 428 | +
|
| 429 | + ``covered_items`` is how many leading conversation items the summary |
| 430 | + covers. It is recorded in the marker because the marker is appended to the |
| 431 | + end of the conversation, after the turns kept verbatim in the buffer, so |
| 432 | + the marker's own position cannot mark the boundary (LCORE-4219). |
| 433 | + """ |
366 | 434 | await client.items.create( |
367 | 435 | conversation_id, |
368 | 436 | add_items_request=build_add_items_request( |
369 | 437 | [ |
370 | 438 | { |
371 | 439 | "type": "message", |
372 | 440 | "role": "user", |
373 | | - "content": f"{MARKER_SENTINEL} {summary_text}", |
| 441 | + "content": ( |
| 442 | + f"{MARKER_SENTINEL} {MARKER_COVERS_PREFIX}{covered_items}] " |
| 443 | + f"{summary_text}" |
| 444 | + ), |
374 | 445 | } |
375 | 446 | ] |
376 | 447 | ), |
@@ -516,8 +587,18 @@ async def _persist_new_summary_chunk( # pylint: disable=too-many-arguments,too- |
516 | 587 | user_id: str, |
517 | 588 | skip_user_id_check: bool, |
518 | 589 | ) -> None: |
519 | | - """Persist a fresh summary chunk: write the OGX marker + best-effort cache.""" |
520 | | - await _write_summary_marker(client, conversation_id, summary.summary_text) |
| 590 | + """Persist a fresh summary chunk: write the OGX marker + best-effort cache. |
| 591 | +
|
| 592 | + ``summary.summarized_through_turn`` is the number of leading conversation |
| 593 | + items the summary covers; it is recorded in the marker so later requests |
| 594 | + keep replaying the buffered turns (LCORE-4219). |
| 595 | + """ |
| 596 | + await _write_summary_marker( |
| 597 | + client, |
| 598 | + conversation_id, |
| 599 | + summary.summary_text, |
| 600 | + summary.summarized_through_turn, |
| 601 | + ) |
521 | 602 | _store_cached_summary(cache, user_id, conversation_id, summary, skip_user_id_check) |
522 | 603 |
|
523 | 604 |
|
@@ -659,12 +740,11 @@ async def apply_compaction( # pylint: disable=too-many-arguments,too-many-posit |
659 | 740 | encoding_name=encoding_name, |
660 | 741 | ) |
661 | 742 | if old_items: |
662 | | - already = len(items) - len(recent_items) |
663 | 743 | summary = await summarize_chunk( |
664 | 744 | client, |
665 | 745 | model, |
666 | 746 | old_items, |
667 | | - summarized_through_turn=already + len(old_items), |
| 747 | + summarized_through_turn=_covered_item_count(items, keep_items), |
668 | 748 | encoding_name=encoding_name, |
669 | 749 | ) |
670 | 750 | await _persist_new_summary_chunk( |
|
0 commit comments