"""One causal event reducer shared by replay and a live local JSONL follower."""
from collections import Counter, defaultdict
from copy import deepcopy
import hashlib
import json
import time
from statistics import mean, median

PARAMETERS = dict(directed=True, two_level=True, seed=11, num_trials=5,
                  num_threads=1, silent=True, markov_time=1,
                  teleportation_probability=.15, no_self_links=False)


def quantile(xs, p):
    if not xs:
        return None
    s = sorted(xs); x = (len(s)-1)*p; i = int(x)
    return s[i] + (s[min(i+1, len(s)-1)]-s[i])*(x-i)


def infomap_partition(nodes, weights):
    from infomap import Infomap
    if not nodes:
        return [], None
    ids = {s: i for i, s in enumerate(sorted(nodes))}
    rev = {v: k for k, v in ids.items()}
    im = Infomap(**PARAMETERS)
    for s, i in ids.items():
        im.add_node(i)
    for (u, v), weight in sorted(weights.items()):
        if weight:
            im.add_link(ids[u], ids[v], float(weight))
    im.run()
    groups = defaultdict(list)
    for i, module in im.get_modules().items():
        groups[module].append(rev[i])
    return sorted([sorted(v) for v in groups.values()], key=lambda x: (-len(x), x)), float(im.codelength)


class EntryCatalog:
    """Only explicit reviewed members, scoped by source Activity; no text heuristic."""
    def __init__(self, groups=(), allow_oracle=False):
        self.groups = list(groups)
        self.allow_oracle = allow_oracle
        self.by_transition = defaultdict(list)
        self.by_identity = defaultdict(list)
        for g in self.groups:
            for t in g.get('transition_ids', []):
                self.by_transition[(g['run_id'], t)].append(g)
            for m in g.get('members', []):
                if not m.get('concrete_identity'):
                    raise ValueError('ModelAction alone cannot define a functional entry')
                self.by_identity[(g['run_id'], m['source_state'], m['model_action_identity'], m['concrete_identity'])].append(g)

    def match(self, event):
        candidates = self.by_transition[(event['run_id'], event['transition_identity'])] + self.by_identity[(event['run_id'], event['source_state'], event['model_action_identity'], event.get('concrete_identity'))]
        out = {}
        for g in candidates:
            if g.get('category') != 'I' or g.get('status', 'exploratory') not in ('confirmed', 'exploratory'):
                continue
            if g.get('oracle', False):
                if not self.allow_oracle:
                    continue
            elif g.get('known_at_event_seq') is None or g['known_at_event_seq'] > event['event_seq']:
                continue
            activity = g.get('source_activity')
            if not activity or activity == 'UNKNOWN':
                if g.get('source_state') != event['source_state']:
                    continue
            elif activity != event.get('source_activity'):
                continue
            out[g['group_id']] = g
        return list(out.values())


def region_metrics(events, members, recent=100):
    R = set(members); inside = [e for e in events if e['source_state'] in R]
    internal = [e for e in inside if e['target_state'] in R]
    entries = [e for e in events if e['source_state'] not in R and e['target_state'] in R]
    exits = [e for e in inside if e['target_state'] not in R]
    episodes = []; active = None; exited = set(); reentries = 0
    for e in events:
        u, v, seg = e['source_state'] in R, e['target_state'] in R, e['_segment']
        if active and (active['segment'] != seg or not u):
            active['right_censored'] = True; episodes.append(active); active = None
        if u and active is None:
            active = dict(start=e['event_seq'], end=e['event_seq'], segment=seg, left_censored=True, right_censored=True, steps=0)
        if u:
            active['steps'] += 1; active['end'] = e['event_seq']
        if not u and v:
            if seg in exited:
                reentries += 1
            active = dict(start=e['event_seq'], end=e['event_seq'], segment=seg, left_censored=False, right_censored=True, steps=0)
        if u and not v:
            active['right_censored'] = False; episodes.append(active); active = None; exited.add(seg)
    if active:
        episodes.append(active)
    dwell = [p['steps'] for p in episodes if not p['left_censored'] and not p['right_censored']]
    tail = events[-recent:]; latest = [e for e in tail if e['source_state'] in R]
    def progress(es):
        snaps = [e['model_progress'] for e in es if isinstance(e.get('model_progress'), dict)]
        return dict(steps=len(es), new_state=sum(e['_new_state'] for e in es),
                    new_transition=sum(e['_new_transition'] for e in es),
                    new_pair=sum(e['_new_pair'] for e in es),
                    native_new_model=sum(e.get('native_new_model_action') is True for e in es),
                    native_progress_records=len(snaps),
                    native_selected_unvisited=sum(s.get('selected_model_action_unvisited') is True for s in snaps),
                    native_selected_unsaturated=sum(s.get('selected_model_action_saturated') is False for s in snaps),
                    native_unvisited_positive=sum((s.get('unvisited_model_action_count') or 0) > 0 for s in snaps),
                    native_state_saturated=sum(s.get('native_state_saturated') is True for s in snaps))
    patterns = Counter()
    for p in episodes:
        if not p['left_censored'] and not p['right_censored']:
            pattern = tuple((e['source_state'], e['model_action_identity'], e['target_state']) for e in events if p['start'] <= e['event_seq'] <= p['end'] and e['_segment'] == p['segment'])
            patterns[pattern] += 1
    return dict(size=len(R), source_steps=len(inside), budget_share=len(inside)/len(events) if events else 0,
                internal_flow=len(internal), boundary_entries=len(entries), boundary_exits=len(exits),
                reentries=reentries, p_stay=len(internal)/len(inside) if inside else None,
                recent_window=len(tail), recent_steps=len(latest), recent_budget_share=len(latest)/len(tail) if tail else 0,
                repeated_state_share=sum(not e['_new_state'] for e in inside)/len(inside) if inside else None,
                repeated_transition_share=sum(not e['_new_transition'] for e in inside)/len(inside) if inside else None,
                complete_dwell_count=len(dwell), dwell_mean=mean(dwell) if dwell else None,
                dwell_median=median(dwell) if dwell else None, dwell_p90=quantile(dwell, .9), dwell_max=max(dwell, default=None),
                episodes=episodes, progress=progress(inside), recent_progress=progress(latest),
                top_paths=[dict(count=n, path=list(p)) for p, n in patterns.most_common(3)])


def lineage(previous, current, next_id):
    """Keep every nonempty intersection; inherit only unambiguous mutual >= .5 match."""
    links = []; parents = defaultdict(list); children = defaultdict(list)
    for old in previous:
        a = set(old['members'])
        for new in current:
            b = set(new['members']); intersection = len(a & b)
            if not intersection:
                continue
            link = dict(previous=old['id'], current=new['id'], previous_lineage=old['lineage_id'],
                        intersection=intersection, jaccard=intersection/len(a | b),
                        old_fraction=intersection/len(a), new_fraction=intersection/len(b),
                        relation='identical' if a == b else 'expansion' if a < b else 'contraction' if b < a else 'overlap')
            links.append(link); parents[new['id']].append(link); children[old['id']].append(link)
    events = []
    for old, rel in children.items():
        if len(rel) > 1:
            events.append(dict(kind='split', previous=[old], current=[l['current'] for l in rel]))
    for new in current:
        rel = parents[new['id']]
        if len(rel) > 1:
            events.append(dict(kind='merge', previous=[l['previous'] for l in rel], current=[new['id']]))
        if len(rel) == 1 and len(children[rel[0]['previous']]) == 1 and rel[0]['jaccard'] >= .5:
            new['lineage_id'] = rel[0]['previous_lineage']; new['lineage_status'] = rel[0]['relation']
        else:
            new['lineage_id'] = 'L%05d' % next_id; next_id += 1
            new['lineage_status'] = 'birth' if not rel else 'uncertain_split_merge_or_drift'
    return links, events, next_id


class FlowShadow:
    def __init__(self, run_id, catalog=None, partitioner=infomap_partition, recent=100):
        self.run_id = run_id; self.catalog = catalog or EntryCatalog(); self.partitioner = partitioner
        self.recent = recent; self.max_seq = 0; self.records = {}; self.execution_ids = {}
        self.nodes = set(); self.weights = Counter(); self.action_weights = Counter(); self.triples = set()
        self.events = []; self.occurrences = []; self.snapshots = []; self.segment = 0; self.last = None
        self.next_lineage = 1; self.snapshot_count = 0; self.diagnostics = Counter(); self.boundary_pending = False

    def consume(self, row):
        e = deepcopy(row)
        if e.get('run_id') != self.run_id:
            raise ValueError('a flow graph cannot mix runs')
        seq = e['event_seq']; fingerprint = hashlib.sha256(json.dumps(e, sort_keys=True).encode()).hexdigest()
        if seq in self.records:
            if self.records[seq] != fingerprint:
                raise ValueError('conflicting duplicate event sequence')
            self.diagnostics['duplicate_event'] += 1; return False
        if seq <= self.max_seq:
            raise ValueError('out of order event; never silently use future/late evidence')
        if self.max_seq and seq != self.max_seq+1:
            self.boundary_pending = True; self.diagnostics['event_sequence_gap'] += 1
        self.max_seq = seq; self.records[seq] = fingerprint
        kind = e.get('event_kind')
        if kind in ('RESTART', 'RESET', 'TRACE_BOUNDARY', 'DATA_GAP') or e.get('is_restart') is True:
            self.boundary_pending = True; self.diagnostics['boundaries'] += 1; return False
        if kind != 'TRANSITION' or e.get('execution_confirmed') is not True:
            self.diagnostics['not_confirmed_transition'] += 1; return False
        if e.get('history_rebuild') is True:
            self.diagnostics['history_only'] += 1; self.boundary_pending = True; return False
        identity = e.get('transition_identity')
        if not identity or not e.get('source_state') or not e.get('target_state') or not e.get('model_action_identity'):
            self.boundary_pending = True; self.diagnostics['missing_identity'] += 1; return False
        if identity in self.execution_ids:
            prior = self.execution_ids[identity]
            if (prior['source_state'], prior['target_state'], prior['model_action_identity']) != (e['source_state'], e['target_state'], e['model_action_identity']):
                raise ValueError('conflicting execution identity')
            self.diagnostics['duplicate_execution'] += 1; return False
        gui_gap = bool(self.last and self.last.get('target_gui_id') is not None and e.get('source_gui_id') is not None and self.last['target_gui_id'] != e['source_gui_id'])
        if self.last and (gui_gap or self.boundary_pending or e.get('gap_before') or self.last['target_state'] != e['source_state'] or e.get('segment_id') != self.last.get('segment_id')):
            self.segment += 1
        self.boundary_pending = False; e['_segment'] = self.segment
        u, v = e['source_state'], e['target_state']; triple = (u, e['model_action_identity'], v)
        e['_new_state'] = v not in self.nodes; e['_new_pair'] = (u, v) not in self.weights
        e['_new_transition'] = triple not in self.triples
        self.nodes.update([u, v]); self.weights[u, v] += 1; self.action_weights[triple] += 1; self.triples.add(triple)
        self.execution_ids[identity] = e; self.events.append(e); self.last = e
        snap = self.snapshots[-1] if self.snapshots else None
        for g in self.catalog.match(e):
            community = next((r['id'] for r in snap['communities'] if v in r['members']), None) if snap else None
            self.occurrences.append(dict(group_id=g['group_id'], function=g['function'], event_seq=seq,
                transition_identity=identity, source_state=u, target_state=v, source_activity=e.get('source_activity'),
                model_action_identity=e['model_action_identity'], concrete_identity=e.get('concrete_identity'),
                label_provenance='ORACLE_REPLAY' if g.get('oracle') else 'REVIEWED_KNOWN_AT_EVENT',
                human_confirmed=g.get('human_confirmed', False), category='I', segment=self.segment,
                snapshot_at_execution=snap['id'] if snap else None, community_at_execution=community,
                community_status='KNOWN_FROZEN_TARGET' if community else 'NOT_YET_IN_SNAPSHOT'))
        return True

    def snapshot(self, reasons=()):
        started = time.perf_counter(); partitions, codelength = self.partitioner(set(self.nodes), dict(self.weights))
        infomap_ms = (time.perf_counter()-started)*1000
        number = self.snapshot_count+1; sid = 'S%04d' % number
        communities = [dict(id=sid+'-C%04d' % (i+1), members=sorted(mm)) for i, mm in enumerate(partitions)]
        flat = [s for r in communities for s in r['members']]
        assert len(flat) == len(set(flat)) and set(flat) == self.nodes, 'partition must cover observed graph exactly'
        previous = self.snapshots[-1]['communities'] if self.snapshots else []
        links, changes, self.next_lineage = lineage(previous, communities, self.next_lineage)
        state_to_region = {s:r['id'] for r in communities for s in r['members']}
        # Shared downstream facts use event sets, never sum overlapping entry-window budgets.
        associations = defaultdict(lambda: dict(direct=Counter(), windows=defaultdict(set), occurrences=0, functions=set()))
        windows = []; index = {e['event_seq']:i for i,e in enumerate(self.events)}
        function_sources = defaultdict(set)
        for o in self.occurrences:
            i = index[o['event_seq']]; e = self.events[i]; target = state_to_region[o['target_state']]
            a = associations[(o['group_id'], target)]; a['direct'][o['event_seq']] += 1; a['functions'].add(o['function'])
            function_sources[o['source_state']].add(o['function'])
            for k in (3, 5, 10):
                after = []
                for step in self.events[i+1:i+1+k]:
                    if step['_segment'] != e['_segment']:
                        break
                    after.append(step)
                counts = Counter(state_to_region[x['source_state']] for x in after)
                continuous = Counter(); last_r = None; streak = 0
                for x in after:
                    r = state_to_region[x['source_state']]; streak = streak+1 if r == last_r else 1; last_r = r
                    continuous[r] = max(continuous[r], streak)
                    associations[(o['group_id'], r)]['windows'][k].add(x['event_seq'])
                    associations[(o['group_id'], r)]['functions'].add(o['function'])
                windows.append(dict(group_id=o['group_id'], entry_event_seq=o['event_seq'], window=k,
                    observed_through_event_seq=after[-1]['event_seq'] if after else o['event_seq'],
                    complete=len(after)==k, segment_boundary=len(after)<k and i+1+len(after)<len(self.events),
                    events=[x['event_seq'] for x in after], community_steps=dict(counts), longest_continuous=dict(continuous)))
        hubs = {s for s, fs in function_sources.items() if len(fs)>=2}
        assoc_rows = []
        for (group, community), facts in sorted(associations.items()):
            assoc_rows.append(dict(group_id=group, community_id=community, functions=sorted(facts['functions']),
                direct_target_count=len(facts['direct']), direct_events=sorted(facts['direct']),
                trajectory_distinct_steps={str(k):len(v) for k,v in facts['windows'].items()},
                trajectory_step_sequences={str(k):sorted(v) for k,v in facts['windows'].items()},
                scope='same source Activity; reviewed occurrence only; not function purity'))
        for r in communities:
            r['metrics'] = region_metrics(self.events, r['members'], self.recent)
            r['entry_associations'] = [a for a in assoc_rows if a['community_id']==r['id']]
            r['functions'] = sorted({f for a in r['entry_associations'] for f in a['functions']})
            r['mixed_entry_functions'] = len(r['functions'])>1
            r['navigation_candidate_states'] = sorted(set(r['members']) & hubs)
            r['navigation_candidate_ratio'] = len(r['navigation_candidate_states'])/len(r['members'])
            old = next((p for p in previous if p['lineage_id']==r['lineage_id']), None)
            r['member_delta'] = dict(added=sorted(set(r['members'])-set(old['members']) if old else set(r['members'])),
                                    removed=sorted(set(old['members'])-set(r['members'])) if old else [])
        assert sum(r['metrics']['source_steps'] for r in communities) == len(self.events)
        out = dict(schema=1, run_id=self.run_id, id=sid, reasons=list(reasons), max_event_seq=self.max_seq,
            confirmed_count=len(self.events), nodes=len(self.nodes), state_pairs=len(self.weights),
            total_edge_weight=sum(self.weights.values()), edges=[dict(source=u,target=v,weight=w) for (u,v),w in sorted(self.weights.items())],
            action_edges=[dict(source=u,model_action=a,target=v,weight=w) for (u,a,v),w in sorted(self.action_weights.items())],
            parameters=PARAMETERS, codelength=codelength, infomap_ms=infomap_ms,
            communities=communities, lineage_links=links, lineage_events=changes,
            entry_occurrences=deepcopy(self.occurrences), trajectory_windows=windows,
            diagnostics=dict(self.diagnostics), pending_since_previous=len(self.events)-(self.snapshots[-1]['confirmed_count'] if self.snapshots else 0),
            interpretation='Current partition projected onto PAST evidence only; occurrence community_at_execution stays frozen; no sufficiency/suppress decision')
        out['total_compute_ms']=(time.perf_counter()-started)*1000
        # Full snapshots are persisted by the consumer. Only the last partition is needed in RAM.
        self.snapshot_count = number; self.snapshots[:] = [out]
        return out