"""Source-produced targets -> CoordinateStore/Ref -> actual guarded input port.

The same bridge is exercised with a mock Windows transport in tests. It never
creates screen positions from world coordinates or OCR text bounding boxes.
"""
from dataclasses import dataclass
from datetime import datetime
import hashlib
import json
import threading
from .live_io import InputCommand
from .vendor.coordinate_store import CoordinateStore


@dataclass(frozen=True)
class PreparedAction:
    commands: tuple
    signature: tuple
    evidence_ref: str
    observed_at: str


def fingerprint(value):
    return hashlib.sha256(json.dumps(value, sort_keys=True, ensure_ascii=False, default=str).encode()).hexdigest()


def signature(context):
    window = context["window"]
    if not context.get("character_id") or not context.get("session_id") or not context.get("source_revision"):
        raise ValueError("current_owner_and_source_revision_required")
    return (window.hwnd, window.pid, window.process_started_at, window.exe_path, tuple(window.client_rect),
            context["character_id"], context["session_id"], context["source_revision"])


class SourceActionPort:
    offline_contract = False

    def __init__(self, input_adapter, current_context, *, official_stop=None, pointer_guard=None, escape_guard=None):
        self.input = input_adapter
        self.current_context = current_context
        self.store = CoordinateStore()
        self.cancelled = threading.Event()
        self.official_stop = official_stop
        # (probe_context, current_context, phase) -> current source-bound scene
        # observation; pointer_click additionally returns its verified hit proof.
        self.pointer_guard = pointer_guard
        # (escape_binding, current_context) -> newly read exact owned receiver.
        self.escape_guard = escape_guard

    def prepare(self, proposal, context):
        if proposal.get("status") not in ("ready", "action"):
            raise ValueError("source_target_not_ready")
        current_signature = signature(context)
        window = context["window"]
        if proposal.get("window_signature") and proposal["window_signature"] != context.get("window_signature"):
            raise ValueError("target_window_signature_mismatch")
        if proposal.get("session_id") and proposal["session_id"] != context["session_id"]:
            raise ValueError("target_session_mismatch")
        map_owner = (proposal.get("next_receipt") or {}).get("owner")
        if map_owner:
            if len(map_owner) != 4 or map_owner[0] != window.pid or map_owner[2] != context["character_id"] or map_owner[3] != context["session_id"]:
                raise ValueError("map_target_owner_mismatch")
            if not context.get("observed_process_started_at") or map_owner[1] != context["observed_process_started_at"]:
                raise ValueError("map_process_lifetime_binding_missing_or_changed")
        if proposal.get("frame_id") and proposal["frame_id"] != context.get("frame_id"):
            raise ValueError("source_frame_changed")
        evidence = proposal.get("source_refs") or proposal.get("evidence")
        if not evidence:
            raise ValueError("source_evidence_required")
        captured = proposal.get("observed_at") or (proposal.get("next_receipt") or {}).get("captured_at") or context.get("observed_at")
        if not captured:
            raise ValueError("observation_time_required")
        rows = proposal.get("actions")
        if rows is None:
            action = proposal.get("action")
            rows = [action] if action else [{"kind": proposal.get("input_kind", "click"),
                "role": proposal["target_id"], "point": proposal.get("point_client"), "key": proposal.get("key")}]
        evidence_ref = fingerprint({"proposal": proposal, "signature": current_signature})
        coordinates, prepared = {}, []
        for index, row in enumerate(rows):
            if row.get('kind')=='owned_escape':
                if len(rows)!=1 or row.get('key')!='ESC' or not isinstance(row.get('escape_binding'),dict):
                    raise ValueError('owned_escape_requires_one_current_receiver_boundary')
                row=json.loads(json.dumps(row,ensure_ascii=False,allow_nan=False))
            if row.get('kind') in ('pointer_probe', 'pointer_click', 'pointer_scroll'):
                if len(rows) != 1:
                    raise ValueError('pointer_probe_requires_separate_observation_boundary')
                if not row.get('probe_context') or not row.get('source_rect') or not row.get('safe_rect'):
                    raise ValueError('pointer_source_target_binding_required')
                if not isinstance(row.get('point'), (list, tuple)) or len(row['point']) != 2:
                    raise ValueError('pointer_source_target_point_required')
                # Retain JSON provenance, never live objects or mutable aliases.
                row = json.loads(json.dumps(row, ensure_ascii=False, allow_nan=False))
            role = row.get("role") or f"authored-key-{index}"
            key = f"{index}:{role}"
            coordinates[key] = row.copy()
            prepared.append((key, float(row.get("delay_before_seconds", 0))))
        self.store.publish(coordinates, signature=current_signature)
        return PreparedAction(tuple((self.store.ref(key), delay) for key, delay in prepared), current_signature, evidence_ref, captured)

    def send(self, prepared, operation_id):
        if not isinstance(prepared, PreparedAction):
            raise ValueError("prepared_source_action_required")
        receipts = []
        for index, (reference, delay) in enumerate(prepared.commands):
            if not 0 <= delay <= 60.25:
                return {"status": "unknown" if receipts else "blocked", "attempted": bool(receipts), "reason": "invalid_authored_delay", "receipts": receipts}
            if self.cancelled.wait(delay):
                return {"status": "unknown" if receipts else "blocked", "attempted": bool(receipts), "reason": "user_stop", "receipts": receipts}
            context = self.current_context()
            if signature(context) != prepared.signature:
                self.store.invalidate()
                return {"status": "unknown" if receipts else "blocked", "attempted": bool(receipts), "reason": "owner_or_layout_changed", "receipts": receipts}
            parsed = datetime.fromisoformat(prepared.observed_at.replace("Z", "+00:00"))
            # Prepared coordinates do not expire while the owned tape waits
            # for its authored delay. Owner/window/source, monitor epoch and
            # typed dynamic target guards remain at the actual input boundary.
            if parsed.tzinfo is None:
                return {"status": "unknown" if receipts else "blocked", "attempted": bool(receipts), "reason": "observation_timezone_required", "receipts": receipts}
            row = self.store.resolve(reference, current_signature=signature(context))
            kind, point = row["kind"], row.get("point", row.get("start"))
            if kind == 'official_stop':
                if self.official_stop is None or len(prepared.commands)!=1:
                    return {'status':'blocked','attempted':False,'reason':'owned_stop_adapter_required'}
                receipt=self.official_stop.send(row, f'{operation_id}:{index}', context)
                return {'status':receipt['status'],'attempted':receipt.get('attempted'),
                    'effect_confirmed':False,'receipts':[receipt],'evidence_ref':prepared.evidence_ref}
            kwargs = dict(kind=kind, target_id=row.get("role", "authored-key"), evidence_ref=prepared.evidence_ref,
                          expected_rect=tuple(context["window"].client_rect))
            if kind in ("key", "owned_escape"):
                kwargs["key"] = row["key"]
                if kind=='owned_escape':
                    kwargs.update(escape_binding=row['escape_binding'],
                        input_epoch=row.get('input_epoch') or row['escape_binding'].get('input_epoch'))
            elif kind in ("bag_text", "craft_text"):
                kwargs.update(text=row["text"], text_scope=row["text_scope"])
            else:
                kwargs.update(x=point[0], y=point[1])
                if kind in ("scroll", "map_drag", "map_long_press", "pointer_probe", "pointer_click", "pointer_scroll"):
                    from .coordinate_geometry import CLIENT_XYWH
                    if row.get('safe_rect_format',CLIENT_XYWH)!=CLIENT_XYWH:
                        raise ValueError('actuator_safe_rect_requires_physical_client_xywh')
                    kwargs["safe_rect"] = tuple(row.get("safe_rect") or context.get("safe_rect") or ())
                if kind in ('pointer_probe', 'pointer_click', 'pointer_scroll'):
                    kwargs.update(source_rect=tuple(row.get('source_rect') or ()),
                        input_epoch=row.get('input_epoch') or row['probe_context'].get('input_epoch'),
                        probe_context=row['probe_context'], pointer_verified=row.get('pointer_verified'))
                if kind in ("scroll", "pointer_scroll"):
                    kwargs["delta"] = row["delta"]
                if kind == "map_drag":
                    kwargs.update(end_x=row["end"][0], end_y=row["end"][1], duration=row.get("duration", .25))
                if kind == "map_long_press":
                    kwargs["hold_seconds"] = row["hold_seconds"]
            command = InputCommand(**kwargs)
            if kind in ('pointer_probe', 'pointer_click', 'pointer_scroll'):
                def check_pointer(command, current_window):
                    latest = self.current_context()
                    if signature(latest) != prepared.signature or current_window.identity != latest['window'].identity:
                        raise ValueError('pointer_owner_or_layout_changed')
                    if self.pointer_guard is None:
                        raise ValueError('pointer_current_scene_reader_required')
                    result = self.pointer_guard(command.probe_context, latest, command.kind)
                    # Guard reads can span a UI/owner transition. Check the
                    # calling context again before returning to the writer.
                    if signature(self.current_context()) != prepared.signature:
                        raise ValueError('pointer_owner_or_layout_changed_during_read')
                    return result
                receipt = self.input.send(command, f'{operation_id}:{index}', pointer_guard=check_pointer)
            elif kind=='owned_escape':
                def check_escape(command,current_window):
                    latest=self.current_context();binding=command.escape_binding
                    if (signature(latest)!=prepared.signature or current_window.identity!=latest['window'].identity or
                            binding.get('window_signature')!=latest.get('window_signature') or
                            binding.get('frame_id')!=latest.get('frame_id')):
                        raise ValueError('owned_escape_owner_or_frame_changed')
                    if self.escape_guard is None:raise ValueError('owned_escape_current_receiver_reader_required')
                    result=self.escape_guard(binding,latest)
                    after=self.current_context()
                    if (signature(after)!=prepared.signature or binding.get('window_signature')!=after.get('window_signature') or
                            binding.get('frame_id')!=after.get('frame_id')):
                        raise ValueError('owned_escape_owner_or_frame_changed_during_read')
                    return result
                receipt=self.input.send(command,f'{operation_id}:{index}',escape_guard=check_escape)
            else:
                receipt = self.input.send(command, f"{operation_id}:{index}")
            receipts.append(receipt)
            if receipt.get("status") not in ("sent_unconfirmed", "dry_run"):
                return {"status": "unknown" if receipt.get("attempted") or len(receipts)>1 else "blocked",
                        "attempted": receipt.get("attempted") is not False or len(receipts)>1,
                        "reason": "tape_stopped_at_input_boundary", "receipts": receipts}
        return {"status": "dry_run" if all(r["status"] == "dry_run" for r in receipts) else "sent_unconfirmed",
                "effect_confirmed": False, "receipts": receipts, "evidence_ref": prepared.evidence_ref}

    def stop(self):
        self.cancelled.set()
        self.input.stop("user_stop")
        self.store.invalidate()
