# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/which-invariants-prevent-tool-call-result-atomicity-loss-during-mid-turn-context-compaction.md
# Remnant Warrant:    https://corpusfetch.com/c/which-invariants-prevent-tool-call-result-atomicity-loss-during-mid-turn-context-compaction.warrant.json
# Markdown SHA-256:   f90ffd81b71bded8dfe2158e2e15297f4add1cdd48f3e1b370a9954af7ab2832
# Code AST SHA-256:   4440191fe289bb2e8ecff24ee9bc2a21dc5c6ca6d6e8c04038c218b8c66e092d
# Verified Classes:   IllegalCompactionCut, Record
# Defensive Guards:   1 fail-closed invariants (zero external deps)

from dataclasses import dataclass
from typing import Literal, Sequence


class IllegalCompactionCut(RuntimeError):
    """Raised when a proposed compaction cut would sever a tool_use/tool_result pair."""


@dataclass(frozen=True)
class Record:
    kind: Literal["assistant", "tool_use", "tool_result", "user"]
    call_id: str | None = None
    tokens: int = 0


def pending_calls_at(records: Sequence[Record], cut: int) -> set[str]:
    """Return the set of tool_use ids before `cut` that have no result before `cut`."""
    used: set[str] = set()
    satisfied: set[str] = set()
    for r in records[:cut]:
        if r.kind == "tool_use" and r.call_id is not None:
            used.add(r.call_id)
        elif r.kind == "tool_result" and r.call_id is not None:
            satisfied.add(r.call_id)
    return used - satisfied


def orphan_results_at(records: Sequence[Record], cut: int) -> set[str]:
    """Return tool_result ids before `cut` whose tool_use was dropped (after `cut`)."""
    used_after: set[str] = {
        r.call_id for r in records[cut:]
        if r.kind == "tool_use" and r.call_id is not None
    }
    return {
        r.call_id for r in records[:cut]
        if r.kind == "tool_result" and r.call_id in used_after
    }


def select_legal_cut(
    records: Sequence[Record],
    target_tokens: int,
    max_lookback_tokens: int,
) -> int:
    """Walk backward from the budget target to the nearest pairing-legal cut.

    Raises IllegalCompactionCut if no legal cut exists within the lookback window,
    signaling the caller to defer compaction to the next turn boundary.
    """
    total = sum(r.tokens for r in records)
    if target_tokens >= total:
        return len(records)

    target_index = 0
    acc = 0
    for i, r in enumerate(records):
        acc += r.tokens
        if acc >= target_tokens:
            target_index = i
            break

    floor_index = target_index
    acc = 0
    for j in range(target_index, -1, -1):
        acc += records[j].tokens
        if acc > max_lookback_tokens:
            break
        floor_index = j

    for cut in range(target_index, floor_index - 1, -1):
        if pending_calls_at(records, cut):
            continue
        if orphan_results_at(records, cut):
            continue
        return cut

    raise IllegalCompactionCut(
        f"no pairing-legal cut within {max_lookback_tokens} tokens of target "
        f"{target_tokens}; defer compaction to turn boundary"
    )