# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/where-must-an-instance-refuse-a-runtime-prompt-rewrite-the-append-only-prefix-invariant-and-the-silent-orphaned-tool-call-boundary.md
# Remnant Warrant:    https://corpusfetch.com/c/where-must-an-instance-refuse-a-runtime-prompt-rewrite-the-append-only-prefix-invariant-and-the-silent-orphaned-tool-call-boundary.warrant.json
# Markdown SHA-256:   9ada55019221e70be3424134643feb9594feea4af68329f4fc72c4c5c9bf7df9
# Code AST SHA-256:   ae0ce78ba101e54bb11af4bac8d84e490a26f6276db38b9f3d0d1e2f9db28da1
# Verified Classes:   NonMonotonicRewriteError, OrphanedToolPairError, Message, ContextWindow
# Defensive Guards:   3 fail-closed invariants (zero external deps)

"""Runtime guard for self-editing context windows.

Enforces the append-only prefix invariant and tool-call/result pairing
integrity before any mid-session prompt rewrite is applied.
"""

from dataclasses import dataclass, field
from typing import Literal


class NonMonotonicRewriteError(RuntimeError):
    """Raised when a rewrite would mutate the already-consumed prefix
    or orphan a tool_use/tool_result pair."""


class OrphanedToolPairError(NonMonotonicRewriteError):
    """Raised when a rewrite drops a tool_use or tool_result without its partner."""


@dataclass
class Message:
    role: Literal["system", "user", "assistant", "tool"]
    content: str
    tool_call_id: str | None = None
    tool_name: str | None = None


@dataclass
class ContextWindow:
    messages: list[Message] = field(default_factory=list)
    generation_boundary: int = 0  # index of last token the model produced

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


def _pair_ids(messages: list[Message]) -> tuple[set[str], set[str]]:
    calls = {m.tool_call_id for m in messages
             if m.role == "assistant" and m.tool_call_id}
    results = {m.tool_call_id for m in messages
               if m.role == "tool" and m.tool_call_id}
    return calls, results


def validate_rewrite(current: ContextWindow, proposed: list[Message]) -> None:
    """Raise NonMonotonicRewriteError if the rewrite violates the invariant."""
    # 1. Prefix identity: consumed prefix must be byte-identical.
    prefix_before = current.consumed_prefix()
    prefix_after = proposed[: current.generation_boundary]
    if prefix_before != prefix_after:
        raise NonMonotonicRewriteError(
            f"rewrite mutates consumed prefix "
            f"[0:{current.generation_boundary}]; refusing non-monotonic edit"
        )

    # 2. Pairing integrity across the whole proposed window.
    calls, results = _pair_ids(proposed)
    if calls != results:
        orphaned_calls = calls - results
        orphaned_results = results - calls
        raise OrphanedToolPairError(
            f"unpaired tool ids: calls={orphaned_calls or '{}'} "
            f"results={orphaned_results or '{}'}"
        )


def apply_rewrite(current: ContextWindow, proposed: list[Message]) -> ContextWindow:
    validate_rewrite(current, proposed)
    return ContextWindow(messages=proposed,
                         generation_boundary=current.generation_boundary)


def self_edit(current: ContextWindow, new_system_prompt: str) -> ContextWindow:
    """Model-initiated rewrite. Treated as untrusted; must pass validation."""
    if not current.messages or current.messages[0].role != "system":
        raise NonMonotonicRewriteError("context has no leading system message")
    proposed = [Message(role="system", content=new_system_prompt)] + current.messages[1:]
    # If the system message was already consumed, this will raise.
    return apply_rewrite(current, proposed)