lambdagent/demo/demo2_store_independence.py

#!/usr/bin/env python3
"""
Demo 2: 并行安全分析 — Store Independence 检查
===============================================
构造 IoT 场景: ClimateAgent + LightAgent 同时修改 userComfort，
lambdagent 的 store_analysis 在编译时发现写入冲突。

核心信息:
  - 两个 Agent 并行修改同一个 store key → 竞态条件
  - 在真实设备上: 空调和灯互相打架
  - lambdagent 的 Proposition 30 保证并行安全
"""

import sys
import os
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))

from lambdagent import Lam, Pair, StoreConflictError
from lambdagent.store_analysis import check_store_independence, writes


def main():
    print("=" * 60)
    print("Demo 2: 并行安全分析 — Store Independence")
    print("场景: 鸿蒙 IoT 智能家居 — 多 Agent 并行控制")
    print("=" * 60)
    print()

    # ── Step 1: 定义三个 IoT Agent ──
    print("[Step 1] 定义 IoT Agent")

    climate = Lam("ClimateAgent", "根据温度调节空调")
    climate._writes = {"userComfort.temperature", "deviceState.ac"}

    light = Lam("LightAgent", "根据光线调节灯光")
    light._writes = {"userComfort.brightness", "deviceState.light"}

    security = Lam("SecurityAgent", "安防监控")
    security._writes = {"securityState.alert", "deviceState.camera"}

    print(f"  ClimateAgent  writes: {climate._writes}")
    print(f"  LightAgent    writes: {light._writes}")
    print(f"  SecurityAgent writes: {security._writes}")
    print()

    # ── Step 2: 检查 Climate + Light (无冲突) ──
    print("[Step 2] 检查 ClimateAgent ∥ LightAgent")
    try:
        check_store_independence([climate, light])
        print("  ✅ Store independence satisfied — 无写入冲突")
    except StoreConflictError as e:
        print(f"  ❌ [CONFLICT] {e}")
    print()

    # ── Step 3: 构造冲突场景 ──
    print("[Step 3] 构造冲突: 两个 Agent 都写 userComfort")
    print("  (模拟: LightAgent 也想调节温度以配合灯光色温)")

    light_v2 = Lam("LightAgent_v2", "调节灯光 + 色温联动温度")
    light_v2._writes = {"userComfort.temperature", "userComfort.brightness", "deviceState.light"}

    print(f"  ClimateAgent   writes: {climate._writes}")
    print(f"  LightAgent_v2  writes: {light_v2._writes}")
    print()

    try:
        check_store_independence([climate, light_v2])
        print("  ✅ No conflict")
    except StoreConflictError as e:
        print(f"  ❌ [CONFLICT] {e}")
        print()
        print("  → 两个 Agent 同时修改 userComfort.temperature")
        print("  → 在真实设备上: 空调刚设 26°C，灯光联动改成 22°C")
        print("  → 用户体验: 空调忽冷忽热，灯光闪烁")
        print("  → lambdagent 在编译时就阻止了这种部署")
    print()

    # ── Step 4: 三 Agent 全量检查 ──
    print("[Step 4] 三 Agent 全量检查")
    agents = [climate, light, security]
    try:
        check_store_independence(agents)
        print("  ✅ 三个 Agent 并行安全 — 写入集合两两不相交")
        intersection_info = [
            f"    {a._name} ∩ {b._name} = ∅"
            for i, a in enumerate(agents)
            for b in agents[i+1:]
        ]
        for info in intersection_info:
            print(info)
    except StoreConflictError as e:
        print(f"  ❌ [CONFLICT] {e}")
    print()

    print("=" * 60)
    print("结论: Proposition 30 — Pair 安全当且仅当写入集合两两不相交")
    print("鸿蒙 Agent DSL 没有这个检查 → 跨设备并行存在数据竞争风险")
    print("=" * 60)


if __name__ == "__main__":
    main()