------------------------------ MODULE VideoConfSignaling ------------------------------
EXTENDS Integers, Sequences, FiniteSets, TLC

CONSTANTS
  Node,          * 信令节点集合
  SessionId,     * 会话标识集合
  UserId,        * 用户标识集合
  MaxRetries,    * 最大重试次数
  Quorum         * 共识节点数（如 Raft 多数派）

ASSUME Node in SUBSET Node / Cardinality(Node) >= 3
ASSUME Quorum in 2..Cardinality(Node)

IDLE → OFFER_SENT → ANSWER_RECEIVED → ICE_GATHERING → CONNECTED
                    ↓                    ↓
               RE_NEGOTIATING         ICE_FAILED
                    ↓                    ↓
               CONNECTED ←────────── DISCONNECTED

SignalState == {"IDLE", "OFFER_SENT", "ANSWER_RECEIVED", 
                "ICE_GATHERING", "CONNECTED", "RE_NEGOTIATING",
                "ICE_FAILED", "DISCONNECTED"}

vars == << sessionState, pendingOffers, iceCandidates, txnLog >>

TypeOK == 
  / sessionState in [SessionId -> SignalState]
  / pendingOffers in [SessionId -> SUBSET OfferId]
  / iceCandidates in [SessionId -> Seq(CandidateId)]
  / txnLog in Seq(TxnRecord)

InitiateOffer(sid, uid, offerId) ==
  / sessionState[sid] = "IDLE"
  / sessionState' = [sessionState EXCEPT ![sid] = "OFFER_SENT"]
  / pendingOffers' = [pendingOffers EXCEPT ![sid] = @ cup {offerId}]
  / UNCHANGED <<iceCandidates, txnLog>>

HandleAnswer(sid, offerId, answerId) ==
  / sessionState[sid] = "OFFER_SENT"
  / offerId in pendingOffers[sid]
  / sessionState' = [sessionState EXCEPT ![sid] = "ANSWER_RECEIVED"]
  / pendingOffers' = [pendingOffers EXCEPT ![sid] = @  {offerId}]
  / UNCHANGED <<iceCandidates, txnLog>>

TxnPhase == {"PREPARE", "COMMIT", "COMPENSATING", "ROLLED_BACK", "COMMITTED"}

TxnRecord == [sid: SessionId, phase: TxnPhase, 
              steps: Seq(StepName), 
              compensated: Seq(StepName)]

SagaStart(sid) ==
  / sessionState[sid] = "IDLE"
  / txnLog' = Append(txnLog, [sid |-> sid, phase |-> "PREPARE", 
                               steps |-> <<"Reserve", "SignalChannel", "Recording", "Billing">>,
                               compensated |-> << >>])
  / UNCHANGED <<sessionState, pendingOffers, iceCandidates>>

SagaStepSuccess(sid, stepIdx) ==
  / E rec in txnLog: rec.sid = sid / rec.phase = "PREPARE"
  / LET rec == CHOOSE r in txnLog: r.sid = sid / r.phase = "PREPARE"
     IN  stepIdx <= Len(rec.steps)
  / txnLog' = [txnLog EXCEPT ![IndexOf(rec)] = 
                [rec EXCEPT !.steps = Tail(rec.steps)]]
  / UNCHANGED <<sessionState, pendingOffers, iceCandidates>>

SagaCompensate(sid) ==
  / E rec in txnLog: rec.sid = sid / rec.phase = "PREPARE"
  / txnLog' = [txnLog EXCEPT ![IndexOf(rec)] = 
                [rec EXCEPT !.phase = "COMPENSATING",
                          !.compensated = << >>]]
  / UNCHANGED <<sessionState, pendingOffers, iceCandidates>>

Message == [type: {"OFFER", "ANSWER", "CANDIDATE", "TXN_CMD", "TXN_ACK"},
            src: Node, dst: Node, payload: SessionId, msgId: Nat]

network == [inbox: [Node -> Seq(Message)], 
            dropped: Seq(Message), 
            duplicated: Seq(Message)]

NetworkSend(msg) ==
  / network'.inbox[msg.dst] = Append(network.inbox[msg.dst], msg)
  / UNCHANGED <<network.dropped, network.duplicated>>

NetworkDrop(msg) ==
  / network'.dropped = Append(network.dropped, msg)
  / UNCHANGED <<network.inbox, network.duplicated>>

NetworkDuplicate(msg) ==
  / network'.duplicated = Append(network.duplicated, msg)
  / network'.inbox[msg.dst] = Append(network.inbox[msg.dst], msg)
  / UNCHANGED network.dropped

* 1. 会话状态单向流转，禁止非法回退
StateMonotonic == 
  A sid in SessionId:
    ~(sessionState[sid] = "CONNECTED" / 
      Enabled(<InitiateOffer(sid, _, _)>_vars))

* 2. 分布式事务原子性：要么全成功，要么全补偿
TxnAtomicity ==
  A rec in txnLog:
    (rec.phase = "COMMITTED") => (Len(rec.steps) = 0)
    / (rec.phase = "ROLLED_BACK") => (Len(rec.compensated) = Len(originalSteps))

* 3. 信令幂等：同一 msgId 至多处理一次
IdempotentProcessing ==
  A n in Node:
    Cardinality({m in network.inbox[n] : m.msgId = k}) <= 1

* 4. 资源不超卖：并发会话数不超过节点配额
ResourceQuota ==
  A n in Node:
    Cardinality({sid in SessionId: sessionState[sid] in {"CONNECTED", "ICE_GATHERING"} 
                 / Owner(sid) = n}) <= MaxSessionsPerNode

* 最终一致性：发起 Offer 后，若网络稳定，最终达成 CONNECTED 或显式失败
LivenessOfferAnswer ==
  A sid in SessionId:
    sessionState[sid] = "OFFER_SENT" ~> 
    (sessionState[sid] in {"CONNECTED", "DISCONNECTED", "ICE_FAILED"})

* 事务终结性：SAGA 不会永久停留在 PREPARE/COMPENSATING
TxnTermination ==
  A rec in txnLog:
    rec.phase in {"PREPARE", "COMPENSATING"} ~> 
    rec.phase in {"COMMITTED", "ROLLED_BACK"}

CONSTANTS
  Node = {n1, n2, n3}
  SessionId = {s1, s2}
  UserId = {u1, u2}
  MaxRetries = 2
  Quorum = 2

INVARIANTS TypeOK StateMonotonic TxnAtomicity IdempotentProcessing ResourceQuota
PROPERTIES LivenessOfferAnswer TxnTermination

ACTION_CONSTRAINTS
  * 限制网络非确定性，避免状态爆炸
  NetworkDrop(_) => FALSE  * 先跑无丢包基线
  NetworkDuplicate(_) => FALSE

SYMMETRY
  Permutations(Node)  * 节点对称性归约

IceCandidate == [type: {"HOST", "SRFLX", "PRFLX", "RELAY"}, 
                 priority: Nat, 
                 foundation: String,
                 component: 1..2]  * RTP/RTCP

IceState == [localCandidates: Seq(IceCandidate),
             remoteCandidates: Seq(IceCandidate),
             nominatedPair: [local: IceCandidate, remote: IceCandidate] | NULL,
             controlling: BOOLEAN,      * ICE Controlling 角色
             checkList: Seq(PairCheck)] * 连通性检查队列

PairCheck == [local: IceCandidate, remote: IceCandidate, 
              state: {"WAITING", "IN_PROGRESS", "SUCCEEDED", "FAILED"},
              nominated: BOOLEAN]

* 关键动作：受控方收到 USE-CANDIDATE 标记时的提名逻辑
ControllerNominate(sid, pairIdx) ==
  / iceState[sid].controlling = TRUE
  / pairIdx in DOMAIN iceState[sid].checkList
  / iceState[sid].checkList[pairIdx].state = "SUCCEEDED"
  / iceState' = [iceState EXCEPT 
       ![sid].nominatedPair = [local |-> iceState[sid].checkList[pairIdx].local,
                               remote |-> iceState[sid].checkList[pairIdx].remote],
       ![sid].checkList[pairIdx].nominated = TRUE]
  / UNCHANGED <<sessionState, ...>>

* 角色冲突解决：双方均为 Controlling 时，通过 Tie-Breaker 值裁决
RoleConflictResolve(sid) ==
  / iceState[sid].controlling = TRUE
  / ReceivedRoleConflict(sid)  * 抽象网络消息
  / LET myTieBreaker == GenerateTieBreaker(sid)
         peerTieBreaker == ExtractTieBreaker(ReceivedMsg(sid))
     IN  IF myTieBreaker > peerTieBreaker
         THEN iceState' = [iceState EXCEPT ![sid].controlling = TRUE]  * 保持
         ELSE iceState' = [iceState EXCEPT ![sid].controlling = FALSE] * 降级
  / UNCHANGED <<sessionState, ...>>

CONSTANT MaxClockDrift  * 最大时钟漂移比例，如 10% 即 0.1

LogicalClock == [node in Node |-> Nat]  * 每节点逻辑时间

Tick(n) ==  * 时间推进，模拟漂移
  / LogicalClock' = [LogicalClock EXCEPT ![n] = @ + 1 + Drift()]
  / UNCHANGED <<vars, network>>

Drift() ==  * 非确定性漂移
  E d in 0..MaxClockDrift: TRUE

TimerWheel == [node in Node |-> [timerId in TimerId |-> 
            [deadline: Nat, callback: Action, cancelled: BOOLEAN]]]

ScheduleTimer(n, timerId, deadline, action) ==
  / ~TimerWheel[n][timerId].cancelled
  / TimerWheel' = [TimerWheel EXCEPT 
       ![n][timerId] = [deadline |-> deadline, callback |-> action, cancelled |-> FALSE]]
  / UNCHANGED <<vars, network, LogicalClock>>

FireTimers(n) ==
  E timerId in DOMAIN TimerWheel[n]:
    / TimerWheel[n][timerId].deadline <= LogicalClock[n]
    / ~TimerWheel[n][timerId].cancelled
    / TimerWheel[n][timerId].callback  * 执行回调（如 RetransmitOffer）
    / TimerWheel' = [TimerWheel EXCEPT ![n][timerId].cancelled = TRUE]
    / UNCHANGED <<vars, network, LogicalClock>>

---------------------------- MODULE SignalingCore ----------------------------
INSTANCE SignalingTypes  * 共享类型定义
VARIABLES sigState, pendingMsg, ...
SigNext == InitiateOffer / HandleAnswer / IceGather / ...
SigSpec == Init / [][SigNext]_vars / Fairness
* 导出契约：对外承诺的安全性/活性
SigContract == INSTANCE ContractTemplate WITH 
  Safety <- SigSafetyInvariants, 
  Liveness <- SigLivenessProps

---------------------------- MODULE TxnCoordinator ----------------------------
INSTANCE TxnTypes
VARIABLES txnLog, participantState, ...
TxnNext == SagaStart / SagaStep / SagaCompensate / ...
TxnSpec == Init / [][TxnNext]_vars / Fairness
TxnContract == INSTANCE ContractTemplate WITH ...

---------------------------- MODULE ComposedSystem ----------------------------
INSTANCE SignalingCore AS Sig
INSTANCE TxnCoordinator AS Txn
* 组合胶水：信令层调用事务层创建会话资源
ComposedNext == 
  / Sig.SigNext 
  / Txn.TxnNext
  / SigTxnBridge  * 跨模块动作，如：Sig.InitiateOffer -> Txn.SagaStart

* 假设-保证验证：在验证 Sig 时，假设 Txn 满足 TxnContract；反之亦然
THEOREM ComposedSystem => Sig.SigContract!Safety / Txn.TxnContract!Safety

# tools/tla2go/generator.py (伪代码)
class TLAStateMachineGenerator:
    def __init__(self, tla_ast):
        self.states = tla_ast.extract_enum("SignalState")
        self.transitions = tla_ast.extract_actions(["InitiateOffer", "HandleAnswer", ...])
        self.invariants = tla_ast.extract_invariants(["StateMonotonic", "IdempotentProcessing"])

    def generate_go_fsm(self) -> str:
        # 1. 生成状态定义
        code = "type SessionState intnconst (n"
        for s in self.states: code += f"    State{s} SessionState = iotan"
        code += ")nn"
        
        # 2. 生成带守卫的迁移方法
        for act in self.transitions:
            guard = self._translate_expr(act.guard)
            updates = self._translate_updates(act.updates)
            code += f"func (s *Session) {act.name}(ctx context.Context, args {act.args}) error {{n"
            code += f"    if !({guard}) {{ return ErrInvalidTransition }}n"
            code += f"    {updates}n"
            code += f"    s.emitEvent({act.name}Event{{...}})n"
            code += "    return niln}nn"
        
        # 3. 生成不变量运行时断言（开发/测试环境开启）
        if self.invariants:
            code += "// 运行时不变量检查n"
            for inv in self.invariants:
                code += f"func check{inv.name}(s *Session) {{ assert({self._translate_expr(inv.expr)}) }}n"
        
        return code

# .github/workflows/tla-driven.yml
jobs:
  tla-verify:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - name: Run TLC Model Check
        run: make tla-check CONFIG=ci_3node_2session.cfg
      - name: Generate Code from Spec
        run: python tools/tla2go/generator.py specs/VideoConfSignaling.tla > internal/fsm/signaling_gen.go
      - name: Verify Generated Code Compiles
        run: go build ./internal/fsm/...
      - name: Run Contract Tests (Pact)
        run: make contract-test
  # 阻断合并：若 TLC 失败或生成代码编译失败，PR 不可合并

CONSTANT MaxConcurrentSessions, MaxPendingMessages, WorkerPoolSize

Resources == [slots: Nat, msgQueue: Seq(Message), workers: Nat]

AcquireSlot(sid) ==
  / Resources.slots > 0
  / Resources' = [Resources EXCEPT !.slots = @ - 1]
  / UNCHANGED <<vars, network>>

ReleaseSlot(sid) ==
  / Resources.slots < MaxConcurrentSessions
  / Resources' = [Resources EXCEPT !.slots = @ + 1]
  / UNCHANGED <<vars, network>>

EnqueueMsg(msg) ==
  / Len(Resources.msgQueue) < MaxPendingMessages
  / Resources' = [Resources EXCEPT !.msgQueue = Append(@, msg)]
  / UNCHANGED <<vars, network>>

WorkerProcess ==
  / Resources.workers > 0
  / Resources.msgQueue # << >>
  / LET msg == Head(Resources.msgQueue)
     IN  / ProcessMessage(msg)  * 调用核心状态机动作
         / Resources' = [Resources EXCEPT 
              !.msgQueue = Tail(@),
              !.workers = @ - 1]
         / UNCHANGED <<vars, network>>

* 无饥饿：若有消息待处理且有空闲 Worker，最终必被处理
NoStarvation ==
  A msg in Resources.msgQueue:
    (Resources.workers > 0) ~> (msg notin Resources.msgQueue)

* 背压生效：队列满时不再接受新消息（防 OOM）
BackpressureEffective ==
  Len(Resources.msgQueue) = MaxPendingMessages => 
    ~Enabled(<EnqueueMsg(_)>_vars)

* 最大延迟界（定性）：消息入队到出队步数 < MaxSteps
MaxLatencyBound ==
  A msg in Resources.msgQueue:
    (msg in Resources.msgQueue) ~> (msg notin Resources.msgQueue) 
    * 结合 TLC 统计最大步数，换算物理时间上界

// 会话建立成功率（含丢包、重试、超时）
Pmax=? [ F<=100 "sessionState=CONNECTED" ]

// 平均会话建立步数（近似延迟）
R{"steps"}min=? [ F "sessionState=CONNECTED" ]

// 分布式事务补偿触发概率
P=? [ F "txnPhase=COMPENSATING" ]

// 资源耗尽概率（槽位满）
P=? [ F "Resources.slots=0" ]

specs/
├── VideoConfSignaling.tla          # 主规约
├── types/
│   ├── SignalingTypes.tla          # 共享类型（独立版本 v1.2.0）
│   └── TxnTypes.tla
├── contracts/
│   ├── SignalingContract.tla       # 对外契约（语义化版本）
│   └── TxnContract.tla
├── configs/
│   ├── ci_3node_2session.cfg       # CI 基线配置
│   ├── perf_5node_10session.cfg    # 性能基线配置
│   └── chaos_7node.cfg             # 混沌工程配置
└── proofs/
    ├── StateMonotonic.tla          # TLAPS 证明脚本
    └── TxnAtomicity.tla

* 补充最小复现规约片段
BuggyReInvite(sid) ==
  / sessionState[sid] = "ICE_GATHERING"  * 允许在采集期重协商
  / sessionState' = [sessionState EXCEPT ![sid] = "OFFER_SENT"]
  / pendingOffers' = [pendingOffers EXCEPT ![sid] = @ cup {newOfferId}]
  / UNCHANGED iceState  * 关键缺陷：未重置 ICE 状态机
  / UNCHANGED <<txnLog, network>>

* TLC 反例轨迹片段：
1. InitiateOffer(s1) -> OFFER_SENT
2. HandleAnswer(s1) -> ANSWER_RECEIVED
3. IceGather(s1) -> ICE_GATHERING (checkList=[pair1, pair2], nominatedPair=NULL)
4. BuggyReInvite(s1) -> OFFER_SENT (checkList 保留, nominatedPair 保留)
5. HandleAnswer(s1) -> ANSWER_RECEIVED
6. IceGather(s1) -> ICE_GATHERING (checkList 追加新 pair, 旧 pair 状态丢失)
7. ControllerNominate(s1, oldPairIdx) -> nominatedPair=oldPair (但旧 pair 已不在新 checkList)
8. MediaPathEstablished(s1) 使用 nominatedPair -> 连接失败/黑屏

* 修正：重协商必须显式重置 ICE 子状态机
SafeReInvite(sid) ==
  / sessionState[sid] in {"CONNECTED", "ICE_GATHERING", "ICE_FAILED"}
  / sessionState' = [sessionState EXCEPT ![sid] = "OFFER_SENT"]
  / pendingOffers' = [pendingOffers EXCEPT ![sid] = @ cup {newOfferId}]
  / iceState' = [iceState EXCEPT 
       ![sid] = [localCandidates |-> << >>, 
                 remoteCandidates |-> << >>,
                 nominatedPair |-> NULL,
                 checkList |-> << >>,
                 controlling |-> TRUE]]  * 重协商发起方强制为 Controlling
  / UNCHANGED <<txnLog, network>>