#!/usr/bin/env python3
"""Original single-process coalescing lab. Python 3.11+ standard library.
Owned threads and synthetic in-memory inputs only; no network, files or model.
This group shares in-flight work; it does not retain completed results as a cache.
"""
import argparse
import copy
import json
import platform
import threading
from concurrent.futures import Future, TimeoutError

CASES = ("naive", "singleflight", "leader-error", "waiter-timeout", "source-changed", "separate-keys")
NORTH = ("north", "guide-7", "renderer-1")
SOUTH = ("south", "guide-7", "renderer-1")


class Inputs:
    def __init__(self):
        self.condition = threading.Condition()
        self.source = {"north": {"revision": 1, "body": "Inspect the north inlet."}, "south": {"revision": 1, "body": "Inspect the south outlet."}}
        self.captures = []
        self.release = threading.Event()

    def build(self, key, fail=False):
        with self.condition:
            value = {"workspace": key[0], **copy.deepcopy(self.source[key[0]])}
            self.captures.append({"key": list(key), "input": copy.deepcopy(value)})
            self.condition.notify_all()
        if not self.release.wait(5):
            raise RuntimeError("fixture_release_not_received")
        if fail:
            raise RuntimeError("injected_builder_failure")
        return value

    def wait_for_builds(self, count):
        with self.condition:
            if not self.condition.wait_for(lambda: len(self.captures) >= count, 5):
                raise RuntimeError("expected_builds_did_not_start")


class FlightGroup:
    """Authored in-flight registry, not the Go package or a production cache.
    Pending registration and same-key join share one lock. Cleanup occurs before
    result delivery, including failure. Every owned thread is joined on close.
    """
    def __init__(self, coalesce=True):
        self.lock = threading.Lock()
        self.pending = {}
        self.threads = []
        self.events = []
        self.coalesce = coalesce

    def join(self, key, build):
        with self.lock:
            if self.coalesce and key in self.pending:
                self.events.append({"event": "joined", "key": list(key)})
                return self.pending[key], True
            future = Future()
            if self.coalesce:
                self.pending[key] = future
            self.events.append({"event": "registered", "key": list(key)})
            thread = threading.Thread(target=self._run, args=(key, future, build))
            self.threads.append(thread)
            thread.start()
            return future, False

    def _run(self, key, future, build):
        if not future.set_running_or_notify_cancel():
            with self.lock:
                if self.pending.get(key) is future:
                    del self.pending[key]
                self.events.append({"event": "cancelled-before-start", "key": list(key)})
            return
        try:
            value, error = build(), None
        except Exception as caught:
            value, error = None, caught
        with self.lock:
            if self.pending.get(key) is future:
                del self.pending[key]
            self.events.append({"event": "completed-error" if error else "completed-value", "key": list(key)})
        if error:
            future.set_exception(error)
        else:
            future.set_result(value)

    def pending_keys(self):
        with self.lock:
            return [list(key) for key in self.pending]

    def close(self):
        for thread in self.threads:
            thread.join(timeout=6)
        if any(thread.is_alive() for thread in self.threads):
            raise RuntimeError("owned_builder_thread_still_running")


def observe(future, timeout=3):
    try:
        return {"state": "value", "value": copy.deepcopy(future.result(timeout)), "error": None}
    except TimeoutError:
        return {"state": "wait-timeout", "value": None, "error": "TimeoutError"}
    except RuntimeError as error:
        return {"state": "builder-error", "value": None, "error": str(error)}


def run_case(name, callers):
    inputs = Inputs()
    group = FlightGroup(coalesce=name != "naive")
    try:
        keys = [NORTH, SOUTH] if name == "separate-keys" else [NORTH] * callers
        futures, shared = [], []
        for key in keys:
            future, joined = group.join(key, lambda key=key: inputs.build(key, fail=name == "leader-error"))
            futures.append(future)
            shared.append(joined)
        expected_builds = len(keys) if name in ("naive", "separate-keys") else 1
        inputs.wait_for_builds(expected_builds)
        before = {"builds_started": len(inputs.captures), "pending_keys": group.pending_keys(), "future_done": [f.done() for f in futures]}
        timeout_observation = None
        if name == "waiter-timeout":
            timeout_observation = observe(futures[1], timeout=0.02)
            if timeout_observation["state"] != "wait-timeout":
                raise RuntimeError("waiter_did_not_time_out_before_release")
            before["builder_still_running_after_waiter_timeout"] = futures[0].running() and not futures[0].done()
        if name == "source-changed":
            with inputs.condition:
                inputs.source["north"] = {"revision": 2, "body": "Inspect the revised north scope."}
        inputs.release.set()
        completed = [observe(future) for future in futures]
        # Preserve the timed-out caller's observation. Later completion is only
        # a separate fixture observer result, not retroactive caller success.
        outcomes = copy.deepcopy(completed)
        if timeout_observation:
            outcomes[1] = timeout_observation
        after_wave = {"builds_started": len(inputs.captures), "pending_keys": group.pending_keys()}
        later = None
        if name in ("singleflight", "leader-error"):
            future, joined = group.join(NORTH, lambda: inputs.build(NORTH))
            later = {"shared": joined, "result": observe(future), "builds_started": len(inputs.captures)}
        return {"case": name, "requested_keys": [list(key) for key in keys], "joined_existing": shared, "before_release": before, "caller_outcomes": outcomes, "fixture_completion_observer": completed, "after_first_wave": after_wave, "later_call": later, "source_after": copy.deepcopy(inputs.source), "captured_builds": copy.deepcopy(inputs.captures), "events": copy.deepcopy(group.events), "completed_result_cache": False}
    finally:
        inputs.release.set()
        group.close()


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--case", choices=CASES)
    parser.add_argument("--callers", type=int, choices=(2, 3), default=2, help="Same-key callers; separate-keys always uses two workspaces")
    args = parser.parse_args()
    cases = [args.case] if args.case else CASES
    reports = []
    for name in cases:
        result = run_case(name, args.callers)
        # This is recorded only after run_case's finally block joined its threads.
        result["owned_threads_joined"] = True
        reports.append(result)
    print(json.dumps({"schema": "lockedin.cache-coalescing-lab/v1", "python": platform.python_version(), "callers": args.callers, "scope": "One process; synthetic inputs and owned builder threads; no authorization or distributed guarantee", "cases": reports}, indent=2))


if __name__ == "__main__":
    main()