# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/what-structural-signals-indicate-context-compaction-has-silently-invalidated-a-shared-working-state-across-concurrent-subagents.md
# Remnant Warrant:    https://corpusfetch.com/c/what-structural-signals-indicate-context-compaction-has-silently-invalidated-a-shared-working-state-across-concurrent-subagents.warrant.json
# Markdown SHA-256:   fd405c5c57f417a9b143186c9945acc7978c94c084f04c32adaf484ac46bc478
# Code AST SHA-256:   64007ff6d470563d4f17b568bb6ce8b41559d2c9bcd7f65d4cd33bbb8ac822dd
# Verified Classes:   CompactionConsistencyError, LedgerEntry, SharedLedger
# Defensive Guards:   3 fail-closed invariants (zero external deps)

from __future__ import annotations

from dataclasses import dataclass, field
from typing import Dict


class CompactionConsistencyError(RuntimeError):
    """Raised when a write from a possibly-compacted subagent violates shared-state provenance."""


@dataclass
class LedgerEntry:
    value: str
    revision: int


@dataclass
class SharedLedger:
    entries: Dict[str, LedgerEntry] = field(default_factory=dict)

    def apply_write(self, agent_id: str, key: str, value: str, revision: int,
                    compacted: bool, payload_hash: str, seen_hash: str | None) -> None:
        # Idempotent retry of the identical write is always safe.
        if seen_hash is not None and seen_hash == payload_hash:
            return

        current = self.entries.get(key)

        # Rule 1: a compacted agent may not mutate without a fresh read.
        if compacted:
            raise CompactionConsistencyError(
                f"agent {agent_id} is compacted and attempted write to {key!r} "
                f"without re-validating against authoritative state"
            )

        # Rule 2: revisions must strictly advance on a mutating write.
        if current is not None and revision <= current.revision:
            raise CompactionConsistencyError(
                f"agent {agent_id} wrote {key!r} at revision {revision}, "
                f"but authoritative revision is {current.revision}; writer's model is stale"
            )

        self.entries[key] = LedgerEntry(value=value, revision=revision)


def commit_batch(ledger: SharedLedger, agent_id: str, writes: list[dict],
                 compacted: bool, baseline_breadth: int) -> None:
    # Rule 3: write-set breadth collapse signals operating from a summary.
    if baseline_breadth > 0 and len(writes) < baseline_breadth * 0.5:
        raise CompactionConsistencyError(
            f"agent {agent_id} write-set breadth {len(writes)} is below 50% of "
            f"baseline {baseline_breadth}; audit for silent compaction"
        )
    for w in writes:
        ledger.apply_write(
            agent_id=agent_id,
            key=w["key"],
            value=w["value"],
            revision=w["revision"],
            compacted=compacted,
            payload_hash=w["payload_hash"],
            seen_hash=w.get("seen_hash"),
        )