# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/which-invariants-prevent-double-counting-of-shared-working-set-tokens-across-nested-subagent-context-budgets.md
# Remnant Warrant:    https://corpusfetch.com/c/which-invariants-prevent-double-counting-of-shared-working-set-tokens-across-nested-subagent-context-budgets.warrant.json
# Markdown SHA-256:   ffc3d99d26e87063b202089675f3e1425915ade18d4fd11907eab9ad84197561
# Code AST SHA-256:   0607413d9be6732549fb287e9f92ec2a57df936aed6c23fafeb07f2781ed6ad7
# Verified Classes:   SharedWorkingSetBudgetError, Ledger
# Defensive Guards:   1 fail-closed invariants (zero external deps)

from dataclasses import dataclass, field
from typing import Dict


class SharedWorkingSetBudgetError(RuntimeError):
    """Raised when nested subagent materialization violates budget conservation."""
    pass


@dataclass
class Ledger:
    allocation: int
    materialized: int = 0
    # hash -> number of windows that have hydrated it
    refs: Dict[str, int] = field(default_factory=dict)

    def materialize(self, content_hash: str, cost: int, child_id: str) -> None:
        # Re-hydration by a sibling is a fresh debit, never free.
        self.materialized += cost
        self.refs[content_hash] = self.refs.get(content_hash, 0) + 1
        if self.materialized > self.allocation:
            raise SharedWorkingSetBudgetError(
                f"child {child_id} hydration of {content_hash} (cost={cost}) "
                f"pushes ancestor materialized={self.materialized} "
                f"over allocation={self.allocation}; "
                f"refs={self.refs.get(content_hash)}"
            )


def hydrate_shared(ledger: Ledger, content_hash: str, cost: int, child_id: str) -> None:
    # Storage dedup is irrelevant here; budget dedup is per-materialization.
    ledger.materialize(content_hash, cost, child_id)