"""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()