#!/usr/bin/env python3
from __future__ import annotations
import json
import math
import re
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]

CONDITIONS = {
    'weekday','time_range','date_range','cron_trigger','stable_for','delay','state_duration','on_change','rate_limit','rolling_stat','oscillates',
    'outdoor_temperature','device_temperature','zone_temperature','ha_state','ha_numeric','ha_attribute',
    'ha_available','house_mode','device_state','zone_state','group_state','night_mode','constant','shared_input'
}
LOGIC = {'logic_and','logic_or','logic_not'}
ACTIONS = {'zone_thermostat','device_action','group_action','ha_service_action'}
ALL = CONDITIONS | LOGIC | ACTIONS

passed = 0

def check(value, message):
    global passed
    if not value:
        raise AssertionError(message)
    passed += 1


def cron_field_valid(field: str, lo: int, hi: int, allow_seven=False):
    upper = 7 if allow_seven else hi
    if not field.strip(): return False
    for part in field.split(','):
        part = part.strip()
        if part == '*': continue
        if part.startswith('*/'):
            try: step = int(part[2:])
            except ValueError: return False
            if not (0 < step <= hi - lo + 1): return False
            continue
        if '-' in part:
            try: a,b = map(int, part.split('-',1))
            except ValueError: return False
            if not (lo <= a <= b <= upper): return False
            continue
        try: v = int(part)
        except ValueError: return False
        if not (lo <= v <= upper): return False
    return True


def cron_valid(expr: str):
    f=expr.split()
    return len(f)==5 and cron_field_valid(f[0],0,59) and cron_field_valid(f[1],0,23) and cron_field_valid(f[2],1,31) and cron_field_valid(f[3],1,12) and cron_field_valid(f[4],0,6,True)


def cron_part_match(part, value, lo, hi):
    if part == '*': return True
    if part.startswith('*/'):
        step=int(part[2:]); return lo <= value <= hi and (value-lo)%step==0
    if '-' in part:
        a,b=map(int,part.split('-',1)); return a <= value <= b and a >= lo and b <= hi
    v=int(part); return v==value and lo <= v <= hi


def cron_field_match(field,value,lo,hi):
    return any(cron_part_match(p.strip(),value,lo,hi) for p in field.split(','))


def cron_matches(expr: str, dt: datetime):
    if not cron_valid(expr): return False
    f=expr.split(); # Python Monday=0; backend Sunday=0
    weekday=(dt.weekday()+1)%7
    return (cron_field_match(f[0],dt.minute,0,59) and cron_field_match(f[1],dt.hour,0,23)
            and cron_field_match(f[2],dt.day,1,31) and cron_field_match(f[3],dt.month,1,12)
            and (cron_field_match(f[4],weekday,0,7) or (weekday==0 and cron_field_match(f[4],7,0,7))))

@dataclass
class Runtime:
    since: datetime|None=None
    samples: list[tuple[datetime,float]]=field(default_factory=list)
    last_value: object|None=None

def timed_gate(rt: Runtime, input_value: bool, seconds: int, now: datetime):
    if not input_value:
        rt.since=None; return False
    if rt.since is None: rt.since=now
    return (now-rt.since).total_seconds() >= seconds

def state_duration(rt: Runtime, input_value: bool, minimum: int, maximum: int|None, now: datetime):
    if not input_value:
        rt.since=None; return False,0
    if rt.since is None: rt.since=now
    elapsed=max(0,int((now-rt.since).total_seconds()))
    return elapsed >= minimum and (maximum is None or elapsed <= maximum), elapsed

def changed(rt: Runtime, current):
    result=rt.last_value is not None and rt.last_value != current
    rt.last_value=current
    return result

def rate_limit_status(rt: Runtime, maximum: int, period: int, now: datetime):
    cutoff=now-timedelta(seconds=period)
    rt.samples=[x for x in rt.samples if x[0] >= cutoff]
    return len(rt.samples) < maximum, len(rt.samples)

def rate_limit_record(rt: Runtime, period: int, now: datetime):
    cutoff=now-timedelta(seconds=period)
    rt.samples=[x for x in rt.samples if x[0] >= cutoff]
    rt.samples.append((now,1.0))

def rolling(rt: Runtime, sample: float, window: int, statistic: str, now: datetime):
    cutoff=now-timedelta(seconds=window)
    rt.samples=[x for x in rt.samples if x[0] >= cutoff]
    rt.samples.append((now,sample))
    vals=[v for _,v in rt.samples]
    if statistic == 'median':
        vals=sorted(vals); mid=len(vals)//2
        return (vals[mid-1]+vals[mid])/2 if len(vals)%2==0 else vals[mid]
    return sum(vals)/len(vals)

def oscillation_metrics(values):
    if len(values)<3: return None
    span=max(values)-min(values)
    prev=0; changes=0
    for a,b in zip(values,values[1:]):
        d=b-a; sign=1 if d>1e-6 else -1 if d<-1e-6 else 0
        if not sign: continue
        if prev and sign != prev: changes += 1
        prev=sign
    return span,changes

def oscillates(rt: Runtime, sample: float, window: int, min_span: float, min_changes: int, now: datetime):
    cutoff=now-timedelta(seconds=window)
    rt.samples=[x for x in rt.samples if x[0] >= cutoff]
    rt.samples.append((now,sample))
    metrics=oscillation_metrics([v for _,v in rt.samples])
    return bool(metrics and metrics[0] >= min_span and metrics[1] >= min_changes)


NUMERIC_OPS = {'lt','lte','gt','gte','eq','neq'}
TEXT_OPS = {'eq','neq'}
DEVICE_STATE_FIELDS = {'enabled','online','power','mode','fan_speed','swing_vertical','swing_horizontal','quiet','turbo','light','air','xfan','health','sleep'}
ZONE_STATE_FIELDS = {'enabled','mode','active_preset','demand','control_owner','device_manual_override','local_thermostat_power'}
THERMOSTAT_PRESETS = {'auto','comfort','sleep','away','custom'}
THERMOSTAT_MODES = {'auto','heat','cool'}
GROUP_PRESETS = {'auto','comfort','sleep','away','custom'}
GROUP_MODES = {'auto','house','cool','heat'}
ROLLING_SOURCES = {'outdoor_temperature','device_temperature','zone_temperature','ha_numeric'}

def nonempty(value):
    return isinstance(value,str) and bool(value.strip())

def finite_number(value):
    return isinstance(value,(int,float)) and not isinstance(value,bool) and math.isfinite(float(value))

def optional_text(value):
    return value.strip() if isinstance(value,str) and value.strip() else None

def validate_numeric_comparison(config, name):
    check(config.get('operator','lt') in NUMERIC_OPS, f'{name}: unsupported numeric operator')
    check(finite_number(config.get('value')), f'{name}: numeric comparison needs finite value')

def validate_text_comparison_config(config, name):
    check(config.get('operator','eq') in TEXT_OPS, f'{name}: unsupported text operator')
    check('value' in config, f'{name}: state comparison needs value')

def validate_source_config(config, name):
    source=config.get('source')
    check(source in ROLLING_SOURCES, f'{name}: unsupported statistic source')
    if source == 'device_temperature': check(nonempty(config.get('device_id')), f'{name}: device source needs device_id')
    if source == 'zone_temperature': check(nonempty(config.get('zone_id')), f'{name}: zone source needs zone_id')
    if source == 'ha_numeric': check(nonempty(config.get('entity_id')), f'{name}: HA source needs entity_id')

def validate_node_config(node, name):
    kind=node['kind']; c=node.get('config',{})
    check(isinstance(c,dict), f'{name}/{node.get("id")}: config is not an object')
    tag=f'{name}/{node.get("id")}:{kind}'
    if kind == 'weekday':
        days=c.get('days'); check(isinstance(days,list) and bool(days), f'{tag}: weekdays missing')
        check(all(isinstance(d,int) and not isinstance(d,bool) and 1 <= d <= 7 for d in days), f'{tag}: invalid weekday')
    elif kind == 'time_range':
        for key in ('start','end'):
            try: datetime.strptime(c.get(key,''),'%H:%M')
            except (TypeError,ValueError): check(False, f'{tag}: invalid {key} time')
            else: check(True, f'{tag}: valid {key} time')
    elif kind == 'date_range':
        try:
            start=datetime.strptime(c.get('start',''),'%Y-%m-%d').date(); end=datetime.strptime(c.get('end',''),'%Y-%m-%d').date()
        except (TypeError,ValueError):
            check(False, f'{tag}: invalid date range')
        else:
            check(start <= end, f'{tag}: reversed date range')
    elif kind == 'cron_trigger':
        check(nonempty(c.get('expression')) and cron_valid(c['expression']), f'{tag}: invalid cron')
    elif kind in {'stable_for','delay'}:
        seconds=c.get('seconds'); check(isinstance(seconds,int) and not isinstance(seconds,bool) and 1 <= seconds <= 604800, f'{tag}: invalid duration')
    elif kind == 'state_duration':
        minimum=c.get('min_seconds',0); maximum=c.get('max_seconds')
        check(isinstance(minimum,int) and not isinstance(minimum,bool) and 0 <= minimum <= 604800, f'{tag}: invalid min duration')
        check(maximum is None or (isinstance(maximum,int) and not isinstance(maximum,bool) and minimum <= maximum <= 604800), f'{tag}: invalid max duration')
        check(minimum > 0 or maximum is not None, f'{tag}: empty duration range')
    elif kind == 'on_change':
        check(c.get('mode','result') in {'result','value'}, f'{tag}: invalid change mode')
    elif kind == 'rate_limit':
        count=c.get('max_count'); period=c.get('period_seconds')
        check(isinstance(count,int) and not isinstance(count,bool) and 1 <= count <= 1000, f'{tag}: invalid max count')
        check(isinstance(period,int) and not isinstance(period,bool) and 1 <= period <= 2678400, f'{tag}: invalid rate period')
    elif kind == 'rolling_stat':
        validate_source_config(c,tag)
        window=c.get('window_seconds'); check(isinstance(window,int) and not isinstance(window,bool) and 10 <= window <= 604800, f'{tag}: invalid window')
        check(c.get('statistic') in {'mean','median'}, f'{tag}: invalid statistic')
        validate_numeric_comparison(c,tag)
    elif kind == 'oscillates':
        validate_source_config(c,tag)
        window=c.get('window_seconds'); check(isinstance(window,int) and not isinstance(window,bool) and 10 <= window <= 604800, f'{tag}: invalid window')
        check(finite_number(c.get('min_span')) and float(c['min_span']) > 0, f'{tag}: min_span must be positive')
        changes=c.get('min_direction_changes'); check(isinstance(changes,int) and not isinstance(changes,bool) and 1 <= changes <= 1000, f'{tag}: invalid direction-change count')
    elif kind == 'outdoor_temperature':
        validate_numeric_comparison(c,tag)
    elif kind == 'device_temperature':
        check(nonempty(c.get('device_id')), f'{tag}: device_id missing'); validate_numeric_comparison(c,tag)
    elif kind == 'zone_temperature':
        check(nonempty(c.get('zone_id')), f'{tag}: zone_id missing'); validate_numeric_comparison(c,tag)
    elif kind == 'ha_state':
        check(nonempty(c.get('entity_id')), f'{tag}: entity_id missing'); validate_text_comparison_config(c,tag)
    elif kind == 'ha_numeric':
        check(nonempty(c.get('entity_id')), f'{tag}: entity_id missing'); validate_numeric_comparison(c,tag)
    elif kind == 'ha_attribute':
        check(nonempty(c.get('entity_id')), f'{tag}: entity_id missing')
        check(nonempty(c.get('attribute')), f'{tag}: attribute missing')
        check(c.get('operator','eq') in NUMERIC_OPS, f'{tag}: invalid attribute operator')
        check('value' in c, f'{tag}: attribute value missing')
    elif kind == 'ha_available':
        check(nonempty(c.get('entity_id')), f'{tag}: entity_id missing')
    elif kind == 'house_mode':
        check(c.get('value') in {'cool','heat','off'}, f'{tag}: invalid house mode'); validate_text_comparison_config(c,tag)
    elif kind == 'device_state':
        check(nonempty(c.get('device_id')), f'{tag}: device_id missing')
        check(c.get('field') in DEVICE_STATE_FIELDS, f'{tag}: invalid device field'); validate_text_comparison_config(c,tag)
    elif kind == 'zone_state':
        check(nonempty(c.get('zone_id')), f'{tag}: zone_id missing')
        check(c.get('field') in ZONE_STATE_FIELDS, f'{tag}: invalid zone field'); validate_text_comparison_config(c,tag)
    elif kind == 'group_state':
        check(nonempty(c.get('group_id')), f'{tag}: group_id missing')
        check(c.get('field') == 'power_enabled', f'{tag}: invalid group field'); validate_text_comparison_config(c,tag)
    elif kind == 'constant':
        check(isinstance(c.get('value'),bool), f'{tag}: constant needs boolean')
    elif kind == 'shared_input':
        check(nonempty(c.get('input_id')), f'{tag}: input_id missing')
    elif kind == 'zone_thermostat':
        check(nonempty(c.get('zone_id')), f'{tag}: zone_id missing')
        preset=optional_text(c.get('preset')) or 'comfort'; check(preset in THERMOSTAT_PRESETS, f'{tag}: invalid preset')
        mode=optional_text(c.get('mode')); check(mode is None or mode in THERMOSTAT_MODES, f'{tag}: invalid mode')
        if preset == 'custom': check(finite_number(c.get('setpoint')) and 8 <= float(c['setpoint']) <= 30, f'{tag}: invalid custom target')
        if 'power' in c and c['power'] is not None: check(isinstance(c['power'],bool), f'{tag}: power is not boolean')
        if 'cooldown_seconds' in c: check(isinstance(c['cooldown_seconds'],int) and c['cooldown_seconds'] >= 0, f'{tag}: invalid cooldown')
    elif kind == 'device_action':
        check(nonempty(c.get('device_id')), f'{tag}: device_id missing')
        fields=('power','mode','target_temperature','fan_speed','swing_vertical','swing_horizontal','quiet','turbo','light','air','xfan','health','sleep')
        check(any(c.get(field) is not None for field in fields), f'{tag}: empty device action')
    elif kind == 'group_action':
        check(nonempty(c.get('group_id')), f'{tag}: group_id missing')
        mode=optional_text(c.get('mode')); preset=optional_text(c.get('preset'))
        check(mode is None or mode in GROUP_MODES, f'{tag}: invalid group mode')
        check(preset is None or preset in GROUP_PRESETS, f'{tag}: invalid group preset')
        check(c.get('power') is not None or mode is not None or preset is not None, f'{tag}: empty group action')
        if preset == 'custom': check(finite_number(c.get('setpoint')) and 8 <= float(c['setpoint']) <= 30, f'{tag}: invalid custom target')
    elif kind == 'ha_service_action':
        check(nonempty(c.get('domain')), f'{tag}: domain missing'); check(nonempty(c.get('service')), f'{tag}: service missing')
        # Empty entity_id is intentionally treated as omitted by backend flow_string().
        check(isinstance(c.get('data',{}),dict), f'{tag}: HA service data must be object')
    elif kind in LOGIC or kind == 'night_mode':
        pass

def flow_action_truth_table(flow, action_id):
    by={n['id']:n for n in flow.get('nodes',[])}
    incoming={}
    for edge in flow.get('edges',[]): incoming.setdefault(edge['to'],[]).append(edge['from'])
    ancestors=set(); stack=list(incoming.get(action_id,[]))
    while stack:
        node_id=stack.pop()
        if node_id in ancestors: continue
        ancestors.add(node_id); stack.extend(incoming.get(node_id,[]))
    primitive=[node_id for node_id in ancestors if by[node_id]['kind'] not in LOGIC and by[node_id]['kind'] not in {'stable_for','delay','state_duration','on_change','rate_limit'}]
    check(len(primitive) <= 16, f'action {action_id}: truth table unexpectedly large')

    def eval_node(node_id, assigned, memo):
        if node_id in memo: return memo[node_id]
        node=by[node_id]; kind=node['kind']; ins=incoming.get(node_id,[])
        if kind == 'logic_and': value=bool(ins) and all(eval_node(x,assigned,memo) for x in ins)
        elif kind == 'logic_or': value=bool(ins) and any(eval_node(x,assigned,memo) for x in ins)
        elif kind == 'logic_not': value=len(ins)==1 and not eval_node(ins[0],assigned,memo)
        elif kind in {'stable_for','delay','state_duration','on_change','rate_limit'}: value=len(ins)==1 and eval_node(ins[0],assigned,memo)
        else:
            predecessors=all(eval_node(x,assigned,memo) for x in ins)
            value=predecessors and assigned.get(node_id,False)
        memo[node_id]=value; return value

    outcomes=[]
    for mask in range(1 << len(primitive)):
        assigned={node_id: bool(mask & (1 << index)) for index,node_id in enumerate(primitive)}
        memo={}
        outcomes.append(bool(incoming.get(action_id)) and all(eval_node(x,assigned,memo) for x in incoming.get(action_id,[])))
    return outcomes

def validate_graph(flow, name):
    nodes=flow.get('nodes',[]); edges=flow.get('edges',[])
    check(bool(nodes), f'{name}: no nodes')
    ids=[n.get('id') for n in nodes]
    check(all(ids) and len(ids)==len(set(ids)), f'{name}: duplicate/empty node ids')
    by={n['id']:n for n in nodes}
    check(all(n.get('kind') in ALL for n in nodes), f'{name}: unsupported node kind')
    check(any(n['kind'] in ACTIONS for n in nodes), f'{name}: no action')
    edge_ids=[e.get('id') for e in edges]
    check(all(edge_ids) and len(edge_ids)==len(set(edge_ids)), f'{name}: duplicate/empty edge ids')
    pairs=[]
    incoming={}
    outgoing={}
    for e in edges:
        a,b=e.get('from'),e.get('to')
        check(a in by and b in by and a != b, f'{name}: invalid edge endpoint')
        check(by[a]['kind'] not in ACTIONS, f'{name}: action has outgoing edge')
        pairs.append((a,b)); incoming.setdefault(b,[]).append(a); outgoing.setdefault(a,[]).append(b)
    check(len(pairs)==len(set(pairs)), f'{name}: duplicate connections')
    # DAG
    temp=set(); done=set()
    def visit(x):
        if x in done: return
        check(x not in temp, f'{name}: cycle')
        temp.add(x)
        for y in outgoing.get(x,[]): visit(y)
        temp.remove(x); done.add(x)
    for x in ids: visit(x)
    for n in nodes:
        validate_node_config(n, name)
        ins=incoming.get(n['id'],[])
        if n['kind']=='logic_not': check(len(ins)==1, f'{name}: NOT arity')
        if n['kind'] in {'logic_and','logic_or'}: check(len(ins)>=1, f'{name}: logic arity')
        if n['kind'] in {'stable_for','delay','state_duration','on_change','rate_limit'}: check(len(ins)==1, f'{name}: stateful gate arity')
        if n['kind'] in ACTIONS: check(len(ins)>=1, f'{name}: action without condition')
        if n['kind'] == 'rate_limit':
            targets=[by[target] for target in outgoing.get(n['id'],[]) if target in by]
            check(bool(targets) and all(target['kind'] in ACTIONS for target in targets), f'{name}: rate limit must be directly before action')
        if n['kind'] == 'on_change' and n.get('config',{}).get('mode','result') == 'value':
            sources=[by[source] for source in incoming.get(n['id'],[]) if source in by]
            blocked=LOGIC | {'stable_for','delay','state_duration','on_change','rate_limit','rolling_stat','oscillates'}
            check(len(sources)==1 and sources[0]['kind'] not in blocked, f'{name}: value-change mode needs direct source input')
    for action in (n for n in nodes if n['kind'] in ACTIONS):
        outcomes=flow_action_truth_table(flow, action['id'])
        check(any(outcomes), f'{name}/{action["id"]}: action condition graph can never become true')
    return len(nodes),len(edges)


def main():
    # Deterministic backend semantics.
    check(cron_valid('*/5 * * * *'), 'cron */5 invalid')
    check(cron_valid('0,15,30,45 6-18 * * 1-5'), 'cron list/range invalid')
    for bad in ('* * * *','*/0 * * * *','61 * * * *','* 24 * * *','* * 0 * *','* * * 13 *','* * * * 8','*/100 * * * *'):
        check(not cron_valid(bad), f'cron accepted invalid: {bad}')
    monday=datetime(2026,9,7,10,15)
    check(cron_matches('15 10 * * 1', monday), 'cron exact weekday mismatch')
    check(cron_matches('15 10 * * 1-7', monday), 'cron weekday range containing 7 mismatch')
    check(not cron_matches('16 10 * * 1', monday), 'cron false positive')
    sunday=datetime(2026,9,6,8,0)
    check(cron_matches('0 8 * * 0', sunday) and cron_matches('0 8 * * 7', sunday) and cron_matches('0 8 * * 1-7', sunday), 'cron Sunday 0/7/range mismatch')

    base=datetime(2026,9,2,12,0,0)
    rt=Runtime(); check(not timed_gate(rt,True,30,base),'stable fired immediately'); check(not timed_gate(rt,True,30,base+timedelta(seconds=29)),'stable fired early'); check(timed_gate(rt,True,30,base+timedelta(seconds=30)),'stable did not fire'); check(not timed_gate(rt,False,30,base+timedelta(seconds=31)),'stable did not reset'); check(not timed_gate(rt,True,30,base+timedelta(seconds=32)),'stable restart should wait')
    rt=Runtime(); check(not timed_gate(rt,True,3,base),'delay immediate'); check(timed_gate(rt,True,3,base+timedelta(seconds=3)),'delay did not pass after wait')

    rt=Runtime(); check(state_duration(rt,True,10,30,base)==(False,0),'state duration immediate'); check(state_duration(rt,True,10,30,base+timedelta(seconds=10))==(True,10),'state duration minimum'); check(state_duration(rt,True,10,30,base+timedelta(seconds=30))==(True,30),'state duration maximum boundary'); check(state_duration(rt,True,10,30,base+timedelta(seconds=31))==(False,31),'state duration exceeded maximum'); check(state_duration(rt,False,10,30,base+timedelta(seconds=32))==(False,0) and rt.since is None,'state duration reset')
    rt=Runtime(); check(not changed(rt,'off'),'change fired on first observation'); check(not changed(rt,'off'),'change fired without change'); check(changed(rt,'on'),'change not detected'); check(not changed(rt,'on'),'change repeated without new edge')
    rt=Runtime(); rate_limit_record(rt,60,base); rate_limit_record(rt,60,base+timedelta(seconds=10)); check(rate_limit_status(rt,2,60,base+timedelta(seconds=20))==(False,2),'rate limit did not block'); check(rate_limit_status(rt,2,60,base+timedelta(seconds=61))==(True,1),'rate limit did not prune rolling window')

    synthetic={
        'nodes':[
            {'id':'source','kind':'constant','config':{'value':True}},
            {'id':'duration','kind':'state_duration','config':{'min_seconds':10,'max_seconds':60}},
            {'id':'change','kind':'on_change','config':{'mode':'result'}},
            {'id':'limit','kind':'rate_limit','config':{'max_count':2,'period_seconds':3600}},
            {'id':'action','kind':'ha_service_action','config':{'domain':'switch','service':'turn_on','entity_id':'switch.test','data':{}}},
        ],
        'edges':[
            {'id':'e1','from':'source','to':'duration'}, {'id':'e2','from':'duration','to':'change'},
            {'id':'e3','from':'change','to':'limit'}, {'id':'e4','from':'limit','to':'action'},
        ]
    }
    check(validate_graph(synthetic,'synthetic-new-stateful-flow')==(5,4),'new stateful flow graph validation failed')

    rt=Runtime(); check(math.isclose(rolling(rt,10,60,'mean',base),10),'mean one'); check(math.isclose(rolling(rt,20,60,'mean',base+timedelta(seconds=10)),15),'mean two'); check(math.isclose(rolling(rt,30,60,'median',base+timedelta(seconds=20)),20),'median odd'); check(math.isclose(rolling(rt,40,60,'median',base+timedelta(seconds=30)),25),'median even'); check(math.isclose(rolling(rt,100,60,'mean',base+timedelta(seconds=100)),100),'window prune')
    rt=Runtime();
    for i,v in enumerate((20,22,19,23,20)):
        result=oscillates(rt,v,300,3,2,base+timedelta(seconds=i*10))
    check(result,'oscillation not detected')
    rt=Runtime();
    for i,v in enumerate((20,21,22,23,24)):
        result=oscillates(rt,v,300,3,2,base+timedelta(seconds=i*10))
    check(not result,'monotonic trend detected as oscillation')

    # Static backend/frontend wiring checks.
    engine=(ROOT/'src/engine/automations.rs').read_text()
    api=(ROOT/'src/api/flows.rs').read_text()
    js=(ROOT/'web/js/flows.js').read_text()
    html=(ROOT/'web/index.html').read_text()
    for kind in ('cron_trigger','stable_for','delay','state_duration','on_change','rate_limit','rolling_stat','oscillates'):
        check(f'"{kind}"' in api and f'"{kind}"' in engine, f'{kind}: backend wiring missing')
        check(kind in js and f'data-flow-add="{kind}"' in html, f'{kind}: UI wiring missing')
    meta_block=js.split('const FLOW_NODE_META = Object.freeze({',1)[1].split('});',1)[0]
    meta_kinds=set(re.findall(r'^\s*([a-z_]+):\s*\{', meta_block, re.M))
    palette_kinds=set(re.findall(r'data-flow-add="([a-z_]+)"', html))
    check(meta_kinds == ALL, f'UI metadata/backend kind mismatch: missing={ALL-meta_kinds}, extra={meta_kinds-ALL}')
    check(palette_kinds == ALL, f'palette/backend kind mismatch: missing={ALL-palette_kinds}, extra={palette_kinds-ALL}')
    check('"ha_service_action" => Ok(None)' in api, 'HA action missing dry-run support')
    check('home_assistant::call_service' in engine and 'action_ha_domain' in engine, 'HA action runtime missing')
    check('format!("ha:{entity_id}")' in engine, 'HA target conflict claim missing')
    check('Some(&mut runtime)' in api, 'dry-run does not use stateful runtime')
    check('conditions.iter().any(|condition| condition.kind == "cron_trigger")' in engine and 'last.minute() == now.minute()' in engine, 'CRON same-minute duplicate guard missing')
    check('previous.config == current.config' in api and 'runtime.retain' in api, 'runtime reset-on-edit protection missing')
    check('flow_record_rate_limited_execution' in engine and 'Ok(true)' in engine, 'rate limit is not recorded after successful execution')
    check('if !item.enabled { continue; }' in engine and 'if !ready || !should_fire { continue; }' in engine, 'Flow runtime may be skipped during action cooldown')
    check('rate-limit block must be placed directly before an action' in api, 'rate-limit placement guard missing')
    check('on-change value mode needs one direct source/condition input' in api, 'on-change value-source guard missing')
    models=(ROOT/'src/models/flow.rs').read_text()
    core=(ROOT/'web/js/core.js').read_text()
    settings_api=(ROOT/'src/api/settings.rs').read_text()
    check('pub draft: bool' in models and '#[serde(default)]' in models, 'Flow draft persistence field missing')
    check('fn validate_flow_draft_graph' in api, 'draft structural validator missing')
    check(api.count('if input.draft { validate_flow_draft_graph(&input)?; } else { validate_flow_graph(&input)?; }') >= 3, 'draft create/update/import validation split missing')
    check('enabled: if draft { false } else { input.enabled }' in api, 'draft does not force Flow disabled')
    check(api.count('(prepare_draft_flow(flow), vec![], vec![])') >= 3, 'draft save/import may compile executable outputs')
    check('flow.compiled_schedule_ids.clear()' in api and 'flow.compiled_automation_ids.clear()' in api, 'draft output clearing missing')
    check('"draft": flow.draft' in api, 'draft state missing from Flow export')
    check('import contains an executable Flow draft' in settings_api, 'configuration import does not enforce draft safety invariant')
    check('error.status = response.status' in core, 'frontend API errors do not expose HTTP validation status')
    check("error.status !== 400" in js and "draft:true" in js and "enabled:false" in js, 'save-as-draft retry flow missing')
    check("if (flow.draft) return toast(tr('flow.draftCannotEnable')" in js, 'draft quick-enable guard missing')
    check("draft: app.flowDraft.draft === true" in js, 'draft portability state missing from source payload')
    runtime_leaf=engine.split('async fn flow_leaf_observation',1)[1].split('pub async fn evaluate_flow_conditions_trace',1)[0]
    for kind in CONDITIONS - {'stable_for','delay','state_duration','on_change','rate_limit','rolling_stat','oscillates'}:
        check(f'\"{kind}\" =>' in runtime_leaf or (kind == 'shared_input' and 'condition.kind == \"shared_input\"' in runtime_leaf), f'{kind}: runtime leaf implementation missing')
    evaluator=engine.split('pub async fn evaluate_flow_conditions_trace',1)[1].split('async fn flow_conditions_match',1)[0]
    for kind in ('stable_for','delay','state_duration','on_change','rate_limit','rolling_stat','oscillates','logic_and','logic_or','logic_not'):
        check(f'\"{kind}\"' in evaluator, f'{kind}: evaluator implementation missing')
    for category in ('trigger','time','timeop','sensor','logic','action','haaction'):
        check(f'flow-palette-{category}' in html, f'palette category {category} missing')
    css=(ROOT/'web/styles.css').read_text()
    for marker in ('.flow-palette-group button', 'min-height:30px', 'padding:6px 8px', '.flow-palette-timeop', '.flow-palette-haaction', '.flow-editor-title input:hover', '.flow-editor-title input:focus', '.flow-draft-badge'):
        check(marker in css, f'compact/color Flow CSS missing: {marker}')

    # Translation pack correctness: all new keys must be inside translations, not root.
    for lang in ('pl','en'):
        pack=json.loads((ROOT/f'lang/{lang}.json').read_text())
        check(not any(k.startswith('flow.') for k in pack), f'{lang}: flow translations leaked to root')
        for key in ('flow.triggers','flow.timeOps','flow.haActions','flow.node.cronTrigger','flow.node.stableFor','flow.node.stateDuration','flow.node.onChange','flow.node.rateLimit','flow.node.delay','flow.node.rollingStat','flow.node.oscillates','flow.node.haServiceAction','flow.draft','flow.draftStatus','flow.draftNoExecution','flow.draftDisabledHint','flow.draftCannotEnable','flow.saveAsDraftConfirm','flow.savedAsDraft'):
            check(bool(pack.get('translations',{}).get(key)), f'{lang}: missing {key}')
        title_keys=set(re.findall(r"titleKey: '([^']+)'", meta_block))
        missing_titles=sorted(key for key in title_keys if not pack.get('translations',{}).get(key))
        check(not missing_titles, f'{lang}: missing Flow node title translations: {missing_titles}')
        referenced_flow_keys=set(re.findall(r'''(?:data-i18n(?:-[a-z]+)?=|tr\()\s*[\"'](flow\.[A-Za-z0-9_.]+)[\"']''', js + '\n' + html))
        missing_referenced=sorted(key for key in referenced_flow_keys if key not in pack.get('translations',{}))
        check(not missing_referenced, f'{lang}: missing referenced Flow translations: {missing_referenced}')

    # Regression pass over every bundled Flow preset.
    preset_files=sorted((ROOT/'presets').glob('*.json'))
    check(len(preset_files) >= 30, 'unexpectedly small preset library')
    nodes=edges=0
    for f in preset_files:
        data=json.loads(f.read_text())
        n,e=validate_graph(data['flow'],f.name); nodes+=n; edges+=e
    print(f'PASS: {passed} logical assertions; {len(preset_files)} presets; {nodes} nodes; {edges} edges')

if __name__=='__main__': main()