import java.util.HashMap;
import java.util.Map;

/** Single-stream, single-process model. No persistence, distributed leases or real encoder control. */
public final class StreamState {
    public enum Status { RECEIVING, PLAYABLE, DISCONNECTED }
    private static final class Job {
        final String spec;
        boolean running, done;
        long fence;
        Job(String spec) { this.spec = spec; }
    }
    private final int capacity;
    private final Map<String, Job> jobs = new HashMap<>();
    private long generation = 0, nextFence = 0;
    private Status status = Status.DISCONNECTED;
    public StreamState(int capacity) {
        if (capacity < 1) throw new IllegalArgumentException("capacity");
        this.capacity = capacity;
    }
    // Only the accepted-connection path may create a new generation.
    public synchronized long connect() {
        generation++;
        status = Status.RECEIVING;
        jobs.clear(); // Invalidates model claims only; production must fence/stop real workers.
        return generation;
    }
    public synchronized boolean event(long gen, Status next) {
        if (gen != generation || gen == 0) return false;
        if (status == next) return true;
        if (status == Status.DISCONNECTED) return false;
        // Becoming PLAYABLE is reserved for a validated current worker result.
        if (next != Status.DISCONNECTED) return false;
        status = next;
        jobs.values().forEach(job -> job.running = false);
        return true;
    }
    public synchronized boolean submit(long gen, String key, String spec) {
        if (gen != generation || status == Status.DISCONNECTED) throw new IllegalStateException("stale input");
        if (key.isBlank() || spec.isBlank()) throw new IllegalArgumentException("empty identity");
        Job old = jobs.get(key);
        if (old != null) {
            if (!old.spec.equals(spec)) throw new IllegalArgumentException("idempotency conflict");
            return false;
        }
        jobs.put(key, new Job(spec));
        return true;
    }
    public synchronized long claim(long gen, String key) {
        Job job = jobs.get(key);
        if (gen != generation || status == Status.DISCONNECTED || job == null || job.done || job.running)
            throw new IllegalStateException("not claimable");
        if (jobs.values().stream().filter(j -> j.running).count() >= capacity)
            throw new IllegalStateException("capacity exhausted");
        job.running = true;
        return job.fence = ++nextFence;
    }
    public synchronized boolean expire(long gen, String key, long fence) {
        Job job = jobs.get(key);
        if (gen != generation || job == null || !job.running || job.fence != fence) return false;
        job.running = false;
        return true;
    }
    public synchronized boolean complete(long gen, String key, long fence, boolean outputValidated) {
        Job job = jobs.get(key);
        if (gen != generation || status == Status.DISCONNECTED || job == null || job.fence != fence || !outputValidated) return false;
        if (job.done) return true;
        if (!job.running) return false;
        job.running = false; job.done = true; status = Status.PLAYABLE;
        return true;
    }
    public synchronized Status status() { return status; }
}