"""Real bound PTY workload; input state never waits for terminal output."""

import argparse
import hashlib
import json
import os
from pathlib import Path
import queue
import subprocess
import sys
import threading
import time
import uuid


def now_ms():
    return time.time_ns() // 1_000_000


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--spt", required=True)
    parser.add_argument("--home", required=True)
    parser.add_argument("--id", required=True)
    parser.add_argument("--session-id", required=True)
    parser.add_argument("--period-ms", type=int, default=100)
    args = parser.parse_args()
    if sys.platform != "win32":
        parser.error("the refresh-freeze producer requires Windows ConPTY")
    if args.period_ms <= 0:
        parser.error("--period-ms must be positive")
    home = Path(args.home)
    ambient_home = os.environ.get("SPT_HOME", "")
    if not home.is_absolute() or not ambient_home or not Path(ambient_home).is_absolute():
        parser.error("--home and explicit SPT_HOME must be absolute")
    home = home.resolve(strict=True)
    if not home.is_dir() or Path(ambient_home).resolve(strict=True) != home:
        parser.error("--home must be the existing explicit SPT_HOME directory")
    local = os.environ.get("LOCALAPPDATA")
    if not local:
        parser.error("LOCALAPPDATA is required to exclude the resident home")
    resident = (Path(local) / "spt-core").resolve()
    if home == resident or resident in home.parents:
        parser.error("refusing the resident SPT home")
    spt = Path(args.spt)
    if not spt.is_absolute() or not spt.is_file():
        parser.error("--spt must name an absolute existing executable")
    spt = spt.resolve(strict=True)

    # Binding uses only explicit identity arguments, never an inherited perch.
    env = {
        key: value for key, value in os.environ.items()
        if not key.upper().startswith(("SPT_", "OWL_", "CLAUDE_"))
    }
    env["SPT_HOME"] = str(home)
    generation = uuid.uuid4().hex[:8]
    bind = subprocess.run(
        [str(spt), "api", "--adapter", "evidencerig", "bind", args.id,
         "--set-session-id", args.session_id, "--type", "live_agent"],
        env=env, cwd=home, capture_output=True, text=True,
        encoding="utf-8", errors="replace", timeout=20,
    )
    (home / "generator-bind.json").write_text(json.dumps({
        "exit": bind.returncode, "stdout": bind.stdout, "stderr": bind.stderr,
        "pid": os.getpid(), "generation": generation,
        "session_id": args.session_id, "home": str(home),
    }, indent=2) + "\n", encoding="utf-8")
    if bind.returncode:
        return bind.returncode

    # #314's sole experimental variable: the real root client stays identical.
    # The outer native job contains this bounded, non-interacting descendant.
    control = os.environ.get("LEAF314_CONTROL", "leaf")
    if control not in ("leaf", "descendant"):
        raise ValueError("unknown leaf experiment arm")
    if control == "descendant":
        descendant = subprocess.Popen(
            [sys.executable, "-c", "import time; time.sleep(180)"],
            stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL,
            stderr=subprocess.DEVNULL, creationflags=subprocess.CREATE_NO_WINDOW,
        )
        (home / "control-descendant.json").write_text(
            json.dumps({"pid": descendant.pid, "parent_pid": os.getpid()}) + "\n",
            encoding="utf-8",
        )

    # SimpleQueue.put never waits for the stdout writer. Its FIFO preserves
    # actual generated output; no priority lane, dropped backlog, or refresh arm.
    # The enclosing rig bounds this process's lifetime (and hence backlog).
    output = queue.SimpleQueue()
    output.put(f"GEN_READY {generation} {os.getpid()} {now_ms()}\n")

    def write_output():
        while True:
            sys.stdout.write(output.get())
            sys.stdout.flush()

    def progress():
        counter = 0
        while True:
            output.put(f"PROG {counter} {now_ms()} {generation}\n")
            counter += 1
            time.sleep(args.period_ms / 1000)

    writer = threading.Thread(target=write_output, daemon=True)
    writer.start()
    threading.Thread(target=progress, daemon=True).start()
    ordinal = 0
    chain = ""
    state_path = home / "generator-state.json"
    state_pending = home / "generator-state.json.pending"
    with (home / "generator-input.jsonl").open("a", encoding="utf-8") as ledger:
        for line in sys.stdin:
            tag = line.strip()
            if not tag:
                continue
            if any(character.isspace() for character in tag):
                raise ValueError("input tags must be single nonempty tokens")
            received = now_ms()
            ordinal += 1
            chain = hashlib.sha256((chain + "\0" + tag).encode("utf-8")).hexdigest()
            row = {
                "tag": tag, "ordinal": ordinal, "chain": chain,
                "generated_ms": received, "generation": generation,
            }
            serialized = json.dumps(row) + "\n"
            # Persist the real input and resulting state before enqueueing ACK.
            # Neither file operation acquires any terminal-output lock.
            ledger.write(serialized)
            ledger.flush()
            os.fsync(ledger.fileno())
            with state_pending.open("w", encoding="utf-8") as state:
                state.write(serialized)
                state.flush()
                os.fsync(state.fileno())
            os.replace(state_pending, state_path)
            output.put(
                f"ACK {tag} {ordinal} {received}\n"
                f"STATE {ordinal} {chain[:16]} {received}\n"
            )
    # EOF does not manufacture a successful completion or stop the workload.
    writer.join()
    return 2


if __name__ == "__main__":
    raise SystemExit(main())