validate_tool_packages.py

Python · 103 lines · 5,777 bytes · wiki path 10 Architecture/contracts/validate_tool_packages.py · download the raw file · cited from Contracts

Same folder: Connectors.Abstractions.cs · configuration-change.schema.json · configuration-input.schema.json · connector-capability.schema.json · eval-set.schema.json · execution-obligations.schema.json · ledger-record.schema.json · message-envelope.schema.json · model-turn.schema.json · module-manifest.schema.json · openapi.yaml · realtime-events.schema.json · runtime-state.schema.json · tool-package.schema.json · validate_case_journeys.py · validate_contracts.py · validate_runtime_contracts.py · write-shape.schema.json

"""Tool/skill contract examples only; no package download, execution or benchmark."""
from pathlib import Path
import copy
import json
from jsonschema import Draft202012Validator
from referencing import Registry
from referencing.jsonschema import DRAFT202012

HERE = Path(__file__).resolve().parent
PACKAGE = json.loads((HERE/'tool-package.schema.json').read_text())
SKILL = json.loads((HERE/'drafts/skill-manifest.schema.json').read_text())
REGISTRY = Registry().with_resource(PACKAGE['$id'],DRAFT202012.create_resource(PACKAGE))


def package_errors(package):
    errors=[e.message for e in Draft202012Validator(PACKAGE).iter_errors(package)]
    if errors:
        return errors
    for key in ('input_schema','output_schema'):
        try:
            Draft202012Validator.check_schema(package[key])
        except Exception as e:
            errors.append(key+': '+str(e))
    keys=[(d['key'],d['version']) for d in package['dependencies']]
    if len(keys)!=len(set(keys)):
        errors.append('duplicate dependency')
    return errors


def skill_errors(skill):
    errors=[e.message for e in Draft202012Validator(SKILL,registry=REGISTRY).iter_errors(skill)]
    if errors:
        return errors
    packages=skill.get('tool_bindings',[])
    identities=[(p['tool_key'],p['version']) for p in packages]
    if len(identities)!=len(set(identities)):
        errors.append('duplicate tool binding')
    operations={o['operation_id'] for o in skill['operations']}
    for package in packages:
        errors+=package_errors(package)
        if not set(package['connector_operations'])<=operations:
            errors.append('undeclared connector operation')
    if not operations and not packages:
        errors.append('skill has no executable capability')
    entry=skill.get('entry_tool')
    if entry and (entry['tool_key'],entry['version']) not in identities:
        errors.append('entry tool is not bound')
    if any(o['mode'] in ('write','send','ddl') for o in skill['operations']) and not skill['writes']:
        errors.append('effect hidden by writes=false')
    return errors


def main():
    data=json.loads((HERE/'fixtures/tool-package.examples.json').read_text())
    errors=[];count=0
    def check(name,ok):
        nonlocal count
        count+=1
        if not ok:errors.append(name)
    Draft202012Validator.check_schema(PACKAGE)
    for ex in data['examples']:
        found=package_errors(ex['package'])
        check(ex['name'],bool(found)==(ex['expected']=='invalid'))
    p=data['examples'][0]['package']
    skill={'skill_key':'support.family_check','version':1,'module':'support','inputs':p['input_schema'],
           'operations':[],'tool_bindings':[p],'entry_tool':{'tool_key':p['tool_key'],'version':p['version']},
           'assertions':[{'name':'complete family coverage','expect':'Every requested member is checked; unsupported rules are explicit.'}], 'writes':False}
    check('pure local tool needs no fictional connector',not skill_errors(skill))
    changed=copy.deepcopy(skill);changed['entry_tool']['version']=99
    check('entry cannot drift from pinned package','entry tool is not bound' in skill_errors(changed))
    changed=copy.deepcopy(skill);changed['tool_bindings'].append(copy.deepcopy(p))
    check('duplicate binding refused','duplicate tool binding' in skill_errors(changed))
    changed=copy.deepcopy(skill);changed['tool_bindings'][0]['effect']='connector';changed['tool_bindings'][0]['connector_operations']=['oracle.read']
    check('package cannot smuggle an operation','undeclared connector operation' in skill_errors(changed))
    changed=copy.deepcopy(skill);changed['operations']=[{'operation_id':'mail.send','connector':'mail','mode':'send'}]
    check('no hidden send under pure skill','effect hidden by writes=false' in skill_errors(changed))
    call={'skill_key':skill['skill_key'],'skill_version':1,'tool_key':p['tool_key'],'tool_version':1,'input':{'members':[],'rules':[]}}
    inv=PACKAGE['$defs']['invocation']
    check('invocation names the pinned binding',not list(Draft202012Validator(inv).iter_errors(call)))
    check('invocation cannot choose an arbitrary executable',bool(list(Draft202012Validator(inv).iter_errors({**call,'executable':'elsewhere.exe'}))))
    check('tool input must meet its contract',bool(list(Draft202012Validator(p['input_schema']).iter_errors({}))))
    check('unsupported is a typed output',not list(Draft202012Validator(p['output_schema']).iter_errors({'outcome':'unsupported','checks':[]})))
    check('success cannot be invented as another output kind',bool(list(Draft202012Validator(p['output_schema']).iter_errors({'outcome':'probably fine','checks':[]}))))
    check('resolved direct tool invocation',not invocation_errors(skill,call))
    check('no unbound tool version','tool binding not found' in invocation_errors(skill,{**call,'tool_version':99}))
    check('no changed skill version','skill binding mismatch' in invocation_errors(skill,{**call,'skill_version':99}))
    print(f'{count} tool/skill contract checks; {len(errors)} failures')
    for error in errors:print('FAIL',error)
    return bool(errors)


def invocation_errors(skill,call):
    errors=[e.message for e in Draft202012Validator(PACKAGE['$defs']['invocation']).iter_errors(call)]
    if errors:return errors
    if (call['skill_key'],call['skill_version'])!=(skill['skill_key'],skill['version']):
        return ['skill binding mismatch']
    tool=next((p for p in skill.get('tool_bindings',[]) if (p['tool_key'],p['version'])==(call['tool_key'],call['tool_version'])),None)
    if tool is None:return ['tool binding not found']
    return [e.message for e in Draft202012Validator(tool['input_schema']).iter_errors(call['input'])]


if __name__=='__main__':
    raise SystemExit(main())