use std::path::Path;
use std::sync::Arc;
use crate::clock::{MockClock, SharedClock, SystemClock};
use crate::delegation::Delegation;
use crate::error::CoreError;
use crate::gate::{Gate, GateDecision};
use crate::intent::SpendIntent;
use crate::wal::{fnv1a_64, Wal, WalDecision, WalRecord};
#[derive(Debug)]
pub struct WanningState {
gate: Gate,
wal: Option<Wal>,
}
impl WanningState {
pub fn new(clock: SharedClock) -> Self {
Self {
gate: Gate::new(clock),
wal: None,
}
}
pub fn with_wal(clock: SharedClock, wal_path: impl AsRef<Path>) -> Result<Self, CoreError> {
Ok(Self {
gate: Gate::new(clock),
wal: Some(Wal::open(wal_path)?),
})
}
pub fn live(wal_path: impl AsRef<Path>) -> Result<Self, CoreError> {
Self::with_wal(Arc::new(SystemClock), wal_path)
}
pub fn live_resuming(wal_path: impl AsRef<Path>) -> Result<Self, CoreError> {
let path = wal_path.as_ref();
let wal = Wal::open(path)?;
let resumed = Self::replay(path)?;
Ok(Self {
gate: resumed.gate.with_clock(Arc::new(SystemClock)),
wal: Some(wal),
})
}
pub fn gate(&self) -> &Gate {
&self.gate
}
pub fn wal_path(&self) -> Option<&Path> {
self.wal.as_ref().map(Wal::path)
}
pub fn wal_line_count(&self) -> Option<u64> {
self.wal.as_ref().map(Wal::line_count)
}
pub fn last_wal_line(&self) -> Option<u64> {
self.wal_line_count()
}
pub fn audit_chain_tail(&self) -> Option<u64> {
self.wal.as_ref().map(Wal::chain_tail)
}
pub fn register_delegation(&mut self, delegation: Delegation) -> Result<(), CoreError> {
delegation.validate()?;
if self.gate.delegation(&delegation.id).is_some() {
return Err(CoreError::DuplicateDelegation(delegation.id));
}
let record = WalRecord::RegisterDelegation {
ts: self.now(),
delegation: delegation.clone(),
};
if let Some(wal) = self.wal.as_mut() {
wal.append(&record)?;
}
self.gate.register_delegation(delegation)
}
pub fn revoke(&mut self, delegation_id: &str) -> Result<(), CoreError> {
if self.gate.delegation(delegation_id).is_none() {
return Err(CoreError::UnknownDelegation(delegation_id.to_string()));
}
let record = WalRecord::Revoke {
ts: self.now(),
delegation_id: delegation_id.to_string(),
};
if let Some(wal) = self.wal.as_mut() {
wal.append(&record)?;
}
self.gate.revoke(delegation_id)
}
pub fn decide(&mut self, intent: &SpendIntent) -> Result<GateDecision, CoreError> {
let ts = self.now();
let verdict = self.gate.evaluate_at(intent, ts);
let spent_after = match verdict {
GateDecision::Allow { budget_after_cents } => budget_after_cents,
GateDecision::Deny { .. } => self.gate.spent_cents(&intent.delegation_id).unwrap_or(0),
};
let record = WalRecord::Decide {
ts,
decision: match verdict {
GateDecision::Allow { .. } => WalDecision::Allow,
GateDecision::Deny { .. } => WalDecision::Deny,
},
delegation_id: intent.delegation_id.clone(),
intent: intent.clone(),
reason: verdict.deny_reason(),
budget_after_cents: spent_after,
};
if let Some(wal) = self.wal.as_mut() {
wal.append(&record)?;
}
match verdict {
GateDecision::Allow { budget_after_cents } => {
let after = self.gate.commit_at(intent, ts)?;
debug_assert_eq!(after, budget_after_cents);
Ok(GateDecision::Allow {
budget_after_cents: after,
})
}
deny => Ok(deny),
}
}
pub fn state_hash(&self) -> u64 {
let snapshot = serde_json::json!({
"delegations": self.gate.delegations().collect::<Vec<_>>(),
"spent_cents": self.gate.ledger().entries().collect::<Vec<_>>(),
"revoked": self.gate.revocations().iter().collect::<Vec<_>>(),
"used_nonces": self.gate.replay_registry().iter().collect::<Vec<_>>(),
"policy_states": self.gate.policy_states().collect::<Vec<_>>(),
});
fnv1a_64(snapshot.to_string().as_bytes())
}
fn now(&self) -> u64 {
self.gate.clock().now()
}
pub fn replay(wal_path: impl AsRef<Path>) -> Result<Self, CoreError> {
let records = crate::wal::read_verified(wal_path)?.records;
let clock = MockClock::new(0);
let mut state = WanningState::new(Arc::new(clock.clone()));
for (line_no, record) in records {
let record_ts = record.ts();
clock.set_now(record_ts);
match record {
WalRecord::RegisterDelegation { delegation, .. } => state
.gate
.register_delegation(delegation)
.map_err(|e| CoreError::WalMismatch {
line: line_no,
message: format!("重放注册失败: {e}"),
})?,
WalRecord::Revoke { delegation_id, .. } => state
.gate
.revoke(&delegation_id)
.map_err(|e| CoreError::WalMismatch {
line: line_no,
message: format!("重放撤销失败: {e}"),
})?,
WalRecord::Decide {
decision,
intent,
reason,
budget_after_cents,
..
} => {
let ts = record_ts;
let verdict = state.gate.evaluate_at(&intent, ts);
match (verdict, decision, reason) {
(
GateDecision::Allow {
budget_after_cents: recomputed,
},
WalDecision::Allow,
None,
) => {
if recomputed != budget_after_cents {
return Err(CoreError::WalMismatch {
line: line_no,
message: format!(
"放行记录的累计消费与重算不一致:记录 {budget_after_cents} / 重算 {recomputed}"
),
});
}
state.gate.commit_at(&intent, ts).map_err(|e| {
CoreError::WalMismatch {
line: line_no,
message: format!("重放扣减失败: {e}"),
}
})?;
}
(
GateDecision::Deny { reason: recomputed },
WalDecision::Deny,
Some(recorded_reason),
) if recomputed == recorded_reason => {
}
(verdict, decision, reason) => {
return Err(CoreError::WalMismatch {
line: line_no,
message: format!(
"重算判定与记录不一致:重算 {verdict:?} / 记录 {decision:?} reason={reason:?}"
),
});
}
}
}
}
}
Ok(state)
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::clock::{Clock, MockClock};
use crate::gate::DenyReason;
fn tmp_wal(tag: &str) -> std::path::PathBuf {
let dir = std::env::temp_dir().join("wanning-state-tests");
std::fs::create_dir_all(&dir).expect("建临时目录");
dir.join(format!("{tag}-{}.jsonl", std::process::id()))
}
fn delegation() -> Delegation {
Delegation::new(
"d1",
"boss",
"claude-code",
1000,
1000,
2000,
"agent:claude-code",
)
}
fn long_lived_delegation() -> Delegation {
Delegation::new(
"d1",
"boss",
"claude-code",
1000,
1000,
SystemClock.now().checked_add(86_400).expect("有效期溢出"),
"agent:claude-code",
)
}
fn intent(nonce: u64, amount_cents: u64) -> SpendIntent {
SpendIntent::new("d1", nonce, amount_cents, "jd:shop-1", "grocery", "测试")
}
#[test]
fn allow_and_deny_are_both_recorded() {
let path = tmp_wal("both");
let clock = MockClock::new(1500);
let mut state = WanningState::with_wal(Arc::new(clock.clone()), &path).expect("开 WAL");
state.register_delegation(delegation()).expect("注册");
let a = state.decide(&intent(1, 500)).expect("判定");
assert!(a.is_allow());
let d = state.decide(&intent(2, 9000)).expect("判定");
assert_eq!(d.deny_reason(), Some(DenyReason::OverBudget));
let records = crate::wal::read_records(&path).expect("读回");
assert_eq!(records.len(), 3, "注册 + 放行 + 拒绝");
let (_, first_decide) = &records[1];
let (_, second_decide) = &records[2];
match (first_decide.kind(), first_decide.ts(), second_decide.kind()) {
("decide", 1500, "decide") => {}
other => panic!("记录形状不符: {other:?}"),
}
let crate::wal::WalRecord::Decide {
decision,
reason,
budget_after_cents,
..
} = second_decide
else {
panic!("第二条决策记录应是 Decide");
};
assert_eq!(*decision, WalDecision::Deny);
assert_eq!(*reason, Some(DenyReason::OverBudget));
assert_eq!(*budget_after_cents, 500, "拒绝不改账本,累计消费仍是 500");
}
#[test]
fn write_ahead_audit_failure_leaves_state_untouched() {
let dir = std::env::temp_dir().join("wanning-state-tests");
std::fs::create_dir_all(&dir).expect("建临时目录");
let path = dir.join(format!("dir-as-wal-{}.jsonl", std::process::id()));
std::fs::create_dir_all(&path).expect("占位为目录");
let err = WanningState::with_wal(Arc::new(MockClock::new(1500)), &path).unwrap_err();
assert!(matches!(err, CoreError::WalIo(_)), "{err}");
}
#[test]
fn replay_rebuilds_state_and_is_deterministic() {
let path = tmp_wal("replay");
let clock = MockClock::new(1500);
let mut state = WanningState::with_wal(Arc::new(clock.clone()), &path).expect("开 WAL");
state.register_delegation(delegation()).expect("注册");
state.decide(&intent(1, 500)).expect("放行");
state.decide(&intent(2, 9000)).expect("超额拒");
state.decide(&intent(3, 100)).expect("再放行");
state.revoke("d1").expect("撤销");
state.decide(&intent(4, 100)).expect("撤销后拒");
let live_hash = state.state_hash();
assert_eq!(
state.gate().spent_cents("d1"),
Some(600),
"实时累计消费 = 500 + 100"
);
let replayed = WanningState::replay(&path).expect("回放");
let hash_once = replayed.state_hash();
let replayed_again = WanningState::replay(&path).expect("回放二遍");
let hash_twice = replayed_again.state_hash();
assert_eq!(hash_once, hash_twice, "回放两遍 hash 必相同(确定性)");
assert_eq!(hash_once, live_hash, "回放态必须与实时态完全一致");
assert_eq!(replayed.gate().spent_cents("d1"), Some(600));
assert!(replayed.gate().is_revoked("d1"));
assert!(
replayed
.gate()
.replay_registry()
.contains("agent:claude-code", 1),
"重放登记也必须被重建"
);
assert_eq!(replayed.wal_line_count(), None, "回放态不追加记录");
}
#[test]
fn replay_uses_recorded_ts_so_expiry_reproduces() {
let path = tmp_wal("expiry");
let clock = MockClock::new(1500);
let mut state = WanningState::with_wal(Arc::new(clock.clone()), &path).expect("开 WAL");
state.register_delegation(delegation()).expect("注册");
state.decide(&intent(1, 100)).expect("放行");
clock.set_now(2000); state.decide(&intent(2, 100)).expect("过期拒");
let replayed = WanningState::replay(&path).expect("回放");
assert_eq!(replayed.state_hash(), state.state_hash());
}
#[test]
fn replay_fails_closed_on_corrupted_line() {
let path = tmp_wal("replay-corrupt");
let mut state = WanningState::with_wal(Arc::new(MockClock::new(1500)), &path).expect("开");
state.register_delegation(delegation()).expect("注册");
drop(state);
use std::io::Write;
let mut f = std::fs::OpenOptions::new()
.append(true)
.open(&path)
.expect("开");
f.write_all(b"{\"kind\":\"decide\",\"ts\":1,\"dele\n")
.expect("追加坏行");
drop(f);
match WanningState::replay(&path) {
Err(CoreError::WalBadLine { line, .. }) => assert_eq!(line, 2),
other => panic!("应 fail-closed 报错,实际 {other:?}"),
}
}
#[test]
fn replay_fails_closed_when_record_disagrees_with_recomputation() {
let path = tmp_wal("replay-tampered");
let mut state = WanningState::with_wal(Arc::new(MockClock::new(1500)), &path).expect("开");
state.register_delegation(delegation()).expect("注册");
state.decide(&intent(1, 100)).expect("放行");
drop(state);
use std::io::Write;
let verified = crate::wal::read_verified(&path).expect("读已有历史");
let forged = crate::wal::WalLine {
seq: verified.records.len() as u64 + 1,
prev: verified.tail,
rec: WalRecord::Decide {
ts: 1500,
decision: WalDecision::Allow,
delegation_id: "d1".to_string(),
intent: intent(1, 100),
reason: None,
budget_after_cents: 200,
},
};
let mut f = std::fs::OpenOptions::new()
.append(true)
.open(&path)
.expect("开");
f.write_all(serde_json::to_string(&forged).unwrap().as_bytes())
.and_then(|()| f.write_all(b"\n"))
.expect("追加");
drop(f);
match WanningState::replay(&path) {
Err(CoreError::WalMismatch { line, message }) => {
assert_eq!(line, 3, "不一致要指到行");
assert!(message.contains("不一致"), "{message}");
}
other => panic!("篡改记录必须 fail-closed,实际 {other:?}"),
}
}
#[test]
fn audit_chain_tail_matches_independent_read_side_recompute() {
let path = tmp_wal("chain-tail");
let mut state = WanningState::with_wal(Arc::new(MockClock::new(1500)), &path).expect("开");
state.register_delegation(delegation()).expect("注册");
state.decide(&intent(1, 500)).expect("放行");
state.decide(&intent(2, 9000)).expect("超额拒");
state.revoke("d1").expect("撤销");
let live_tail = state.audit_chain_tail().expect("必有 WAL");
let verified = crate::wal::read_verified(&path).expect("读回验链");
assert_eq!(verified.tail, live_tail, "读侧独立重算链尾 == 实时链尾");
assert_eq!(
WanningState::replay(&path).expect("回放").state_hash(),
state.state_hash(),
"链验过后,回放对账照常成立"
);
}
#[test]
fn live_resuming_fails_closed_on_broken_chain() {
let path = tmp_wal("resume-chain");
{
let mut state =
WanningState::with_wal(Arc::new(MockClock::new(1500)), &path).expect("开");
state.register_delegation(delegation()).expect("注册");
state.decide(&intent(1, 100)).expect("放行");
state.decide(&intent(2, 9000)).expect("超额拒");
}
let mut lines = crate::wal::raw_lines(&path).expect("读 WAL");
let mut value: serde_json::Value = serde_json::from_str(&lines[1]).expect("行是 JSON");
value["rec"]["intent"]["memo"] = serde_json::json!("被改写的备注");
lines[1] = value.to_string();
std::fs::write(&path, lines.join("\n") + "\n").expect("重写 WAL");
match WanningState::live_resuming(&path) {
Err(CoreError::WalChainBroken { line, .. }) => {
assert_eq!(line, 3, "断链点 = 被改行的下一行(prev 对不上)")
}
other => panic!("链断裂必须拒启,实际 {other:?}"),
}
}
#[test]
fn state_hash_changes_when_state_changes() {
let path = tmp_wal("hash");
let clock = MockClock::new(1500);
let mut state = WanningState::with_wal(Arc::new(clock.clone()), &path).expect("开");
state.register_delegation(delegation()).expect("注册");
let h0 = state.state_hash();
state.decide(&intent(1, 100)).expect("放行");
let h1 = state.state_hash();
state.revoke("d1").expect("撤销");
let h2 = state.state_hash();
assert_ne!(h0, h1, "扣减后 hash 必变");
assert_ne!(h1, h2, "撤销后 hash 必变");
}
#[test]
fn empty_wal_replays_to_empty_state() {
let path = tmp_wal("empty");
std::fs::write(&path, "").expect("写空文件");
let replayed = WanningState::replay(&path).expect("空 WAL 是合法状态");
assert_eq!(
replayed.state_hash(),
WanningState::new(Arc::new(MockClock::new(0))).state_hash()
);
}
#[test]
fn live_resuming_carries_ledger_revocations_and_nonces() {
let path = tmp_wal("resume");
{
let clock = MockClock::new(1500);
let mut state = WanningState::with_wal(Arc::new(clock.clone()), &path).expect("开");
state
.register_delegation(long_lived_delegation())
.expect("注册");
state.decide(&intent(1, 500)).expect("放行");
state.decide(&intent(2, 100)).expect("再放行");
state.revoke("d1").expect("撤销");
}
let resumed = WanningState::live_resuming(&path).expect("续跑");
assert_eq!(resumed.gate().spent_cents("d1"), Some(600));
assert!(resumed.gate().is_revoked("d1"), "撤销必须跨重启存活");
assert_eq!(
resumed.state_hash(),
WanningState::replay(&path).expect("回放").state_hash(),
"续跑态与回放态必须一致"
);
assert!(
resumed.gate().clock().now() > 1_700_000_000,
"续跑必须用系统时钟,得到 {}",
resumed.gate().clock().now()
);
let mut resumed = resumed;
let deny = resumed.decide(&intent(3, 100)).expect("判定");
assert_eq!(deny.deny_reason(), Some(DenyReason::Revoked));
let replay_deny = resumed.decide(&intent(1, 100)).expect("判定");
assert_eq!(replay_deny.deny_reason(), Some(DenyReason::Revoked));
assert_eq!(resumed.gate().spent_cents("d1"), Some(600), "账本不动");
let records = crate::wal::read_records(&path).expect("读回");
assert_eq!(records.len(), 6, "注册+2 放行+撤销+续跑后 2 条拒绝");
}
#[test]
fn live_resuming_on_fresh_wal_starts_empty() {
let path = tmp_wal("resume-fresh");
let mut state = WanningState::live_resuming(&path).expect("新 WAL 直接续跑=空账开张");
assert_eq!(
state.state_hash(),
WanningState::replay(&path).expect("回放").state_hash()
);
state
.register_delegation(long_lived_delegation())
.expect("注册");
assert!(state.decide(&intent(1, 100)).expect("判定").is_allow());
}
#[test]
fn live_resuming_fails_closed_on_corrupted_wal() {
let path = tmp_wal("resume-corrupt");
{
let mut state =
WanningState::with_wal(Arc::new(MockClock::new(1500)), &path).expect("开");
state.register_delegation(delegation()).expect("注册");
}
use std::io::Write;
let mut f = std::fs::OpenOptions::new()
.append(true)
.open(&path)
.expect("开");
f.write_all(b"{\"kind\":\"decide\",\"ts\":1,\"dele\n")
.expect("追加坏行");
drop(f);
match WanningState::live_resuming(&path) {
Err(CoreError::WalBadLine { line, .. }) => assert_eq!(line, 2),
other => panic!("审计损坏必须拒启,实际 {other:?}"),
}
}
}