"""Durable common graph executor; all quest-specific values are graph data."""
from copy import deepcopy
from datetime import datetime, timezone
import json
import uuid
from .arbeit_engine import Journal, stamp, utcnow
from .arbeit_graph import digest
from .material_execution import MaterialExecution


def current_observation_clock(snapshot):
    """Validate the actual reader interval; never relabel individual evidence."""
    clock=snapshot.get('observation_clock')
    if clock is None:return None
    if not isinstance(clock,dict) or clock.get('source')!='GameObserver.read:perf_counter':
        raise ValueError('observation_clock_source_invalid')
    if (not isinstance(clock.get('producer_id'),str) or not clock['producer_id']
            or any(not clock.get(k) or clock[k]!=snapshot.get(k) for k in
                ('sequence','observation_id','session_id','character_id','process_started_at'))
            or type(clock.get('started_ns')) is not int or type(clock.get('completed_ns')) is not int
            or not 0<=clock['started_ns']<=clock['completed_ns']):
        raise ValueError('observation_clock_scope_or_interval_invalid')
    return deepcopy(clock)


def later_same_wall_transaction(previous,current):
    return bool(previous and current
        and all(current.get(k)==previous.get(k) for k in
            ('source','producer_id','session_id','character_id','process_started_at'))
        and current['sequence']>previous['sequence']
        and current['observation_id']!=previous['observation_id']
        and current['started_ns']>=previous['completed_ns']
        and current['completed_ns']>previous['completed_ns'])


class GraphExecutor(MaterialExecution):
    def __init__(self, graph, path, *, mode="offline-contract", resume=False):
        if mode not in ("offline-contract", "live"):
            raise ValueError("Unknown execution mode")
        self.graph = deepcopy(graph)
        self.journal = Journal(path, resume=resume)
        if resume:
            self.data = json.loads(self.journal.path.read_text(encoding="utf8"))
            if self.data["mode"] != mode or self.data["graph_fingerprint"] != digest(graph):
                self.journal.close()
                raise ValueError("Saved graph mode or sources changed; reconcile before resume")
            if self.data["status"] not in ("complete", "offline_contract_passed"):
                self.data["status"], self.data["reason"] = "paused", "explicit_resume_required"
        else:
            self.data = {"schema": "masukja.graph-run.v1", "run_id": uuid.uuid4().hex, "mode": mode,
                "quest_id": graph["quest_id"], "graph_fingerprint": digest(graph),
                "status": "paused", "reason": "explicit_start_required", "cursor": 0,
                "character_id": None, "session_id": None, "quest_accept_time": None, "last_sequence": 0, "last_observed_at": None,
                "nodes": {}, "pending": None, "events": [], "recovery": None}
        self.data.setdefault("goal_reservations", [])
        self.save()

    def save(self):
        self.journal.write(self.data)

    def report(self):
        return {"quest_id": self.graph["quest_id"], "title": self.graph["title"],
            "status": self.data["status"], "reason": self.data["reason"], "mode": self.data["mode"],
            "cursor": self.data["cursor"], "node_count": len(self.graph["nodes"]),
            "node": self.current_node(), "pending": self.data["pending"], "recovery": self.data["recovery"],
            "active_pending":self.reported_pending(), "material_supply":self.data.get('material_supply'),
            "wait_contract": deepcopy(self.data.get('wait_contract')), "journal": str(self.journal.path)}

    def current_node(self):
        return self.graph["nodes"][self.data["cursor"]] if self.data["cursor"] < len(self.graph["nodes"]) else None

    def event(self, kind, reason, **data):
        self.data["events"].append({"at": utcnow(), "kind": kind, "reason": reason, **data})

    def start(self, *, explicit=False, blockers=()):
        if self.data["status"] in ("complete", "offline_contract_passed"):
            return self.report()
        if not explicit:
            return self.boundary("explicit_start_required", "paused")
        if blockers:
            self.event("preflight", "preflight_incomplete", blockers=list(blockers))
            return self.boundary("preflight_incomplete", "blocked")
        return self.boundary("explicit_start", "running")

    def boundary(self, reason, status="waiting", *, wait_contract=None):
        self.data["reason"], self.data["status"] = reason, status
        self.data['wait_contract']=deepcopy(wait_contract) if status=='waiting' else None
        self.event("boundary", reason, status=status)
        self.save()
        return self.report()

    def stop(self, reason="user_stop"):
        if self.data["status"] in ("complete", "offline_contract_passed"):
            return self.report()
        return self.boundary(reason, "paused")

    def fresh_owned_proof(self, proof, snapshot):
        if (not proof.get("source") or proof.get("session_id") != snapshot["session_id"] or
                proof.get("character_id") != snapshot["character_id"]):
            return False
        try:
            now = datetime.now(timezone.utc) if self.data["mode"] == "live" else stamp(snapshot["observed_at"])
            return -.5 <= (now-stamp(proof["observed_at"])).total_seconds() <= 5
        except (KeyError, TypeError, ValueError):
            return False

    def recover_bag(self, node, snapshot, action_port):
        from .bag_recovery import plan_bag_recovery
        recovery = self.data["recovery"]
        if recovery is None:
            recovery = self.data["recovery"] = {"parent_node": node["id"], "kind": "bag_obstruction",
                "parent_pending_preserved": bool(self.data["pending"]), "child_pending": None}
        child = recovery.get("child_pending")
        result = plan_bag_recovery(snapshot, (child or {}).get("planner_receipt"))
        self.event("recovery_operation", result.get("reason"), node_id=node["id"], result=result)
        if result.get("status") in ("recovered", "complete") and result.get("effect_evidence"):
            self.event("recovery", "bag_obstruction_cleared", parent_node=node["id"], recovery=recovery,
                evidence=result["effect_evidence"])
            self.data["recovery"] = None
            self.save()
            return None
        if result.get("status") != "action":
            return self.boundary(result.get("reason", "bag_recovery_effect_unconfirmed"))
        if child and result.get("prior_effect_confirmed") is not True:
            return self.boundary("pending_recovery_input_requires_current_effect")
        if (self.data["mode"] == "offline-contract") != bool(getattr(action_port, "offline_contract", False)):
            return self.boundary("recovery_port_mode_mismatch")
        proposal = result.get("proposal")
        try:
            prepared = action_port.prepare(proposal, snapshot["input_context"])
        except Exception as exc:
            return self.boundary("recovery_target_preparation_failed:" + str(exc))
        operation_id = f"{self.data['run_id']}:{node['id']}:recovery:{len(self.data['events'])}"
        recovery["child_pending"] = {"operation_id": operation_id, "status": "intent_persisted",
            "proposal": proposal, "planner_receipt": result["next_receipt"], "after_sequence": snapshot["sequence"]}
        self.save()
        try:
            sent = action_port.send(prepared, operation_id)
        except Exception as exc:
            sent = {"status": "unknown", "reason": type(exc).__name__ + ":" + str(exc)}
        recovery["child_pending"].update(status=sent.get("status", "unknown"), input_receipt=sent)
        if sent.get("status") == "blocked" and sent.get("attempted") is False:
            self.event("recovery_input_not_attempted", sent.get("reason"), intent=recovery["child_pending"])
            recovery["child_pending"] = None
        return self.boundary("recovery_input_requires_effect_observation")

    def tick(self, observer, action_port, *, planner=None):
        if self.data["status"] not in ("running", "waiting"):
            return self.report()
        if planner is None:
            from .arbeit_operations import plan_operation
            planner = plan_operation
        node = self.current_node()
        if node is None:
            raise ValueError("No current node outside a terminal run")
        # Read ONLY the declared boundary. No periodic OCR is inserted into a
        # fixed tape; SourceActionPort consumes that tape itself.
        try:
            snapshot = observer.read(self.observation_node())
        except Exception as exc:
            return self.boundary("observation_error:" + type(exc).__name__ + ":" + str(exc))
        if snapshot.get("mode") != self.data["mode"]:
            return self.boundary("observation_mode_mismatch")
        if snapshot.get("ok") is not True:
            self.event("observation_gap", snapshot.get("reason", "observation_unavailable"), missing=snapshot.get("missing", []))
            return self.boundary(snapshot.get("reason", "observation_unavailable"))
        required = ("character_id", "session_id", "observation_source", "observed_at", "sequence")
        if any(not snapshot.get(key) for key in required):
            return self.boundary("observation_provenance_incomplete")
        observed = stamp(snapshot["observed_at"])
        if self.data["mode"] == "live" and not -.5 <= (datetime.now(timezone.utc)-observed).total_seconds() <= 5:
            return self.boundary("current_observation_stale_or_future")
        try:
            clock=current_observation_clock(snapshot)
        except ValueError as exc:
            return self.boundary(str(exc))
        previous_wall=stamp(self.data['last_observed_at']) if self.data['last_observed_at'] else None
        equal_wall_ordered=(previous_wall==observed and later_same_wall_transaction(
            self.data.get('last_observation_clock'),clock))
        previous_clock=self.data.get('last_observation_clock')
        same_tick_unresolved=(previous_wall==observed and previous_clock and clock
            and snapshot['sequence']>self.data['last_sequence']
            and all(clock.get(k)==previous_clock.get(k) for k in
                ('source','producer_id','session_id','character_id','process_started_at'))
            and clock['observation_id']!=previous_clock['observation_id']
            and clock['started_ns']==clock['completed_ns']==previous_clock['completed_ns'])
        if same_tick_unresolved:
            # Do not consume the sequence, call the planner or confirm effects.
            # A later real read can supply an ordered interval; no clock is invented.
            return self.boundary('observation_clock_tick_unresolved')
        if snapshot["sequence"] <= self.data["last_sequence"] or (
                previous_wall is not None and observed<=previous_wall and not equal_wall_ordered):
            return self.boundary("repeated_observation")
        if self.data["character_id"] and (snapshot["character_id"] != self.data["character_id"] or snapshot["session_id"] != self.data["session_id"]):
            return self.stop("character_or_session_changed")
        self.data.update(character_id=snapshot["character_id"], session_id=snapshot["session_id"],
            last_sequence=snapshot["sequence"], last_observed_at=snapshot["observed_at"], last_observation_clock=clock)
        direct_pending=self.data.get('pending') or {}
        direct_scope=((direct_pending.get('planner_receipt') or {}).get('request_binding') or {}).get('before_scope')
        direct_accepted_effect=None
        if direct_scope:
            from .direct_dialogue_request import reconcile_direct_dialogue_request
            reconciliation=reconcile_direct_dialogue_request(node,snapshot,direct_pending)
            if reconciliation.get('status')=='effect_observed' and reconciliation['effect_evidence'].get('state')=='accepted':
                direct_accepted_effect=reconciliation['effect_evidence']
        def original_direct_occurrence(proof):
            # A rejected current proof must not poison the graph's global
            # accept-time latch before the exact pending request reconciles.
            return not direct_scope or (direct_accepted_effect is not None and proof==direct_accepted_effect)
        accepted_times = {p.get("accept_time") for p in snapshot.get("state_proofs", [])
            if p.get("quest_id") == self.graph["quest_id"] and p.get("state") == "accepted" and p.get("confirmed") is True
            and self.fresh_owned_proof(p, snapshot) and original_direct_occurrence(p)
            and type(p.get("accept_time")) is int and p["accept_time"] > 0}
        if len(accepted_times) > 1:
            return self.stop("current_quest_occurrence_ambiguous")
        if accepted_times:
            accept_time = next(iter(accepted_times))
            if self.data.get("quest_accept_time") and self.data["quest_accept_time"] != accept_time:
                return self.stop("quest_occurrence_changed")
            self.data["quest_accept_time"] = accept_time
        from .goal_reservations import reconcile_reservations
        reservations, gaps = reconcile_reservations(self.graph, self.data["goal_reservations"], snapshot,
            self.data.get("quest_accept_time"), self.fresh_owned_proof)
        self.data["goal_reservations"] = reservations
        inherited_reservations=snapshot.get('goal_reservations',[])
        inherited_gaps=snapshot.get('goal_reservation_gaps',[])
        snapshot['goal_reservations']=reservations+[r for r in inherited_reservations if r not in reservations]
        snapshot['goal_reservation_gaps']=gaps+[r for r in inherited_gaps if r not in gaps]
        self._material_protection(snapshot)
        self.save()
        if self.data.get('material_supply'):
            return self._tick_materials(snapshot,observer,action_port,planner)
        acceptance_pending=self.data['pending']
        acceptance_receipt=(acceptance_pending or {}).get('planner_receipt') or {}
        from .acceptance_movement import acceptance_map_card_effect,acceptance_native_effect
        acceptance_proof=(acceptance_map_card_effect(node,snapshot,acceptance_receipt)
            or acceptance_native_effect(node,snapshot,acceptance_receipt))
        native_acceptance_resolved=acceptance_proof is not None
        if self.data["recovery"] is None and (acceptance_pending is None or native_acceptance_resolved) and (node["op"] in ("quest.offer", "dialogue.accept") or
                (node["op"] == "move.target" and node.get("params", {}).get("role") == "acceptance")):
            accepted = next((n for n in self.graph["nodes"] if n["op"] == "quest.accepted"), None)
            if accepted is not None:
                accepted_snapshot=(dict(snapshot,state_proofs=[acceptance_proof])
                    if acceptance_proof is not None else snapshot)
                proof = planner(accepted, accepted_snapshot, None)
                evidence = proof.get("effect_evidence") or proof.get("evidence")
                if proof.get("status") == "complete" and evidence:
                    target_index = self.graph["nodes"].index(accepted)
                    for previous in self.graph["nodes"][self.data["cursor"]:target_index]:
                        self.data["nodes"][previous["id"]] = {"status": "not_executed_already_accepted",
                            "accepted_evidence": evidence, "not_action_success": True,
                            "resolved_input":acceptance_pending if previous['id']==node['id'] else None}
                    self.data['pending']=None
                    self.data["cursor"] = target_index
                    node = accepted
        # Same-step source array order is not execution precedence. A current
        # typed/uniquely matched objective can select another member, and only
        # explicit step-effect proof may close an unexecuted branch.
        if node.get("objective_index") is not None and self.data["pending"] is None and self.data["recovery"] is None:
            active = snapshot.get("active_objectives", [])
            candidates = [n for n in self.graph["nodes"] if n.get("objective_index") is not None and
                n.get("step_id") == node["step_id"] and n["id"] not in self.data["nodes"] and
                any(a.get("quest_id") == n["quest_id"] and a.get("step_id") == n["step_id"] and
                    a.get("objective_index") == n["objective_index"] for a in active)]
            if candidates:
                node = candidates[0]
                self.data["cursor"] = self.graph["nodes"].index(node)
            elif any(e.get("quest_id") == node["quest_id"] and e.get("step_id") == node["step_id"] and
                     e.get("complete") is True and e.get("source") for e in snapshot.get("step_effects", [])):
                node = next(n for n in self.graph["nodes"] if n["op"] == "quest.step_complete" and n["step_id"] == node["step_id"])
                self.data["cursor"] = self.graph["nodes"].index(node)
        pending = self.data["pending"]
        # Native StartAutoPlayRequested owns an entire quest instance. It may
        # advance an optional branch without executing this journal objective.
        # Only a fresh typed step effect can resolve that pending input; record
        # closure separately from objective success, and retain the receipt.
        native_receipt = (pending or {}).get('planner_receipt') or {}
        native_child = native_receipt.get('child_receipt') or {}
        if (self.data['recovery'] is None and node.get('objective_index') is not None
                and native_child.get('action_scope') == 'quest_instance'):
            scope = native_child.get('requested_objective') or {}
            proofs = [p for p in snapshot.get('state_proofs', [])
                if p.get('state') == 'step_complete' and p.get('confirmed') is True
                and p.get('quest_id') == node['quest_id'] and p.get('step_id') == node['step_id']
                and p.get('instance_id') == scope.get('instance_id')
                and p.get('accept_time') == scope.get('accept_time')
                and stamp(p['observed_at']) > stamp(native_receipt['observed_at'])
                and self.fresh_owned_proof(p, snapshot)]
            if len(proofs) == 1:
                self.data['nodes'][node['id']] = {'status':'not_executed_branch_closed',
                    'not_objective_success':True, 'step_effect_evidence':proofs[0], 'pending':pending}
                self.data['pending'] = pending = None
                node = next(n for n in self.graph['nodes'] if n['op']=='quest.step_complete'
                    and n['step_id']==node['step_id'])
                self.data['cursor'] = self.graph['nodes'].index(node)
                self.event('native_quest_scope_resolved','current_typed_step_closed_pending_branch',
                    evidence=proofs[0], prior_input=native_receipt)
                self.save()
        if snapshot.get("bag_blocked") is True or self.data["recovery"]:
            result = self.recover_bag(node, snapshot, action_port)
            if result is not None:
                return result
        receipt = (pending or {}).get("planner_receipt")
        result = planner(node, snapshot, receipt)
        self.event("operation", result.get("reason", result.get("status", "unknown")), node_id=node["id"], result=result)
        material_result=self._begin_materials(node,snapshot,result)
        if material_result is not None:return material_result
        if result.get('status')=='restore_pending':
            saved=(pending or {}).get('planner_receipt') or {}
            before=(saved.get('child_receipt') or {}).get('before_probe_pending') or None
            proofs=result.get('effect_evidence') or []
            if (saved.get('phase')!='alter.pointer_probe' or 'original_pending' not in result
                    or result['original_pending']!=before or not proofs
                    or result.get('prior_effect_confirmed') is not True
                    or not all(self.fresh_owned_proof(p,snapshot) for p in proofs)):
                return self.boundary('processing_probe_restore_evidence_required')
            self.event('processing_probe_resolved',result['reason'],node_id=node['id'],
                probe_pending=pending,original_pending=before,effect_evidence=proofs)
            self.data['pending']=deepcopy(before)
            return self.boundary(result['reason'],'running')
        if result.get('status')=='supply_retry':
            if (not pending or not result.get('effect_evidence') or result.get('prior_effect_confirmed') is not True
                    or result.get('released_candidate') is None):
                return self.boundary('supply_retry_current_cleanup_evidence_required')
            self.event('supply_candidate_released',result.get('reason','owned_supply_UI_cleanup_confirmed'),
                node_id=node['id'],pending=pending,effect_evidence=result['effect_evidence'],
                candidate=result['released_candidate'])
            self.data['pending']=None
            return self.boundary('current_supply_candidate_released_select_next','running')
        if result.get('status') == 'transition':
            if not result.get('effect_evidence') or not result.get('next_receipt'):
                return self.boundary('intermediate_transition_has_no_effect_evidence')
            self.data['pending'] = {'node_id':node['id'], 'status':'effect_confirmed',
                'planner_receipt':result['next_receipt'], 'resolved_input':pending,
                'intermediate_effect_evidence':result['effect_evidence']}
            return self.boundary(result.get('reason','intermediate_effect_confirmed'), 'running')
        if result.get("status") == "complete":
            evidence = result.get("effect_evidence") or result.get("evidence")
            if not evidence:
                return self.boundary("operation_completion_has_no_evidence")
            self.data["nodes"][node["id"]] = {"status": "complete", "effect_evidence": evidence,
                "pending": pending, "observed_at": snapshot["observed_at"], "sequence": snapshot["sequence"]}
            if node.get('objective_index') is not None and (
                    result.get('individual_objective_success_verified') is False
                    or result.get('individual_craft_success_verified') is False
                    or result.get('not_objective_success') is True):
                self.data['nodes'][node['id']].update(status='not_executed_branch_closed',
                    not_objective_success=True,owned_cleanup_complete=True,
                    completion_basis=result.get('completion_basis','native step ended after owned cleanup'))
            self.data["pending"] = None
            self._material_node_complete(node,evidence)
            if node.get("objective_index") is not None:
                remaining = [n for n in self.graph["nodes"] if n.get("objective_index") is not None and
                    n.get("step_id") == node["step_id"] and n["id"] not in self.data["nodes"]]
                next_node = remaining[0] if remaining else next(n for n in self.graph["nodes"] if n["op"] == "quest.step_complete" and n["step_id"] == node["step_id"])
                self.data["cursor"] = self.graph["nodes"].index(next_node)
            else:
                if node["op"] == "quest.step_complete":
                    for sibling in self.graph["nodes"]:
                        if sibling.get("objective_index") is not None and sibling.get("step_id") == node["step_id"] and sibling["id"] not in self.data["nodes"]:
                            self.data["nodes"][sibling["id"]] = {"status": "not_executed_branch_closed",
                                "step_effect_evidence": evidence, "not_objective_success": True}
                self.data["cursor"] += 1
            terminal = self.data["cursor"] == len(self.graph["nodes"])
            status = ("complete" if self.data["mode"] == "live" else "offline_contract_passed") if terminal else "running"
            return self.boundary("effect_confirmed", status)
        if result.get("status") != "action":
            return self.boundary(result.get("reason") or result.get("wait_reason") or "operation_effect_pending",
                wait_contract=result.get('wait_contract'))
        if pending and result.get("prior_effect_confirmed") is not True:
            return self.boundary("pending_input_requires_current_effect")
        if self.data["mode"] == "offline-contract" and not getattr(action_port, "offline_contract", False):
            return self.boundary("offline_contract_cannot_send_live_input")
        if self.data["mode"] == "live" and getattr(action_port, "offline_contract", False):
            return self.boundary("live_cannot_use_fake_port")
        proposal = result.get("proposal")
        if not proposal:
            return self.boundary("source_proposal_missing")
        operation_id = f"{self.data['run_id']}:{node['id']}:{len(self.data['events'])}"
        # All preparation/geometry calculation is before dispatch and cannot
        # mutate the game. Persist the exact intent before the actuator call.
        try:
            prepared = action_port.prepare(proposal, snapshot["input_context"])
        except Exception as exc:
            return self.boundary("source_target_preparation_failed:" + str(exc))
        self.data["pending"] = {"node_id": node["id"], "operation_id": operation_id,
            "after_sequence": snapshot["sequence"], "status": "intent_persisted",
            "planner_receipt": result.get("next_receipt") or {"phase": "input_sent", "node_id": node["id"], "observation_id": snapshot.get("observation_id")},
            "proposal": proposal}
        self.save()
        try:
            sent = action_port.send(prepared, operation_id)
        except Exception as exc:
            sent = {"status": "unknown", "reason": type(exc).__name__ + ":" + str(exc)}
        self.data["pending"].update(status=sent.get("status", "unknown"), input_receipt=sent)
        if sent.get("status") == "blocked" and sent.get("attempted") is False:
            self.event("input_not_attempted", sent.get("reason", "blocked_before_dispatch"), intent=self.data["pending"])
            # Keep the predecessor's owned action and baseline. A rejected
            # successor is not a reason to restart the production/close flow.
            self.data["pending"] = pending
            return self.boundary("input_blocked_before_dispatch_requires_fresh_prepare")
        return self.boundary("input_requires_effect_observation")

    def close(self):
        self.stop("application_closed")
        self._close_materials()
        self.journal.close()
