用五引擎微内核守住证据边界：开放架构与持续迭代的工程方法

输入：命令类型 + 协议版本 + 调用者上下文 + 业务载荷
输出：结果状态 + 结构化结果 + 可公开元数据 + 错误信息
副作用：通过内核提供的端口访问存储、消息和证据服务

from __future__ import annotations

import hashlib
import json
import time
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Callable

Engine = Callable[[dict[str, Any]], dict[str, Any]]


@dataclass(frozen=True)
class Command:
    engine: str
    actor: str
    payload: dict[str, Any]
    protocol: str = 'v1'


class EvidenceLog:
    def __init__(self, path: str = 'evidence.jsonl') -> None:
        self.path = Path(path)

    def _last_hash(self) -> str:
        if not self.path.exists():
            return '0' * 64
        lines = self.path.read_text(encoding='utf-8').splitlines()
        return json.loads(lines[-1])['chain_hash'] if lines else '0' * 64

    def append(self, command: Command, result: dict[str, Any]) -> dict[str, str]:
        # 示例只保存业务内容摘要，不把原始载荷写入证据日志。
        canonical = json.dumps(
            {'command': command.payload, 'result': result},
            ensure_ascii=False,
            sort_keys=True,
            separators=(',', ':'),
        ).encode('utf-8')
        content_hash = hashlib.sha256(canonical).hexdigest()
        previous_hash = self._last_hash()
        chain_hash = hashlib.sha256(
            f'{previous_hash}:{command.engine}:{content_hash}'.encode('utf-8')
        ).hexdigest()
        record = {
            'time': int(time.time()),
            'engine': command.engine,
            'actor': command.actor,
            'protocol': command.protocol,
            'content_hash': content_hash,
            'previous_hash': previous_hash,
            'chain_hash': chain_hash,
        }
        with self.path.open('a', encoding='utf-8') as file:
            file.write(json.dumps(record, ensure_ascii=False) + '\n')
        return {'content_hash': content_hash, 'chain_hash': chain_hash}


class Kernel:
    def __init__(self, evidence: EvidenceLog) -> None:
        self.evidence = evidence
        self.engines: dict[str, Engine] = {}

    def register(self, name: str, engine: Engine) -> None:
        if name in self.engines:
            raise ValueError(f'engine already registered: {name}')
        self.engines[name] = engine

    def dispatch(self, command: Command) -> dict[str, Any]:
        if command.protocol != 'v1':
            raise ValueError(f'unsupported protocol: {command.protocol}')
        if command.engine not in self.engines:
            raise KeyError(f'unknown engine: {command.engine}')

        # 生产环境应在此处调用策略端口，而不是硬编码授权规则。
        result = self.engines[command.engine](command.payload)
        receipt = self.evidence.append(command, result)
        return {'result': result, 'receipt': receipt}


def trust_eval(data: dict[str, Any]) -> dict[str, Any]:
    score = max(0, min(100, int(data.get('score', 0))))
    return {'normalized_score': score, 'passed': score >= 80}


def asset_registry(data: dict[str, Any]) -> dict[str, Any]:
    return {'asset_id': data['asset_id'], 'status': 'registered'}


def context_engine(data: dict[str, Any]) -> dict[str, Any]:
    return {'language': data.get('language', 'und'), 'terms': data.get('terms', [])}


def policy_engine(data: dict[str, Any]) -> dict[str, Any]:
    return {'allowed': data.get('role') in {'auditor', 'operator'}}


def audit_query(data: dict[str, Any]) -> dict[str, Any]:
    return {'query_id': data.get('query_id'), 'status': 'accepted'}


if __name__ == '__main__':
    kernel = Kernel(EvidenceLog())
    kernel.register('trust-eval', trust_eval)
    kernel.register('asset-registry', asset_registry)
    kernel.register('context', context_engine)
    kernel.register('policy', policy_engine)
    kernel.register('audit-query', audit_query)

    response = kernel.dispatch(
        Command(
            engine='trust-eval',
            actor='developer-42',
            payload={'model': 'demo-model', 'score': 87},
        )
    )
    print(json.dumps(response, ensure_ascii=False, indent=2))

python app.py
cat evidence.jsonl