tegwick
bd85fcd179
Assistant: codex Assistant-Model: gpt-6.1-sol Assistant-Session: 01a1045b-933d-7aa1-b2db-104557487a79
176 lines
11 KiB
Python
176 lines
11 KiB
Python
#!/usr/bin/env python3
|
|
"""Filter a verified preview and prepare pinned consumer events; no inference."""
|
|
import argparse
|
|
from datetime import datetime, timezone
|
|
import hashlib
|
|
import json
|
|
import math
|
|
from pathlib import Path
|
|
import re
|
|
|
|
from reference_preview import ROOT, build, git_blob, release, require
|
|
from classification_coverage import section_values, RESPONSIBILITIES
|
|
|
|
HOST = 'https://forgejo.coulomb.social/coulomb/frontend-patterns'
|
|
SHA = re.compile(r'[0-9a-f]{40}')
|
|
DIGEST = re.compile(r'[0-9a-f]{64}')
|
|
IDENTITY = re.compile(r'[A-Za-z0-9][A-Za-z0-9:_.-]{0,159}')
|
|
|
|
|
|
def published(root=ROOT, version=None):
|
|
base, pin, manifest = release(root, version)
|
|
generated = build(root, version)
|
|
for name, content in generated.items():
|
|
require((root / base / name).read_text() == content, 'Published export changed: ' + name)
|
|
return json.loads(generated['index.json']), pin, manifest
|
|
|
|
|
|
def locator(path, revision):
|
|
return {'path': path, 'revision': revision,
|
|
'url': f'{HOST}/src/commit/{revision}/{path}'}
|
|
|
|
|
|
def lookup(index, pin, manifest, candidates=(), forces=(), responsibilities=(), root=ROOT):
|
|
known = {c['id'] for c in index['candidates']}
|
|
vocab = git_blob(root, manifest['corpus_revision'], 'canon/vocabulary.yaml').decode()
|
|
require(set(candidates) <= known, 'Unknown candidate in selected release')
|
|
require(set(forces) <= set(section_values(vocab, 'forces')), 'Unknown force')
|
|
require(set(responsibilities) <= set(RESPONSIBILITIES), 'Unknown responsibility')
|
|
matches = []
|
|
if candidates or forces or responsibilities:
|
|
for candidate in index['candidates']:
|
|
classification = candidate['classification']
|
|
if candidates and candidate['id'] not in candidates:
|
|
continue
|
|
if not set(forces) <= set(classification['forces']['improves']):
|
|
continue
|
|
if not set(responsibilities) <= set(classification['responsibilities']):
|
|
continue
|
|
record = dict(candidate)
|
|
record['contract'] = locator(candidate['path'], index['corpus_revision'])
|
|
record['evidence'] = [dict(e, locator=locator(e['ref'], pin['packaging_revision']))
|
|
for e in candidate['evidence']]
|
|
matches.append(record)
|
|
ids = {c['id'] for c in matches}
|
|
return {
|
|
'schema_version': 1,
|
|
'status': 'needs-context' if not (candidates or forces or responsibilities) else ('matches' if matches else 'no-match'),
|
|
'reference': {'version': index['version'], 'packaging_revision': pin['packaging_revision'],
|
|
'manifest_sha256': pin['manifest_sha256'], 'corpus_revision': index['corpus_revision']},
|
|
'filters': {'candidates': list(candidates), 'improving_forces': list(forces), 'responsibilities': list(responsibilities)},
|
|
'candidates': matches,
|
|
'relations': [e for e in index['relations'] if e['source'] in ids or e['target'] in ids],
|
|
'entrypoints': {k: locator(v, pin['packaging_revision']) for k, v in index['entrypoints'].items()},
|
|
'limits': ['Metadata filtering is not applicability, ranking or compatible-configuration proof.',
|
|
'Read full pinned contracts, force costs, alternatives and gaps before selecting.',
|
|
'Relations are conditional proposals; no inverse, transitive or automatic selection.',
|
|
'No match means no record satisfies these filters, not that no architecture exists.',
|
|
'This preview contains hypotheses; independent usefulness remains unverified.'],
|
|
}
|
|
|
|
|
|
def text_ok(value, limit=1000):
|
|
return isinstance(value, str) and 0 < len(value.strip()) <= limit
|
|
|
|
|
|
def receipt(source, index, pin, now=None):
|
|
require(isinstance(source, dict), 'Decision input must be an object')
|
|
fields = {'use_id', 'event_id', 'recorded_at', 'kind', 'consumer', 'context', 'decision', 'outcome'}
|
|
require(fields <= source.keys() <= fields | {'previous_event_sha256'}, 'Decision input fields mismatch')
|
|
event = json.loads(json.dumps(source))
|
|
event.update(schema_version=1, reference={
|
|
'version': index['version'], 'packaging_revision': pin['packaging_revision'],
|
|
'manifest_sha256': pin['manifest_sha256']})
|
|
for key in ('use_id', 'event_id'):
|
|
require(isinstance(event[key], str) and IDENTITY.fullmatch(event[key]), 'Invalid event identity')
|
|
require(event['kind'] in {'real', 'synthetic'}, 'Explicit real/synthetic kind required')
|
|
previous = event.setdefault('previous_event_sha256', None)
|
|
require(previous is None or isinstance(previous, str) and DIGEST.fullmatch(previous), 'Invalid predecessor digest')
|
|
require(isinstance(event['recorded_at'], str), 'Timestamp required')
|
|
stamp = datetime.fromisoformat(event['recorded_at'].replace('Z', '+00:00'))
|
|
require(stamp.tzinfo is not None and stamp <= (now or datetime.now(timezone.utc)), 'Invalid/future timestamp')
|
|
consumer = event['consumer']
|
|
require(isinstance(consumer, dict) and set(consumer) == {'repo', 'revision', 'task_ref', 'producer', 'run_ref'}, 'Consumer fields mismatch')
|
|
require(isinstance(consumer['repo'], str) and re.fullmatch(r'[a-z0-9][a-z0-9-]{0,79}', consumer['repo']), 'Invalid consumer repo')
|
|
require(isinstance(consumer['revision'], str) and SHA.fullmatch(consumer['revision']), 'Full consumer revision required')
|
|
require(all(text_ok(consumer[k]) for k in ('task_ref', 'producer', 'run_ref')), 'Missing consumer provenance')
|
|
context = event['context']
|
|
require(isinstance(context, dict) and set(context) == {'problem', 'constraints', 'considered', 'selected'}, 'Context fields mismatch')
|
|
require(text_ok(context['problem']), 'Problem required')
|
|
require(isinstance(context['constraints'], list) and all(text_ok(x, 500) for x in context['constraints']), 'Invalid constraints')
|
|
known = {c['id'] for c in index['candidates']}
|
|
for key in ('considered', 'selected'):
|
|
values = context[key]
|
|
require(isinstance(values, list) and all(isinstance(v, str) for v in values), 'Invalid candidate list')
|
|
require(len(values) == len(set(values)) and set(values) <= known, 'Unknown/duplicate candidate in release')
|
|
require(set(context['selected']) <= set(context['considered']), 'Unconsidered selection')
|
|
decision = event['decision']
|
|
require(isinstance(decision, dict) and set(decision) == {'action', 'reason', 'artifact_ref'}, 'Decision fields mismatch')
|
|
require(decision['action'] in {'applied', 'adapted', 'rejected', 'no-match'}, 'Invalid decision action')
|
|
require(bool(context['selected']) == (decision['action'] in {'applied', 'adapted'}), 'Action/selection mismatch')
|
|
require(text_ok(decision['reason']), 'Decision reason required')
|
|
require(decision['artifact_ref'] is None or text_ok(decision['artifact_ref']), 'Invalid artifact reference')
|
|
outcome = event['outcome']
|
|
defaults = {'value': 'unknown', 'basis': 'consumer-report', 'stage': 'decision',
|
|
'observation': None, 'evidence_refs': [], 'next_observation': None,
|
|
'measured_decision_minutes': None, 'measured_rework_minutes': None, 'comparison_ref': None}
|
|
require(isinstance(outcome, dict) and 'expectation' in outcome and set(outcome) <= set(defaults) | {'expectation'}, 'Outcome fields mismatch')
|
|
event['outcome'] = outcome = dict(defaults, **outcome)
|
|
require(text_ok(outcome['expectation']), 'Observable expectation required')
|
|
require(outcome['value'] in {'helpful', 'harmful', 'mixed', 'no-effect', 'unknown'}, 'Invalid outcome')
|
|
require(outcome['basis'] in {'consumer-report', 'model-assessment', 'observed-result'}, 'Invalid outcome basis')
|
|
require(outcome['stage'] in {'decision', 'implementation', 'test', 'operation'}, 'Invalid outcome stage')
|
|
refs = outcome['evidence_refs']
|
|
require(isinstance(refs, list) and len(refs) <= 10 and all(text_ok(v) for v in refs), 'Invalid outcome evidence')
|
|
require(outcome['observation'] is None or text_ok(outcome['observation']), 'Invalid observation')
|
|
require(outcome['value'] == 'unknown' or refs and text_ok(outcome['observation']), 'Observed outcome needs evidence')
|
|
for key in ('next_observation', 'comparison_ref'):
|
|
require(outcome[key] is None or text_ok(outcome[key]), 'Invalid outcome reference')
|
|
for key in ('measured_decision_minutes', 'measured_rework_minutes'):
|
|
value = outcome[key]
|
|
require(value is None or type(value) in (int, float) and math.isfinite(value) and value >= 0, 'Invalid measured effort')
|
|
return event
|
|
|
|
|
|
def write_receipt(event, relative, consumer_root=None):
|
|
require(isinstance(relative, str) and 0 < len(relative) <= 240
|
|
and re.fullmatch(r'[A-Za-z0-9_./-]+', relative)
|
|
and all(part not in {'', '.', '..'} for part in relative.split('/')), 'Unsafe receipt path')
|
|
root = (consumer_root or Path.cwd()).resolve()
|
|
path = root / relative
|
|
require(path.resolve().is_relative_to(root), 'Receipt escapes consumer root')
|
|
content = (json.dumps(event, indent=2, allow_nan=False) + '\n').encode()
|
|
require(len(content) <= 131072, 'Receipt exceeds collector size limit')
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
with path.open('xb') as stream:
|
|
stream.write(content)
|
|
return {'path': relative, 'sha256': hashlib.sha256(content).hexdigest()}
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument('--version', help='Published version; defaults to current pointer')
|
|
commands = parser.add_subparsers(dest='command', required=True)
|
|
query = commands.add_parser('lookup', help='Intersect force/responsibility filters; candidate IDs form an explicit allowlist')
|
|
query.add_argument('--candidate', action='append', default=[])
|
|
query.add_argument('--force', action='append', default=[])
|
|
query.add_argument('--responsibility', action='append', default=[])
|
|
prepare = commands.add_parser('receipt', help='Prepare a consumer-authored event with verified release pins')
|
|
prepare.add_argument('--input', required=True, type=Path, help='Decision/provenance JSON; see feedback/decision-input.example.json')
|
|
prepare.add_argument('--output', help='New source-relative receipt path under current consumer directory; never overwrite')
|
|
args = parser.parse_args()
|
|
try:
|
|
index, pin, manifest = published(version=args.version)
|
|
if args.command == 'lookup':
|
|
result = lookup(index, pin, manifest, args.candidate, args.force, args.responsibility)
|
|
else:
|
|
require(args.input.stat().st_size <= 131072, 'Decision input too large')
|
|
event = receipt(json.loads(args.input.read_text()), index, pin)
|
|
result = write_receipt(event, args.output) if args.output else event
|
|
print(json.dumps(result, indent=2, allow_nan=False))
|
|
except (ValueError, TypeError, KeyError, OSError) as exc:
|
|
parser.exit(1, f'Reference consumption failed: {exc}\n')
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|