"""Durable foreman-owned fleet observations, acknowledgements, and stop bindings.

Every document/row is schema 1. The canonical selected state owns an adjacent
`.supervision.json`; exact native foreman identities discover it through one
hashed binding under the default state directory. Reads never migrate or write.
Unknown/corrupt files are preserved and mutations refuse them. Transactions use
state_lock/save_state; observations and acknowledgements commit together.

Events describe observations, never task acceptance. Acknowledgement requires
an explicit foreman outcome and evidence receipt; reading/draining cannot ack.
Resolving enrollment ends observation only and grants no task completion.
"""

import hashlib
import json
from datetime import datetime, timedelta
from pathlib import Path

from . import runnable
from .errors import StateError, UsageError
from .state import default_state_path, save_state, state_lock

SCHEMA_VERSION = 1
DEFAULT_RECHECK_SECONDS = 30


def timestamp(value):
    try:
        parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
    except (AttributeError, TypeError, ValueError):
        raise UsageError("Pass an ISO timestamp with an explicit timezone.", {}) from None
    if parsed.tzinfo is None:
        raise UsageError("Pass an ISO timestamp with an explicit timezone.", {})
    return parsed


def text(value, label):
    if not isinstance(value, str) or not value.strip() or any(ord(char) < 32 for char in value):
        raise UsageError("{} must be nonempty single-line text; preserve the original record.".format(label), {})
    return value


def canonical(path):
    return Path(path).expanduser().resolve()


def store_path(state_path):
    return Path(str(canonical(state_path)) + ".supervision.json")


def digest(value):
    return hashlib.sha256(json.dumps(value, sort_keys=True, separators=(",", ":")).encode()).hexdigest()


def report_bound_fingerprint(fingerprint, report):
    """A bound round's dispatch fingerprint: the legacy digest wrapped with the exact report path.

    `apply` stores this form for every new dispatch of a bound round, and
    owner recovery must reproduce it from the same report path; any other path
    or a second wrapping yields a different value.
    """
    return digest({"dispatch": fingerprint, "report": report})


def read_json(path):
    try:
        value = json.loads(Path(path).read_text(encoding="utf-8"))
        if value is None:
            raise StateError("Supervision record {} contains null instead of an owner record. Preserve it and restore a supported backup.".format(path), {})
        return value
    except FileNotFoundError:
        if Path(path).is_symlink():
            raise StateError("Supervision record {} is a dangling link. Restore its original target; do not replace saved history with fresh state.".format(path), {}) from None
        return None
    except (OSError, UnicodeDecodeError, json.JSONDecodeError) as exc:
        raise StateError("Cannot read supervision record {}: {}. Restore its original readable bytes; do not overwrite it.".format(path, exc), {}) from None


def receipt(path):
    source = canonical(text(path, "evidence path"))
    if not Path(path).expanduser().is_absolute():
        raise UsageError("Evidence paths must be absolute; supply the saved foreman outcome or task-ledger path.", {})
    try:
        body = source.read_bytes()
    except OSError as exc:
        raise StateError("Cannot read evidence {}: {}. Save the evidence before recording the outcome.".format(source, exc), {}) from None
    return {"schema_version": 1, "path": str(source), "sha256": hashlib.sha256(body).hexdigest(), "size": len(body)}


def evidence(paths):
    if not isinstance(paths, list) or not paths:
        raise UsageError("Provide at least one absolute saved evidence path for this foreman outcome.", {})
    return [receipt(path) for path in paths]


def empty(state_path):
    return {"schema_version": 1, "state_path": str(canonical(state_path)), "binding": None,
            "members": [], "events": [], "acknowledgements": [], "holds": [], "watchers": []}


def load(state_path):
    path = store_path(state_path)
    data = read_json(path)
    if data is None:
        return empty(state_path)
    try:
        if (not isinstance(data, dict) or data.get("schema_version") != 1
                or data.get("state_path") != str(canonical(state_path))):
            raise ValueError("document identity or schema differs")
        for key in ("members", "events", "acknowledgements", "holds", "watchers"):
            if not isinstance(data.get(key), list) or any(not isinstance(row, dict) or row.get("schema_version") != 1 for row in data[key]):
                raise ValueError("invalid {} rows".format(key))
        if data.get("binding") is not None and (not isinstance(data["binding"], dict) or data["binding"].get("schema_version") != 1):
            raise ValueError("invalid binding")
        if data["binding"] is None and any(data[key] for key in ("members", "events", "acknowledgements", "holds", "watchers")):
            raise ValueError("unbound owner contains active history")
        if data["binding"] is not None:
            saved_binding = data["binding"]
            who = saved_binding["identity"]
            if (not isinstance(who, dict) or set(who) != {"kind", "value", "cwd", "herdr_env", "pane_id"}
                    or identity(who["value"], who["cwd"], who["herdr_env"], kind=who["kind"], pane_id=who["pane_id"]) != who
                    or saved_binding["state_path"] != str(canonical(state_path))
                    or type(saved_binding["generation"]) is not int or saved_binding["generation"] < 1):
                raise ValueError("invalid bound foreman identity")
            timestamp(saved_binding["at"])
        ids = set()
        for row in data["members"]:
            validate_member(row["assignment"])
            if row["id"] != row["assignment"]["id"] or row["id"] in ids or type(row["active"]) is not bool:
                raise ValueError("invalid enrollment identity")
            ids.add(row["id"])
            timestamp(row["at"])
            if not isinstance(row["observed"], dict) or not isinstance(row["refinements"], list):
                raise ValueError("invalid observation cursor")
            for refinement in row["refinements"]:
                if not isinstance(refinement, dict) or refinement.get("schema_version") != 1:
                    raise ValueError("invalid identity refinement")
                timestamp(refinement["at"])
                validate_member({**row["assignment"], **{key: refinement[key] for key in ("pane_id", "native_session")}})
            if row["active"] != (row["resolution"] is None):
                raise ValueError("enrollment resolution differs from active flag")
            if row["resolution"] is not None:
                if row["resolution"].get("schema_version") != 1:
                    raise ValueError("invalid resolution schema")
                timestamp(row["resolution"]["at"])
                text(row["resolution"]["outcome"], "resolution outcome")
                validate_receipts(row["resolution"]["evidence"])
        for seq, row in enumerate(data["events"], 1):
            if type(row["seq"]) is not int or row["seq"] != seq or row["id"] != "event-{}".format(seq) or not isinstance(row["data"], dict):
                raise ValueError("invalid event sequence")
            timestamp(row["at"])
            text(row["kind"], "event kind")
            if row["member"] is not None and row["member"] not in ids:
                raise ValueError("event references unknown member")
        acknowledged = set()
        for row in data["acknowledgements"]:
            if row["event"] in acknowledged or row["event"] not in {event["id"] for event in data["events"]}:
                raise ValueError("invalid acknowledgement")
            acknowledged.add(row["event"])
            text(row["outcome"], "acknowledgement outcome")
            if (not isinstance(row["input_digest"], str) or len(row["input_digest"]) != 64
                    or any(char not in "0123456789abcdef" for char in row["input_digest"])):
                raise ValueError("invalid acknowledgement input identity")
            timestamp(row["at"])
            if type(row["pending"]) is not bool or row["pending"] != (row["recheck_at"] is not None):
                raise ValueError("invalid pending recheck")
            if row["recheck_at"] is not None and timestamp(row["recheck_at"]) < timestamp(row["at"]):
                raise ValueError("recheck must follow acknowledgement")
            validate_receipts(row["evidence"])
        for row in data["holds"]:
            if row["kind"] not in ("waiting_for_user", "handoff") or type(row["through"]) is not int or not 0 <= row["through"] <= len(data["events"]):
                raise ValueError("invalid hold")
            validate_receipts(row["evidence"])
            text(row["resume_condition"], "resume condition")
            text(row["id"], "hold id")
            text(row["members"], "hold members digest")
            timestamp(row["at"])
            if row["resumed_at"] is not None:
                timestamp(row["resumed_at"])
            if not isinstance(row["dispositions"], list):
                raise ValueError("invalid hold dispositions")
            for disposition in row["dispositions"]:
                if not isinstance(disposition, dict) or disposition.get("schema_version") != 1 or disposition["member"] not in ids:
                    raise ValueError("invalid hold disposition")
                text(disposition["outcome"], "disposition outcome")
                validate_receipts(disposition["evidence"])
        for row in data["watchers"]:
            if row["status"] not in ("running", "stopped") or not isinstance(row["process"], dict):
                raise ValueError("invalid watcher")
            timestamp(row["heartbeat"])
            timestamp(row["deadline"])
            if type(row["process"]["pid"]) is not int or row["process"]["pid"] <= 0:
                raise ValueError("invalid process")
            text(row["process"]["identity"], "process identity")
        return data
    except (AttributeError, KeyError, TypeError, ValueError, UsageError) as exc:
        raise StateError("Unsupported or corrupt supervision record {} ({}). Preserve it and restore a supported owner-written backup.".format(path, exc), {}) from None


def validate_receipts(rows):
    if not isinstance(rows, list) or not rows:
        raise ValueError("missing evidence")
    for row in rows:
        if (not isinstance(row, dict) or row.get("schema_version") != 1 or not isinstance(row.get("path"), str)
                or not Path(row["path"]).is_absolute() or not isinstance(row.get("sha256"), str)
                or len(row["sha256"]) != 64 or any(char not in "0123456789abcdef" for char in row["sha256"]) or type(row.get("size")) is not int or row["size"] < 0):
            raise ValueError("invalid evidence receipt")


def transaction(state_path, mutate):
    path = store_path(state_path)
    with state_lock(path):
        data = load(state_path)
        if data["binding"] is None:
            raise StateError("Supervision owner state is missing or unbound. Run `{}` for first use; restore a previously bound owner document before resuming its obligations.".format(
                runnable.command("supervision-bind")), {})
        result = mutate(data)
        save_state(path, data)
        return result


def validate_member(record):
    if not isinstance(record, dict) or set(record) != {"id", "agent", "task", "report", "pane_id", "native_session"}:
        raise UsageError("Enrollment requires id, agent, task, absolute report, pane_id and native_session (nullable identity fields).", {})
    for key in ("id", "agent", "task", "report"):
        text(record[key], key)
    if not Path(record["report"]).is_absolute():
        raise UsageError("Enrollment report must be the absolute path in the worker brief.", {})
    if record["pane_id"] is not None:
        text(record["pane_id"], "pane_id")
    native = record["native_session"]
    if native is not None:
        if not isinstance(native, dict) or not {"source", "agent", "kind", "value"}.issubset(native) or native["kind"] not in ("id", "path"):
            raise UsageError("native_session must preserve a native identity object (source,agent,kind,value; optional pane_id) or null.", {})
        for key in ("source", "agent", "value"):
            text(native[key], "native " + key)


def enroll(state_path, record, at):
    validate_member(record)
    timestamp(at)
    def mutate(data):
        if data["binding"] is None:
            raise UsageError("Bind this foreman's native session with `{}` before enrolling worker assignments.".format(
                runnable.command("supervision-bind")), {})
        previous = next((row for row in data["members"] if row["id"] == record["id"]), None)
        if previous:
            if previous["assignment"] != record:
                raise UsageError("Enrollment ID already names different evidence; reconcile the original assignment before assigning a new ID.", {})
            return previous
        if any(row["active"] and row["assignment"]["agent"] == record["agent"] for row in data["members"]):
            raise UsageError("This worker still has an active enrollment. Reconcile its outcome with `{}` before replacing it.".format(
                    runnable.command("supervision-resolve")), {})
        row = {"schema_version": 1, "id": record["id"], "at": at, "assignment": record,
               "active": True, "observed": {}, "resolution": None, "refinements": []}
        data["members"].append(row)
        return row
    return transaction(state_path, mutate)


def append_event(data, at, member, kind, detail):
    seq = len(data["events"]) + 1
    row = {"schema_version": 1, "id": "event-{}".format(seq), "seq": seq, "at": at,
           "member": member, "kind": kind, "data": detail}
    data["events"].append(row)
    return row


def pending(data):
    done = {row["event"] for row in data["acknowledgements"]}
    return [row for row in data["events"] if row["id"] not in done]


def drain(state_path, through=None):
    data = load(state_path)
    high = len(data["events"]) if through is None else through
    if type(high) is not int or not 0 <= high <= len(data["events"]):
        raise UsageError("Drain through must identify an existing event sequence, or omit it for the current snapshot.", {})
    return {"schema_version": 1, "state_path": data["state_path"], "through": high,
            "events": [row for row in pending(data) if row["seq"] <= high]}


def acknowledge(state_path, record, at):
    timestamp(at)
    if not isinstance(record, dict) or set(record) != {"through", "outcomes"} or not isinstance(record["outcomes"], list) or not record["outcomes"]:
        raise UsageError("Acknowledgement requires the drained through sequence and nonempty outcomes [{event,outcome,evidence}].", {})
    prepared = []
    for row in record["outcomes"]:
        if not isinstance(row, dict) or not {"event", "outcome", "evidence"}.issubset(row) or set(row) - {"event", "outcome", "evidence", "pending", "recheck_at"}:
            raise UsageError("Each acknowledgement requires event, outcome, and evidence paths.", {})
        prepared.append({"schema_version": 1, "at": at, "event": text(row["event"], "event"),
                         "input_digest": digest(row), "outcome": text(row["outcome"], "outcome"),
                         "evidence": row["evidence"], "recheck_at": row.get("recheck_at"), "pending": row.get("pending", row.get("recheck_at") is not None)})
        if type(prepared[-1]["pending"]) is not bool or row.get("recheck_at") is not None and not prepared[-1]["pending"]:
            raise UsageError("pending must be a boolean; a scheduled recheck cannot declare pending false.", {})
    if len({row["event"] for row in prepared}) != len(prepared):
        raise UsageError("Acknowledge each event only once in a request.", {})
    def mutate(data):
        high = record["through"]
        if type(high) is not int or not 0 <= high <= len(data["events"]):
            raise UsageError("Use the through sequence returned by `{}`.".format(runnable.command("supervision-drain")), {})
        for row in prepared:
            event = next((item for item in data["events"] if item["id"] == row["event"]), None)
            if event is None or event["seq"] > high:
                raise UsageError("Acknowledgement includes an event outside its drained snapshot; drain again before handling later events.", {})
            prior = next((item for item in data["acknowledgements"] if item["event"] == row["event"]), None)
            if prior:
                if prior["input_digest"] != row["input_digest"]:
                    raise UsageError("Event already has a different handled outcome; preserve its acknowledgement.", {})
            else:
                if row["recheck_at"] is not None and timestamp(row["recheck_at"]) < timestamp(at):
                    raise UsageError("A pending observation recheck_at must not precede its acknowledgement time.", {})
                row["evidence"] = evidence(row["evidence"])
                if row["pending"] and row["recheck_at"] is None:
                    row["recheck_at"] = (timestamp(at) + timedelta(seconds=DEFAULT_RECHECK_SECONDS)).isoformat()
                data["acknowledgements"].append(row)
        return {"schema_version": 1, "acknowledged": [row["event"] for row in prepared], "remaining": pending(data)}
    return transaction(state_path, mutate)


def resolve(state_path, record, at):
    if not isinstance(record, dict) or set(record) != {"id", "outcome", "evidence"}:
        raise UsageError("Resolution requires enrollment id, foreman outcome, and saved evidence paths.", {})
    result = {"schema_version": 1, "at": at, "outcome": text(record["outcome"], "outcome"), "evidence": evidence(record["evidence"])}
    timestamp(at)
    def mutate(data):
        member = next((row for row in data["members"] if row["id"] == record["id"]), None)
        if member is None:
            raise UsageError("Unknown enrollment ID; use `{}` to inspect active assignments.".format(
                runnable.command("supervision-status")), {})
        if any(row["member"] == member["id"] for row in pending(data)):
            raise UsageError("Handle and acknowledge this assignment's pending observations before resolving its enrollment.", {})
        if not member["active"]:
            if any(member["resolution"][key] != result[key] for key in ("outcome", "evidence")):
                raise UsageError("Enrollment already has a different resolution; preserve it.", {})
        else:
            member.update(active=False, resolution=result)
        return member
    return transaction(state_path, mutate)


def active_digest(data):
    return digest([expected_assignment(row) for row in data["members"] if row["active"]])


def refine(state_path, member_id, pane_id, native_session, at):
    """Fill missing post-send identity without rewriting the enrolled assignment."""
    timestamp(at)
    def mutate(data):
        row = next((item for item in data["members"] if item["id"] == member_id), None)
        if row is None or not row["active"]:
            raise UsageError("Refine only an active enrollment; reconcile its saved identity before replacing it.", {})
        current = expected_assignment(row)
        updates = {"pane_id": pane_id, "native_session": native_session}
        validate_member({**current, **updates})
        for key, value in updates.items():
            if current[key] is not None and value != current[key]:
                raise UsageError("Refinement cannot replace a known worker identity; reconcile the changed session explicitly.", {})
        if any(current[key] != value for key, value in updates.items()):
            row["refinements"].append({"schema_version": 1, "at": at, **updates})
        return row
    return transaction(state_path, mutate)


def expected_assignment(member):
    assignment = dict(member["assignment"])
    for row in member["refinements"]:
        assignment.update({key: row[key] for key in ("pane_id", "native_session")})
    return assignment


def hold(state_path, record, at):
    if (not isinstance(record, dict) or set(record) != {"id", "kind", "resume_condition", "evidence", "dispositions"}
            or record["kind"] not in ("waiting_for_user", "handoff")):
        raise UsageError("Hold requires id, kind (waiting_for_user or handoff), resume_condition, saved user-decision/handoff evidence paths, and dispositions [{member,outcome,evidence}] for every active assignment.", {})
    row = {"schema_version": 1, "id": text(record["id"], "hold id"), "at": at,
           "kind": record["kind"], "resume_condition": text(record["resume_condition"], "resume condition"),
           "evidence": evidence(record["evidence"]), "resumed_at": None}
    dispositions = record["dispositions"]
    if not isinstance(dispositions, list):
        raise UsageError("Hold dispositions must explicitly cover each active assignment.", {})
    prepared = []
    for item in dispositions:
        if not isinstance(item, dict) or set(item) != {"member", "outcome", "evidence"}:
            raise UsageError("Each hold disposition requires member, outcome, and saved evidence paths.", {})
        prepared.append({"schema_version": 1, "member": text(item["member"], "member"),
                         "outcome": text(item["outcome"], "disposition outcome"), "evidence": evidence(item["evidence"])})
    row["dispositions"] = prepared
    timestamp(at)
    def mutate(data):
        if len({item["member"] for item in prepared}) != len(prepared) or {item["member"] for item in prepared} != {item["id"] for item in data["members"] if item["active"]}:
            raise UsageError("Record a named user-held or handed-off disposition for every active assignment; a global hold cannot hide unrelated work.", {})
        if pending(data):
            raise UsageError("Handle pending fleet events before recording a user-held pause or handoff.", {})
        row.update(through=len(data["events"]), members=active_digest(data))
        previous = next((item for item in data["holds"] if item["id"] == row["id"]), None)
        if previous:
            if any(previous[key] != row[key] for key in ("kind", "resume_condition", "evidence", "dispositions", "through", "members")):
                raise UsageError("Hold ID already records a different boundary; use a new ID after reconciling it.", {})
            return previous
        data["holds"].append(row)
        return row
    return transaction(state_path, mutate)


def resume(state_path, at):
    timestamp(at)
    def mutate(data):
        for row in data["holds"]:
            if row["resumed_at"] is None:
                row["resumed_at"] = at
        return {"schema_version": 1, "resumed": True, "pending": pending(data)}
    return transaction(state_path, mutate)


def held(data):
    return any(row["resumed_at"] is None and row["through"] == len(data["events"])
               and row["members"] == active_digest(data) for row in data["holds"])


def identity(value, cwd, environment, *, kind="id", pane_id):
    if kind not in ("id", "path"):
        raise UsageError("Native binding kind must be id or path.", {})
    value = text(value, "native foreman identity")
    if kind == "path":
        if not Path(value).is_absolute():
            raise UsageError("Native transcript binding must use an absolute path.", {})
        value = str(canonical(value))
    return {"kind": kind, "value": value, "cwd": str(canonical(cwd)),
            "herdr_env": text(environment, "HERDR_ENV"), "pane_id": text(pane_id, "HERDR_PANE_ID")}


def binding_path(who, root=None):
    directory = canonical(root) if root is not None else default_state_path().parent / "supervision-bindings"
    return directory / (digest(who) + ".json")


def dispatch_binding(state_path, *, root=None):
    """Read-only dispatch guard: missing historical owner is never legacy mode.

    Scan all saved foreman identities, not only the caller's current identity: a
    context/model change cannot make its earlier obligations disappear. An
    existing empty owner marks an interrupted first bind and is not dispatchable.
    Only a missing owner with no prior discovery may use legacy compatibility.
    """
    data = load(state_path)
    if data["binding"] is not None:
        return data["binding"]
    if store_path(state_path).exists():
        raise StateError("Supervision owner initialization is incomplete. Finish `{}` for this state before dispatching worker input.".format(
            runnable.command("supervision-bind")), {})
    directory = canonical(root) if root is not None else default_state_path().parent / "supervision-bindings"
    _refuse_lost_owner(state_path, directory)
    return None


def _refuse_lost_owner(state_path, directory):
    """Discovery proves prior ownership even after a foreman identity changes."""
    try:
        paths = [path for path in directory.iterdir() if path.suffix == ".json"]
    except FileNotFoundError:
        paths = []
    except OSError as exc:
        raise StateError("Cannot inspect supervision discovery records: {}. Restore directory access before initializing an owner document.".format(exc), {}) from None
    for path in paths:
        prior = read_json(path)
        if isinstance(prior, dict) and prior.get("state_path") == str(canonical(state_path)):
            raise StateError("A saved foreman discovery record already names the missing supervision owner {}. Restore its owner document before rebinding; do not discard unresolved assignments.".format(store_path(state_path)), {})


def bind(state_path, who, at, *, root=None):
    if not isinstance(who, dict) or set(who) != {"kind", "value", "cwd", "herdr_env", "pane_id"}:
        raise UsageError("Binding requires the foreman's exact native kind/value, cwd, HERDR_ENV, and pane_id.", {})
    who = identity(who["value"], who["cwd"], who["herdr_env"], kind=who["kind"], pane_id=who["pane_id"])
    timestamp(at)
    path = binding_path(who, root)
    # An empty unbound first-use owner exists before discovery. No transaction
    # can enroll work until binding commits, so this state is safe to retry.
    # Discovery precedes bound owner. A higher-generation discovery
    # record makes an interrupted handoff block the new foreman, while older
    # sessions can stop once the owner has committed a newer binding.
    with state_lock(store_path(state_path)):
        data = load(state_path)
        old = data["binding"]
        first_use = old is None and not store_path(state_path).exists()
        if first_use:
            _refuse_lost_owner(state_path, path.parent)
        row = old if old is not None and old["identity"] == who else {
            "schema_version": 1, "at": at, "identity": who, "state_path": str(canonical(state_path)),
            "generation": 1 if old is None else old["generation"] + 1}
        with state_lock(path):
            previous = read_json(path)
            if previous is not None and (not isinstance(previous, dict) or previous.get("schema_version") != 1
                    or previous.get("identity") != who or previous.get("state_path") != row["state_path"]):
                raise StateError("Foreman identity already has a conflicting or unreadable state binding. Preserve it and explicitly reconcile the original state path.", {})
            if first_use:
                save_state(store_path(state_path), data)
            save_state(path, row)
            data["binding"] = row
            save_state(store_path(state_path), data)
    return row