MateClaw 2.3.0：用不可变 JSON 为长任务建立可验证的完成门槛

from __future__ import annotations

import hashlib
import json
from pathlib import Path

CURRENT_GENERATION = 7
CURRENT_REQUIREMENTS = {
    'report-produced': 'file_sha256',
    'tests-passed': 'check_passed',
}


def sha256_file(path: str) -> str:
    digest = hashlib.sha256()
    with open(path, 'rb') as stream:
        for chunk in iter(lambda: stream.read(65536), b''):
            digest.update(chunk)
    return digest.hexdigest()


def validate_acceptance(document: dict) -> list[str]:
    errors: list[str] = []

    if document.get('generation') != CURRENT_GENERATION:
        errors.append('acceptance document belongs to a stale generation')

    artifact = document.get('artifact', {})
    if artifact.get('immutable') is not True:
        errors.append('artifact is not marked immutable')

    bindings = document.get('bindings', [])
    indexed = {item.get('requirement_id'): item for item in bindings}

    if set(indexed) != set(CURRENT_REQUIREMENTS):
        errors.append('bindings do not exactly match current requirements')
        return errors

    for requirement_id, check_type in CURRENT_REQUIREMENTS.items():
        binding = indexed[requirement_id]

        if binding.get('generation') != CURRENT_GENERATION:
            errors.append(f'{requirement_id}: stale generation binding')
            continue

        if binding.get('valid') is not True:
            errors.append(f'{requirement_id}: binding is not valid')
            continue

        evidence = binding.get('evidence', {})
        if check_type == 'file_sha256':
            path = evidence.get('path')
            expected = evidence.get('sha256')
            if not path or not Path(path).is_file():
                errors.append(f'{requirement_id}: evidence file is missing')
            elif sha256_file(path) != expected:
                errors.append(f'{requirement_id}: artifact hash changed')

        if check_type == 'check_passed':
            if evidence.get('status') != 'passed':
                errors.append(f'{requirement_id}: check did not pass')

    return errors


def main() -> None:
    Path('report.txt').write_text(
        'Persistent Goal delivery for generation 7\n',
        encoding='utf-8',
    )

    document = {
        'goal_id': 'goal-quarterly-report',
        'generation': 7,
        'artifact': {
            'version': 'artifact-v7',
            'immutable': True,
        },
        'bindings': [
            {
                'requirement_id': 'report-produced',
                'generation': 7,
                'valid': True,
                'evidence': {
                    'path': 'report.txt',
                    'sha256': sha256_file('report.txt'),
                },
            },
            {
                'requirement_id': 'tests-passed',
                'generation': 7,
                'valid': True,
                'evidence': {
                    'status': 'passed',
                    'command': 'python -m unittest',
                },
            },
        ],
    }

    errors = validate_acceptance(document)
    print(json.dumps(document, ensure_ascii=False, indent=2))

    if errors:
        print('\nREJECTED')
        for error in errors:
            print(f'- {error}')
        raise SystemExit(1)

    print('\nACCEPTED: every current requirement is valid')


if __name__ == '__main__':
    main()

python acceptance_gate.py