关键工程不变量：Fail-Closed、崩溃一致性、持久Latch的Python实现

1. 默认失败关闭（0.0V 安全状态）

如果某个操作违反了边界条件、提供了无效的 MAC 标签，或超过了异常阈值，系统不会仅仅抛出一个异常：它会进入一个吸收性的 KILL 锁存器，并将执行器寄存器断电至安全状态（0.0V / 0.0 kW）。

2. 通过两阶段 WAL 实现崩溃一致性

在将日志同步到磁盘之前就在内存中变更状态，会导致幽灵状态失同步。在实际变更之前就写入 COMMITTED，则会产生虚假日志。

• 第一阶段（PREPARE）：记录确切的意图并调用 os.fsync()。
• 第二阶段（EXECUTE）：将状态变更应用到硬件/数据库。
• 第三阶段（COMMIT）：记录最终提交状态并调用 os.fsync()。

如果进程在第二阶段崩溃，重启时的恢复程序会发现悬空的 PREPARE 条目，并立即强制执行失败关闭的 KILL，防止操作继续执行。

3. 跨重启持久化锁存

如果服务器电源循环重置状态为 NOMINAL，那么安全锁就毫无意义。在冷启动时，引擎从磁盘回放并验证哈希链，恢复 state = "KILL"、latched = True，并需要两阶段人工授权才能重置。

💻 最小化 Python 实现

以下是简洁 Python 实现的核心模式：

import hmac
import hashlib
import json
import os
from typing import Dict, Any, Tuple

class ConsequenceIsolationEngine:
    def __init__(self, secret_key: bytes, log_file: str = "audit_hashchain.jsonl"):
        self.secret_key = secret_key
        self.log_file = log_file

        # State registers
        self.state = "NOMINAL"
        self.latched = False
        self.actuator_power_kw = 0.0
        self.prev_hash = "0" * 64
        self.tick = 0

        # Recover full state from persistent disk log
        self._recover_and_verify_log()

    def _recover_and_verify_log(self):
        if not os.path.exists(self.log_file):
            return

        last_hash = "0" * 64
        pending_prepare = None

        with open(self.log_file, "r", encoding="utf-8") as f:
            for line in f:
                if not line.strip(): continue
                entry = json.loads(line)

                # Verify SHA-256 link
                if entry.get("prev_hash") != last_hash:
                    raise ValueError("Corrupt hash chain detected on disk!")

                logged_hash = entry.get("hash")
                data = {k: v for k, v in entry.items() if k != "hash"}
                calc_hash = hashlib.sha256(json.dumps(data, sort_keys=True).encode()).hexdigest()
                if calc_hash != logged_hash:
                    raise ValueError("Hash mismatch on startup audit!")

                last_hash = logged_hash
                self.tick = entry.get("tick", 0) + 1

                if entry.get("event_type") == "PREPARE":
                    pending_prepare = entry
                elif entry.get("event_type") in ("COMMITTED", "FAIL_CLOSED_KILL"):
                    pending_prepare = None
                    # Restore persistent state
                    self.state = entry.get("state", "NOMINAL")
                    self.latched = entry.get("latched", False)
                    self.actuator_power_kw = float(entry.get("power_kw", 0.0))

        self.prev_hash = last_hash

        # Fail-closed if crashed mid-mutation
        if pending_prepare:
            self.state = "KILL"
            self.latched = True
            self.actuator_power_kw = 0.0

    def _wal_sync(self, entry: Dict[str, Any]) -> str:
        canonical = json.dumps(entry, sort_keys=True)
        h = hashlib.sha256(canonical.encode()).hexdigest()
        entry["hash"] = h

        with open(self.log_file, "a", encoding="utf-8") as f:
            f.write(json.dumps(entry) + "\n")
            f.flush()
            os.fsync(f.fileno())  # Guaranteed write to physical disk

        self.prev_hash = h
        self.tick += 1
        return h

    def evaluate_and_execute(self, candidate: Dict[str, Any]) -> Tuple[str, float]:
        if self.latched:
            return self.state, self.actuator_power_kw

        # Verify HMAC tag (Tamper detection)
        provided_mac = candidate.get("mac_tag", "")
        payload = {k: v for k, v in candidate.items() if k != "mac_tag"}
        computed_mac = hmac.new(self.secret_key, json.dumps(payload, sort_keys=True).encode(), hashlib.sha256).hexdigest()

        target_power = candidate.get("target_power", -1.0)
        is_valid = hmac.compare_digest(provided_mac, computed_mac) and (0.0 <= target_power <= 100.0)

        verdict = "OPEN" if is_valid else "KILL"
        next_power = target_power if is_valid else 0.0
        next_state = "NOMINAL" if is_valid else "KILL"
        next_latched = not is_valid

        # 1. WAL PREPARE (Write-Ahead before mutation)
        self._wal_sync({
            "tick": self.tick,
            "event_type": "PREPARE",
            "verdict": verdict,
            "target_power_kw": next_power,
            "prev_hash": self.prev_hash
        })

        # 2. MUTATION
        self.state = next_state
        self.latched = next_latched
        self.actuator_power_kw = next_power

        # 3. WAL COMMIT
        self._wal_sync({
            "tick": self.tick,
            "event_type": "COMMITTED",
            "state": self.state,
            "latched": self.latched,
            "power_kw": self.actuator_power_kw,
            "prev_hash": self.prev_hash
        })

        return self.state, self.actuator_power_kw

随着自主系统承担越来越多的职责：

• 永远不要让 Agent 直接与真实执行器或关键数据库对话。
• 使用两阶段预写日志（os.fsync），这样电力中断不会破坏你的安全状态。
• 在冷启动时持久化锁存——安全跳闸必须能承受服务器重启。