Version the data and feature contract

"""Bind dataset/schema/feature identities before calling the existing feature store.

Local trusted workspace only: hashes detect changes against an expected identity;
they do not authenticate the author or prove quality of data or feature code.
"""
import hashlib
import json
from pathlib import Path
import sqlite3
import sys

HERE = Path(__file__).resolve().parent
sys.path.insert(0, str(HERE / 'feature_history'))
from history import History, Feature, source_rows, minute

INPUTS = ('dataset.json', 'schema.json', 'feature.json')
CODE = ('versions.py', 'feature_history/history.py', 'feature_history/join.sql',
        'feature_history/vendor/availability.py', 'feature_history/vendor/feature_math.py')
SCHEMA = {'name': 'user-event-schema', 'version': 1,
          'fields': ['id', 'entity', 'event_at', 'arrived_at'],
          'time_zone': 'UTC', 'time_resolution': 'minute'}


def digest(raw):
    return hashlib.sha256(raw).hexdigest()


def encoded(value):
    return json.dumps(value, sort_keys=True, separators=(',', ':'), allow_nan=False).encode()


def unique_pairs(pairs):
    result = {}
    for key, value in pairs:
        if key in result:
            raise ValueError('duplicate JSON object key')
        result[key] = value
    return result


def parse(raw):
    def reject(value):
        raise ValueError('nonstandard JSON constant: ' + value)
    return json.loads(raw, object_pairs_hook=unique_pairs, parse_constant=reject)


def identity(value):
    if (not isinstance(value, dict) or not isinstance(value.get('name'), str)
            or not 1 <= len(value['name']) <= 100
            or type(value.get('version')) is not int or value['version'] < 1):
        raise ValueError('identity needs a bounded name and positive integer version')
    return value['name'], value['version']


def inputs(root):
    raw = {name: (Path(root) / name).read_bytes() for name in INPUTS}
    data, schema, spec = (parse(raw[name]) for name in INPUTS)
    for value in (data, schema, spec):
        identity(value)
    if schema != SCHEMA:
        raise ValueError('unsupported schema contract; changing its version does not add a validator')
    if set(data) != {'name', 'version', 'events'} or not isinstance(data['events'], list):
        raise ValueError('dataset needs identity and an event list')
    source_rows(data['events'])  # Reuse the existing event/UTC/minute validation.
    required = {'name', 'version', 'entity', 'window_days', 'aggregation', 'availability'}
    if (set(spec) != required or spec['entity'] != 'user'
            or type(spec['window_days']) is not int or spec['window_days'] < 1
            or spec['aggregation'] not in ('count', 'distinct_days')
            or spec['availability'] not in ('as_known', 'corrected')):
        raise ValueError('unsupported feature definition')
    return raw, data, schema, spec


def capture(root):
    raw, data, schema, spec = inputs(root)
    code = {name: digest((HERE / name).read_bytes()) for name in CODE}
    feature_digest = digest(encoded({'definition': spec, 'implementation': code}))
    manifest = {'format': 1, 'files': {name: digest(blob) for name, blob in raw.items()},
            'code': code,
            'components': [
                {'kind': 'dataset', 'name': data['name'], 'version': data['version'],
                 'digest': digest(raw['dataset.json'])},
                {'kind': 'schema', 'name': schema['name'], 'version': schema['version'],
                 'digest': digest(encoded({'schema': schema, 'validator': code['feature_history/vendor/availability.py']}))},
                {'kind': 'feature', 'name': spec['name'], 'version': spec['version'],
                 'digest': feature_digest}]}
    return manifest, (data, schema, spec)


def snapshot(root):
    return capture(root)[0]


def manifest_id(manifest):
    return digest(encoded(manifest))


class Registry:
    def __init__(self, path=':memory:'):
        self.db = sqlite3.connect(path)
        self.db.execute('''CREATE TABLE IF NOT EXISTS identities (
            kind TEXT, name TEXT, version INTEGER, digest TEXT NOT NULL,
            PRIMARY KEY(kind,name,version))''')

    def close(self):
        self.db.close()

    def bind(self, manifest):
        # All component bindings succeed together, or none of the new ones remain.
        with self.db:
            for item in manifest['components']:
                key = (item['kind'], item['name'], item['version'])
                old = self.db.execute('SELECT digest FROM identities WHERE kind=? AND name=? AND version=?', key).fetchone()
                if old and old[0] != item['digest']:
                    raise ValueError('identity already bound: ' + '/'.join(map(str, key)))
                self.db.execute('INSERT OR IGNORE INTO identities VALUES (?,?,?,?)', (*key, item['digest']))


def run_locked(root, manifest, expected_id, registry, cutoff, published_at):
    if manifest_id(manifest) != expected_id:
        raise ValueError('manifest differs from the expected identity')
    current, (data, _, spec) = capture(root)
    if current != manifest:
        raise ValueError('workspace differs from the locked manifest')
    # Use the parsed bytes that were hashed, without reopening mutable input files.
    registry.bind(manifest)
    feature = Feature(**{key: spec[key] for key in ('name', 'version', 'entity', 'window_days', 'aggregation')})
    store = History()
    try:
        key = store.materialize(data['events'], cutoff, published_at, spec['availability'], feature)
        return {'manifest_id': expected_id, 'feature': key,
                'user_02': store.point_in_time(key, 'user-02', published_at)}
    finally:
        store.close()


def main():
    import argparse
    parser = argparse.ArgumentParser()
    commands = parser.add_subparsers(dest='command', required=True)
    lock = commands.add_parser('lock')
    lock.add_argument('--inputs', type=Path, default=HERE)
    lock.add_argument('--out', type=Path, required=True)
    run = commands.add_parser('run')
    run.add_argument('--inputs', type=Path, default=HERE)
    run.add_argument('--manifest', type=Path, required=True)
    run.add_argument('--expected-id', required=True)
    run.add_argument('--registry', type=Path, required=True)
    run.add_argument('--cutoff', default='2026-01-08T12:00:00Z')
    run.add_argument('--published-at', default='2026-01-08T12:30:00Z')
    args = parser.parse_args()
    if args.command == 'lock':
        manifest = snapshot(args.inputs)
        with args.out.open('x') as file:
            file.write(json.dumps(manifest, indent=2) + '\n')
        print(manifest_id(manifest))
    else:
        manifest = parse(args.manifest.read_bytes())
        # Do not create/open a registry until the expectation and bytes match.
        if manifest_id(manifest) != args.expected_id or snapshot(args.inputs) != manifest:
            raise SystemExit('locked identity or workspace differs; computation refused')
        registry = Registry(args.registry)
        try:
            result = run_locked(args.inputs, manifest, args.expected_id, registry,
                                minute(args.cutoff), minute(args.published_at))
        finally:
            registry.close()
        print(json.dumps(result, indent=2))


if __name__ == '__main__':
    main()

"""Run actual local version checks; rejected candidates never call materialize."""
import argparse
import json
from pathlib import Path
import shutil
import tempfile
from unittest.mock import patch
from versions import HERE, INPUTS, Registry, History, snapshot, manifest_id, run_locked, minute

CUTOFF = minute('2026-01-08T12:00:00Z')
PUBLISHED = minute('2026-01-08T12:30:00Z')


def record():
    registry = Registry()
    try:
        locked = snapshot(HERE)
        expected = manifest_id(locked)
        baseline = run_locked(HERE, locked, expected, registry, CUTOFF, PUBLISHED)
        probes = []
        for name in ('dataset', 'schema', 'feature'):
            with tempfile.TemporaryDirectory() as tmp:
                for file in INPUTS:
                    shutil.copyfile(HERE/file, Path(tmp)/file)
                target = Path(tmp)/(name+'.json')
                value = json.loads(target.read_text())
                if name == 'dataset':
                    value['events'][0]['arrived_at'] = '2026-01-07T16:00:00Z'
                elif name == 'schema':
                    value['version'] = 2
                else:
                    value['window_days'] = 2
                target.write_text(json.dumps(value, indent=2)+'\n')
                with patch.object(History, 'materialize') as materialize:
                    try:
                        run_locked(tmp, locked, expected, registry, CUTOFF, PUBLISHED)
                    except ValueError as error:
                        probes.append({'changed': name, 'refused': True,
                                       'reason': str(error), 'compute_calls': materialize.call_count})
                    else:
                        raise AssertionError('changed input was accepted')
        with tempfile.TemporaryDirectory() as tmp:
            for file in INPUTS:
                shutil.copyfile(HERE/file, Path(tmp)/file)
            path=Path(tmp)/'feature.json';spec=json.loads(path.read_text());spec['window_days']=2
            path.write_text(json.dumps(spec,indent=2)+'\n')
            renamed=snapshot(tmp)
            try:
                run_locked(tmp,renamed,manifest_id(renamed),registry,CUTOFF,PUBLISHED)
            except ValueError as error:
                conflict=str(error)
            else:
                raise AssertionError('same feature version was rebound')
            spec['version']=2;path.write_text(json.dumps(spec,indent=2)+'\n')
            next_manifest=snapshot(tmp)
            next_result=run_locked(tmp,next_manifest,manifest_id(next_manifest),registry,CUTOFF,PUBLISHED)
        return {'mode':'local version checks; synthetic events; no model training',
                'baseline':baseline, 'changed_input_probes':probes,
                'same_version_conflict':conflict,'explicit_feature_v2':next_result,
                'manifest':locked}
    finally:
        registry.close()


def main():
    parser=argparse.ArgumentParser();parser.add_argument('--check',action='store_true');parser.add_argument('--save',type=Path)
    args=parser.parse_args();result=record()
    if args.check and result!=json.loads((HERE/'result.json').read_text()):
        raise SystemExit('version checks differ from bundled result')
    if args.save:args.save.write_text(json.dumps(result,indent=2)+'\n')
    print('Dataset v1 + schema v1 + feature v1: user-02 count =',result['baseline']['user_02']['value'])
    for row in result['changed_input_probes']:
        print(f"Changed {row['changed']}: refused={row['refused']}; compute_calls={row['compute_calls']}")
    print('New manifest with reused feature v1: refused; immutable identity conflict')
    print('Explicit feature v2 (two-day window): user-02 count =',result['explicit_feature_v2']['user_02']['value'])
    print('Manifest identities differ:',result['baseline']['manifest_id']!=result['explicit_feature_v2']['manifest_id'])
    print('Hashes bind inputs and implementation; they do not prove data quality or authorship.')


if __name__=='__main__':main()