Conversation History

Problem Overview

You are building part of a conversation agent. The interviewer gives you an API that returns a conversation's messages one page at a time, and the question builds in parts:

1. Pagination: fetch the entire conversation history by calling the API page after page until it returns no data.
2. Compaction: every message carries a token count. When the history grows past a token limit, drop the oldest messages so the context fits again.

No AI assistance is allowed, and you can pick your language. The algorithms are short. Most of the time goes into reading the provided API and mapping it onto these steps, so read it carefully before writing code.

Part 1: Fetch the Full History

Problem Statement

Given a paginated API, return every message in the conversation in order. Keep calling the API until a page comes back empty.

The API's exact shape is not known. This write-up assumes:

• fetch_page(conversation_id, page) returns one page of messages, and an empty list once there are no more pages.
• Page 0 holds the oldest messages, and each page lists its messages oldest first. Confirm the order: if the API returns newest first, reverse the result before Part 2, which drops from the oldest end.
• A cursor or next-token API changes only how the next request is built, not the loop.

from dataclasses import dataclass
from typing import Callable


@dataclass(frozen=True)
class Message:
    id: str
    role: str  # "user" or "agent"
    content: str
    token_count: int


FetchPage = Callable[[str, int], list[Message]]

Solution

A while loop is all this part needs. Request page 0, append its messages, move to the next page, and stop on the first empty response.

def fetch_conversation_history(
    conversation_id: str,
    fetch_page: FetchPage,
) -> list[Message]:
    history: list[Message] = []
    page = 0

    while True:
        messages = fetch_page(conversation_id, page)
        if not messages:
            return history

        history.extend(messages)
        page += 1

Example

PAGES = {
    0: [
        Message("m1", "user", "Hi, my order has not arrived.", 9),
        Message("m2", "agent", "Sorry to hear that. What is the order number?", 12),
    ],
    1: [
        Message("m3", "user", "It is 4471.", 5),
    ],
}


def fake_fetch_page(conversation_id: str, page: int) -> list[Message]:
    return PAGES.get(page, [])


history = fetch_conversation_history("c1", fake_fetch_page)
[message.id for message in history]
# ['m1', 'm2', 'm3']

Complexity

Let M be the number of messages and P the number of pages.

• Time: O(M + P), plus one API call per page and one final call that returns nothing
• Space: O(M) for the returned history

Part 2: Compact by Token Count

Problem Statement

Each message fetched in Part 1 has a token_count. When the total number of tokens exceeds a limit, compact the history by dropping the oldest messages.

The concrete thresholds given are: once the history exceeds 20,000 tokens, compact it down to fewer than 10,000. Shorter descriptions only say to drop the earliest messages once the history is over the limit. This write-up uses two thresholds; a single limit is the case where both are equal.

For the reference implementation, assume:

• Messages are dropped strictly from the oldest end.
• The newest message is always kept, even if it alone exceeds the target. Confirm this edge case with the interviewer.

Solution

Keep a running token total and drop messages from the front of a queue until the total is at or below the target.

from collections import deque


def compact(
    messages: list[Message],
    max_tokens: int,
    target_tokens: int,
) -> list[Message]:
    if target_tokens > max_tokens:
        raise ValueError("target_tokens must not exceed max_tokens")

    window = deque(messages)
    total = sum(message.token_count for message in window)

    if total <= max_tokens:
        return list(window)

    while len(window) > 1 and total > target_tokens:
        total -= window.popleft().token_count

    return list(window)

Example

history = [
    Message("m1", "user", "...", 4000),
    Message("m2", "agent", "...", 6000),
    Message("m3", "user", "...", 5000),
    Message("m4", "agent", "...", 3000),
    Message("m5", "user", "...", 4000),
]

# 22,000 tokens exceeds 20,000, so drop until the total is at most 10,000.
[message.id for message in compact(history, 20_000, 10_000)]
# ['m4', 'm5']

# With a single 20,000-token limit, dropping m1 is enough.
[message.id for message in compact(history, 20_000, 20_000)]
# ['m2', 'm3', 'm4', 'm5']

Keeping the Window Up to Date

A conversation keeps growing, so recomputing the sum over the full history after every message wastes work. Maintain the window incrementally instead: add each new message on the right and evict from the left only when the limit is crossed. This is a variant of the LeetCode 3 sliding window: extend on the right, shrink on the left while over budget, except that here the window shrinks down to the lower target rather than just back under the limit.

class ConversationWindow:
    def __init__(self, max_tokens: int, target_tokens: int) -> None:
        if target_tokens > max_tokens:
            raise ValueError("target_tokens must not exceed max_tokens")

        self.max_tokens = max_tokens
        self.target_tokens = target_tokens
        self.messages: deque[Message] = deque()
        self.total_tokens = 0

    def add(self, message: Message) -> None:
        self.messages.append(message)
        self.total_tokens += message.token_count

        if self.total_tokens <= self.max_tokens:
            return

        while len(self.messages) > 1 and self.total_tokens > self.target_tokens:
            self.total_tokens -= self.messages.popleft().token_count

    def context(self) -> list[Message]:
        return list(self.messages)

The gap between the two thresholds matters. With a single limit, a full window evicts something on almost every new message. Compacting down to a lower target frees a large block at once, so compaction runs rarely.

The result depends on when compaction runs. The window equals calling compact on the current window after each new message, which can keep more than one call to compact on the full history at the end. For example, with limits of 20 and 10, messages of 15, 6, 5, and 5 tokens leave [6, 5, 5] in the window but [5, 5] from a single compact.

window = ConversationWindow(max_tokens=20_000, target_tokens=10_000)
for message in history:
    window.add(message)

[message.id for message in window.context()]
# ['m4', 'm5']

Complexity

• compact: O(M) time and O(M) space
• ConversationWindow.add: amortized O(1) time, because each message is appended once and evicted at most once
• Space: O(W) for the W messages currently in the window