"""Session-scoped cooperative cancellation, safe across the ZeroGPU worker boundary.

The SDK owns its worker. This app creates no inference thread or GPU generator.
The cancellation marker lives on the shared filesystem and is unlinked only after
the decorated call has returned. A new request cannot reuse the previous marker.
"""
from dataclasses import dataclass
import hashlib
import logging
from pathlib import Path
import secrets
import threading
import time

log = logging.getLogger("zonos.lifecycle")


class Cancelled(Exception):
    pass


class AlreadyRunning(Exception):
    pass


def check_cancelled(marker):
    if Path(marker).exists():
        raise Cancelled()


@dataclass(frozen=True)
class Job:
    id: str
    owner: str
    marker: Path
    started: float


class Jobs:
    def __init__(self, directory):
        self.directory = Path(directory)
        self.directory.mkdir(parents=True, exist_ok=True)
        self._lock = threading.Lock()
        self._active = {}

    def begin(self, owner):
        if not isinstance(owner, str) or not owner:
            raise ValueError("Missing session identity")
        key = hashlib.sha256(owner.encode()).hexdigest()
        with self._lock:
            if key in self._active:
                raise AlreadyRunning()
            identifier = secrets.token_hex(24)
            job = Job(identifier, key, self.directory / (identifier + ".cancel"), time.monotonic())
            self._active[key] = job
        log.info("request_started request=%s", identifier)
        return job

    def cancel(self, owner):
        key = hashlib.sha256(owner.encode()).hexdigest()
        with self._lock:
            job = self._active.get(key)
            if job is None:
                return False
            job.marker.touch(exist_ok=True)
        log.info("cancellation_requested request=%s", job.id)
        return True

    def finish(self, job):
        with self._lock:
            if self._active.get(job.owner) == job:
                del self._active[job.owner]
            job.marker.unlink(missing_ok=True)
        log.info("request_finished request=%s", job.id)

    def active(self):
        with self._lock:
            return len(self._active)