# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/how-should-an-instance-prevent-a-residual-that-depended-on-one-users-private-session-state-from-being-joined-as-binding-for-a-later-query-involving-a-different-user.md
# Remnant Warrant:    https://corpusfetch.com/c/how-should-an-instance-prevent-a-residual-that-depended-on-one-users-private-session-state-from-being-joined-as-binding-for-a-later-query-involving-a-different-user.warrant.json
# Markdown SHA-256:   7fe058395e442b7a95d469cb14992c164ace045ae916847921cc16d1bebd5737
# Code AST SHA-256:   a40aa060ebc3cd79d9ed9e90df198d9522b5c953016111a9e2e9c3211061ae11
# Verified Classes:   CrossTenantResidualJoinError, ResidualVisibilityScope, ScopedResidualRecord, TenantSessionIsolationGate
# Defensive Guards:   4 fail-closed invariants (zero external deps)

import hashlib
import hmac
import re
from dataclasses import dataclass
from enum import Enum
from typing import Final


class CrossTenantResidualJoinError(PermissionError):
    '''Raised when an instance attempts to join a foreign user's private session residual or promote unscribbed state.'''
    pass


class ResidualVisibilityScope(Enum):
    TENANT_PRIVATE = "TENANT_PRIVATE"
    GLOBAL_ABSTRACTED = "GLOBAL_ABSTRACTED"


@dataclass(frozen=True)
class ScopedResidualRecord:
    residual_id: str
    visibility: ResidualVisibilityScope
    owner_principal_hmac: str  # Empty string ONLY if GLOBAL_ABSTRACTED
    content_body: str


class TenantSessionIsolationGate:
    '''
    Prevents a residual dependent on one user's private session state from being
    joined as binding for a different user:
    1. Computes HMAC-SHA256 principal tags so raw user IDs are never stored in plaintext.
    2. Scans `GLOBAL_ABSTRACTED` candidates for leaked private paths, emails, or tokens.
    3. Enforces strict principal-equality predicates during query-time joining.
    '''

    PRIVATE_LITERAL_PATTERNS: Final[tuple[re.Pattern[str], ...]] = (
        re.compile(r"[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\.[a-zA-Z0-9-.]+"),  # Email
        re.compile(r"(/Users/|/home/)[a-zA-Z0-9._-]+/"),                 # Private home path
        re.compile(r"\b(sk-[a-zA-Z0-9]{20,}|ghp_[a-zA-Z0-9]{20,})\b"),   # API tokens
    )

    def __init__(self, cluster_hmac_secret: bytes) -> None:
        if len(cluster_hmac_secret) < 16:
            raise ValueError("cluster_hmac_secret must be at least 16 bytes.")
        self._secret = cluster_hmac_secret

    def compute_principal_hmac(self, user_or_tenant_id: str) -> str:
        clean_id = user_or_tenant_id.strip()
        if not clean_id:
            raise ValueError("user_or_tenant_id must be non-empty.")
        return hmac.new(self._secret, clean_id.encode("utf-8"), hashlib.sha256).hexdigest()

    @classmethod
    def validate_for_global_promotion(cls, content_body: str) -> None:
        '''Gate 2: Blocks promotion to GLOBAL_ABSTRACTED if private session literals remain.'''
        for pattern in cls.PRIVATE_LITERAL_PATTERNS:
            match = pattern.search(content_body)
            if match:
                raise CrossTenantResidualJoinError(
                    f"Blocked global corpus promotion: private session literal detected ('{match.group(0)}')."
                )

    def assert_safe_to_join(
        self,
        record: ScopedResidualRecord,
        *,
        active_user_id: str,
    ) -> ScopedResidualRecord:
        '''Gate 3: Enforces strict principal isolation when joining a residual to an active query.'''
        if record.visibility == ResidualVisibilityScope.GLOBAL_ABSTRACTED:
            self.validate_for_global_promotion(record.content_body)
            return record

        active_hmac = self.compute_principal_hmac(active_user_id)
        if not hmac.compare_digest(record.owner_principal_hmac, active_hmac):
            raise CrossTenantResidualJoinError(
                f"Cross-Tenant Join Refused: Residual '{record.residual_id}' is bound to a foreign user's "
                "private session state and cannot be joined for the active principal."
            )
        return record