1use std::path::Path;
13use std::sync::Arc;
14
15use crate::clock::{MockClock, SharedClock, SystemClock};
16use crate::delegation::Delegation;
17use crate::error::CoreError;
18use crate::gate::{Gate, GateDecision};
19use crate::intent::SpendIntent;
20use crate::wal::{fnv1a_64, Wal, WalDecision, WalRecord};
21
22#[derive(Debug)]
24pub struct WanningState {
25 gate: Gate,
26 wal: Option<Wal>,
27}
28
29impl WanningState {
30 pub fn new(clock: SharedClock) -> Self {
32 Self {
33 gate: Gate::new(clock),
34 wal: None,
35 }
36 }
37
38 pub fn with_wal(clock: SharedClock, wal_path: impl AsRef<Path>) -> Result<Self, CoreError> {
40 Ok(Self {
41 gate: Gate::new(clock),
42 wal: Some(Wal::open(wal_path)?),
43 })
44 }
45
46 pub fn live(wal_path: impl AsRef<Path>) -> Result<Self, CoreError> {
51 Self::with_wal(Arc::new(SystemClock), wal_path)
52 }
53
54 pub fn live_resuming(wal_path: impl AsRef<Path>) -> Result<Self, CoreError> {
69 let path = wal_path.as_ref();
70 let wal = Wal::open(path)?;
72 let resumed = Self::replay(path)?;
73 Ok(Self {
74 gate: resumed.gate.with_clock(Arc::new(SystemClock)),
75 wal: Some(wal),
76 })
77 }
78
79 pub fn gate(&self) -> &Gate {
80 &self.gate
81 }
82
83 pub fn wal_path(&self) -> Option<&Path> {
84 self.wal.as_ref().map(Wal::path)
85 }
86
87 pub fn wal_line_count(&self) -> Option<u64> {
89 self.wal.as_ref().map(Wal::line_count)
90 }
91
92 pub fn last_wal_line(&self) -> Option<u64> {
94 self.wal_line_count()
95 }
96
97 pub fn audit_chain_tail(&self) -> Option<u64> {
102 self.wal.as_ref().map(Wal::chain_tail)
103 }
104
105 pub fn register_delegation(&mut self, delegation: Delegation) -> Result<(), CoreError> {
107 delegation.validate()?;
110 if self.gate.delegation(&delegation.id).is_some() {
111 return Err(CoreError::DuplicateDelegation(delegation.id));
112 }
113 let record = WalRecord::RegisterDelegation {
114 ts: self.now(),
115 delegation: delegation.clone(),
116 };
117 if let Some(wal) = self.wal.as_mut() {
118 wal.append(&record)?;
119 }
120 self.gate.register_delegation(delegation)
121 }
122
123 pub fn revoke(&mut self, delegation_id: &str) -> Result<(), CoreError> {
125 if self.gate.delegation(delegation_id).is_none() {
126 return Err(CoreError::UnknownDelegation(delegation_id.to_string()));
127 }
128 let record = WalRecord::Revoke {
129 ts: self.now(),
130 delegation_id: delegation_id.to_string(),
131 };
132 if let Some(wal) = self.wal.as_mut() {
133 wal.append(&record)?;
134 }
135 self.gate.revoke(delegation_id)
136 }
137
138 pub fn decide(&mut self, intent: &SpendIntent) -> Result<GateDecision, CoreError> {
145 let ts = self.now();
149 let verdict = self.gate.evaluate_at(intent, ts);
150 let spent_after = match verdict {
151 GateDecision::Allow { budget_after_cents } => budget_after_cents,
153 GateDecision::Deny { .. } => self.gate.spent_cents(&intent.delegation_id).unwrap_or(0),
154 };
155 let record = WalRecord::Decide {
156 ts,
157 decision: match verdict {
158 GateDecision::Allow { .. } => WalDecision::Allow,
159 GateDecision::Deny { .. } => WalDecision::Deny,
160 },
161 delegation_id: intent.delegation_id.clone(),
162 intent: intent.clone(),
163 reason: verdict.deny_reason(),
164 budget_after_cents: spent_after,
165 };
166 if let Some(wal) = self.wal.as_mut() {
167 wal.append(&record)?;
168 }
169 match verdict {
170 GateDecision::Allow { budget_after_cents } => {
171 let after = self.gate.commit_at(intent, ts)?;
172 debug_assert_eq!(after, budget_after_cents);
173 Ok(GateDecision::Allow {
174 budget_after_cents: after,
175 })
176 }
177 deny => Ok(deny),
178 }
179 }
180
181 pub fn state_hash(&self) -> u64 {
188 let snapshot = serde_json::json!({
189 "delegations": self.gate.delegations().collect::<Vec<_>>(),
190 "spent_cents": self.gate.ledger().entries().collect::<Vec<_>>(),
191 "revoked": self.gate.revocations().iter().collect::<Vec<_>>(),
192 "used_nonces": self.gate.replay_registry().iter().collect::<Vec<_>>(),
193 "policy_states": self.gate.policy_states().collect::<Vec<_>>(),
194 });
195 fnv1a_64(snapshot.to_string().as_bytes())
196 }
197
198 fn now(&self) -> u64 {
199 self.gate.clock().now()
200 }
201
202 pub fn replay(wal_path: impl AsRef<Path>) -> Result<Self, CoreError> {
209 let records = crate::wal::read_verified(wal_path)?.records;
211 let clock = MockClock::new(0);
212 let mut state = WanningState::new(Arc::new(clock.clone()));
213 for (line_no, record) in records {
214 let record_ts = record.ts();
215 clock.set_now(record_ts);
216 match record {
217 WalRecord::RegisterDelegation { delegation, .. } => state
218 .gate
219 .register_delegation(delegation)
220 .map_err(|e| CoreError::WalMismatch {
221 line: line_no,
222 message: format!("重放注册失败: {e}"),
223 })?,
224 WalRecord::Revoke { delegation_id, .. } => state
225 .gate
226 .revoke(&delegation_id)
227 .map_err(|e| CoreError::WalMismatch {
228 line: line_no,
229 message: format!("重放撤销失败: {e}"),
230 })?,
231 WalRecord::Decide {
232 decision,
233 intent,
234 reason,
235 budget_after_cents,
236 ..
237 } => {
238 let ts = record_ts;
242 let verdict = state.gate.evaluate_at(&intent, ts);
243 match (verdict, decision, reason) {
244 (
245 GateDecision::Allow {
246 budget_after_cents: recomputed,
247 },
248 WalDecision::Allow,
249 None,
250 ) => {
251 if recomputed != budget_after_cents {
252 return Err(CoreError::WalMismatch {
253 line: line_no,
254 message: format!(
255 "放行记录的累计消费与重算不一致:记录 {budget_after_cents} / 重算 {recomputed}"
256 ),
257 });
258 }
259 state.gate.commit_at(&intent, ts).map_err(|e| {
260 CoreError::WalMismatch {
261 line: line_no,
262 message: format!("重放扣减失败: {e}"),
263 }
264 })?;
265 }
266 (
267 GateDecision::Deny { reason: recomputed },
268 WalDecision::Deny,
269 Some(recorded_reason),
270 ) if recomputed == recorded_reason => {
271 }
273 (verdict, decision, reason) => {
274 return Err(CoreError::WalMismatch {
275 line: line_no,
276 message: format!(
277 "重算判定与记录不一致:重算 {verdict:?} / 记录 {decision:?} reason={reason:?}"
278 ),
279 });
280 }
281 }
282 }
283 }
284 }
285 Ok(state)
286 }
287}
288
289#[cfg(test)]
290mod tests {
291 use super::*;
292 use crate::clock::{Clock, MockClock};
293 use crate::gate::DenyReason;
294
295 fn tmp_wal(tag: &str) -> std::path::PathBuf {
296 use std::sync::atomic::{AtomicU64, Ordering};
297 static SEQ: AtomicU64 = AtomicU64::new(0);
298 let dir = std::env::temp_dir().join("wanning-state-tests");
299 std::fs::create_dir_all(&dir).expect("建临时目录");
300 let nanos = std::time::SystemTime::now()
302 .duration_since(std::time::UNIX_EPOCH)
303 .map(|d| d.as_nanos())
304 .unwrap_or(0);
305 dir.join(format!(
306 "{tag}-{}-{}-{nanos}.jsonl",
307 std::process::id(),
308 SEQ.fetch_add(1, Ordering::SeqCst)
309 ))
310 }
311
312 fn delegation() -> Delegation {
313 Delegation::new(
314 "d1",
315 "boss",
316 "claude-code",
317 1000,
318 1000,
319 2000,
320 "agent:claude-code",
321 )
322 }
323
324 fn long_lived_delegation() -> Delegation {
327 Delegation::new(
328 "d1",
329 "boss",
330 "claude-code",
331 1000,
332 1000,
333 SystemClock.now().checked_add(86_400).expect("有效期溢出"),
334 "agent:claude-code",
335 )
336 }
337
338 fn intent(nonce: u64, amount_cents: u64) -> SpendIntent {
339 SpendIntent::new("d1", nonce, amount_cents, "jd:shop-1", "grocery", "测试")
340 }
341
342 #[test]
343 fn allow_and_deny_are_both_recorded() {
344 let path = tmp_wal("both");
345 let clock = MockClock::new(1500);
346 let mut state = WanningState::with_wal(Arc::new(clock.clone()), &path).expect("开 WAL");
347 state.register_delegation(delegation()).expect("注册");
348
349 let a = state.decide(&intent(1, 500)).expect("判定");
351 assert!(a.is_allow());
352 let d = state.decide(&intent(2, 9000)).expect("判定");
354 assert_eq!(d.deny_reason(), Some(DenyReason::OverBudget));
355
356 let records = crate::wal::read_records(&path).expect("读回");
357 assert_eq!(records.len(), 3, "注册 + 放行 + 拒绝");
358 let (_, first_decide) = &records[1];
359 let (_, second_decide) = &records[2];
360 match (first_decide.kind(), first_decide.ts(), second_decide.kind()) {
361 ("decide", 1500, "decide") => {}
362 other => panic!("记录形状不符: {other:?}"),
363 }
364 let crate::wal::WalRecord::Decide {
366 decision,
367 reason,
368 budget_after_cents,
369 ..
370 } = second_decide
371 else {
372 panic!("第二条决策记录应是 Decide");
373 };
374 assert_eq!(*decision, WalDecision::Deny);
375 assert_eq!(*reason, Some(DenyReason::OverBudget));
376 assert_eq!(*budget_after_cents, 500, "拒绝不改账本,累计消费仍是 500");
377 }
378
379 #[test]
380 fn write_ahead_audit_failure_leaves_state_untouched() {
381 let dir = std::env::temp_dir().join("wanning-state-tests");
384 std::fs::create_dir_all(&dir).expect("建临时目录");
385 let nanos = std::time::SystemTime::now()
387 .duration_since(std::time::UNIX_EPOCH)
388 .map(|d| d.as_nanos())
389 .unwrap_or(0);
390 let path = dir.join(format!("dir-as-wal-{nanos}.jsonl"));
391 std::fs::create_dir_all(&path).expect("占位为目录");
392
393 let err = WanningState::with_wal(Arc::new(MockClock::new(1500)), &path).unwrap_err();
394 assert!(matches!(err, CoreError::WalIo(_)), "{err}");
395 }
396
397 #[test]
398 fn replay_rebuilds_state_and_is_deterministic() {
399 let path = tmp_wal("replay");
400 let clock = MockClock::new(1500);
401 let mut state = WanningState::with_wal(Arc::new(clock.clone()), &path).expect("开 WAL");
402 state.register_delegation(delegation()).expect("注册");
403 state.decide(&intent(1, 500)).expect("放行");
404 state.decide(&intent(2, 9000)).expect("超额拒");
405 state.decide(&intent(3, 100)).expect("再放行");
406 state.revoke("d1").expect("撤销");
407 state.decide(&intent(4, 100)).expect("撤销后拒");
408
409 let live_hash = state.state_hash();
410 assert_eq!(
411 state.gate().spent_cents("d1"),
412 Some(600),
413 "实时累计消费 = 500 + 100"
414 );
415
416 let replayed = WanningState::replay(&path).expect("回放");
418 let hash_once = replayed.state_hash();
419 let replayed_again = WanningState::replay(&path).expect("回放二遍");
420 let hash_twice = replayed_again.state_hash();
421
422 assert_eq!(hash_once, hash_twice, "回放两遍 hash 必相同(确定性)");
423 assert_eq!(hash_once, live_hash, "回放态必须与实时态完全一致");
424 assert_eq!(replayed.gate().spent_cents("d1"), Some(600));
425 assert!(replayed.gate().is_revoked("d1"));
426 assert!(
427 replayed
428 .gate()
429 .replay_registry()
430 .contains("agent:claude-code", 1),
431 "重放登记也必须被重建"
432 );
433 assert_eq!(replayed.wal_line_count(), None, "回放态不追加记录");
434 }
435
436 #[test]
437 fn replay_uses_recorded_ts_so_expiry_reproduces() {
438 let path = tmp_wal("expiry");
441 let clock = MockClock::new(1500);
442 let mut state = WanningState::with_wal(Arc::new(clock.clone()), &path).expect("开 WAL");
443 state.register_delegation(delegation()).expect("注册");
444 state.decide(&intent(1, 100)).expect("放行");
445 clock.set_now(2000); state.decide(&intent(2, 100)).expect("过期拒");
447
448 let replayed = WanningState::replay(&path).expect("回放");
449 assert_eq!(replayed.state_hash(), state.state_hash());
450 }
451
452 #[test]
453 fn replay_fails_closed_on_corrupted_line() {
454 let path = tmp_wal("replay-corrupt");
455 let mut state = WanningState::with_wal(Arc::new(MockClock::new(1500)), &path).expect("开");
456 state.register_delegation(delegation()).expect("注册");
457 drop(state);
458 use std::io::Write;
460 let mut f = std::fs::OpenOptions::new()
461 .append(true)
462 .open(&path)
463 .expect("开");
464 f.write_all(b"{\"kind\":\"decide\",\"ts\":1,\"dele\n")
465 .expect("追加坏行");
466 drop(f);
467
468 match WanningState::replay(&path) {
469 Err(CoreError::WalBadLine { line, .. }) => assert_eq!(line, 2),
470 other => panic!("应 fail-closed 报错,实际 {other:?}"),
471 }
472 }
473
474 #[test]
475 fn replay_fails_closed_when_record_disagrees_with_recomputation() {
476 let path = tmp_wal("replay-tampered");
478 let mut state = WanningState::with_wal(Arc::new(MockClock::new(1500)), &path).expect("开");
479 state.register_delegation(delegation()).expect("注册");
480 state.decide(&intent(1, 100)).expect("放行");
481 drop(state);
482 use std::io::Write;
486 let verified = crate::wal::read_verified(&path).expect("读已有历史");
487 let forged = crate::wal::WalLine {
488 seq: verified.records.len() as u64 + 1,
489 prev: verified.tail,
490 rec: WalRecord::Decide {
491 ts: 1500,
492 decision: WalDecision::Allow,
493 delegation_id: "d1".to_string(),
494 intent: intent(1, 100),
495 reason: None,
496 budget_after_cents: 200,
497 },
498 };
499 let mut f = std::fs::OpenOptions::new()
500 .append(true)
501 .open(&path)
502 .expect("开");
503 f.write_all(serde_json::to_string(&forged).unwrap().as_bytes())
504 .and_then(|()| f.write_all(b"\n"))
505 .expect("追加");
506 drop(f);
507
508 match WanningState::replay(&path) {
509 Err(CoreError::WalMismatch { line, message }) => {
510 assert_eq!(line, 3, "不一致要指到行");
511 assert!(message.contains("不一致"), "{message}");
512 }
513 other => panic!("篡改记录必须 fail-closed,实际 {other:?}"),
514 }
515 }
516
517 #[test]
518 fn audit_chain_tail_matches_independent_read_side_recompute() {
519 let path = tmp_wal("chain-tail");
521 let mut state = WanningState::with_wal(Arc::new(MockClock::new(1500)), &path).expect("开");
522 state.register_delegation(delegation()).expect("注册");
523 state.decide(&intent(1, 500)).expect("放行");
524 state.decide(&intent(2, 9000)).expect("超额拒");
525 state.revoke("d1").expect("撤销");
526
527 let live_tail = state.audit_chain_tail().expect("必有 WAL");
528 let verified = crate::wal::read_verified(&path).expect("读回验链");
529 assert_eq!(verified.tail, live_tail, "读侧独立重算链尾 == 实时链尾");
530 assert_eq!(
531 WanningState::replay(&path).expect("回放").state_hash(),
532 state.state_hash(),
533 "链验过后,回放对账照常成立"
534 );
535 }
536
537 #[test]
538 fn live_resuming_fails_closed_on_broken_chain() {
539 let path = tmp_wal("resume-chain");
542 {
543 let mut state =
544 WanningState::with_wal(Arc::new(MockClock::new(1500)), &path).expect("开");
545 state.register_delegation(delegation()).expect("注册");
546 state.decide(&intent(1, 100)).expect("放行");
547 state.decide(&intent(2, 9000)).expect("超额拒");
548 }
549 let mut lines = crate::wal::raw_lines(&path).expect("读 WAL");
550 let mut value: serde_json::Value = serde_json::from_str(&lines[1]).expect("行是 JSON");
551 value["rec"]["intent"]["memo"] = serde_json::json!("被改写的备注");
552 lines[1] = value.to_string();
553 std::fs::write(&path, lines.join("\n") + "\n").expect("重写 WAL");
554
555 match WanningState::live_resuming(&path) {
556 Err(CoreError::WalChainBroken { line, .. }) => {
557 assert_eq!(line, 3, "断链点 = 被改行的下一行(prev 对不上)")
558 }
559 other => panic!("链断裂必须拒启,实际 {other:?}"),
560 }
561 }
562
563 #[test]
564 fn state_hash_changes_when_state_changes() {
565 let path = tmp_wal("hash");
566 let clock = MockClock::new(1500);
567 let mut state = WanningState::with_wal(Arc::new(clock.clone()), &path).expect("开");
568 state.register_delegation(delegation()).expect("注册");
569 let h0 = state.state_hash();
570 state.decide(&intent(1, 100)).expect("放行");
571 let h1 = state.state_hash();
572 state.revoke("d1").expect("撤销");
573 let h2 = state.state_hash();
574 assert_ne!(h0, h1, "扣减后 hash 必变");
575 assert_ne!(h1, h2, "撤销后 hash 必变");
576 }
577
578 #[test]
579 fn empty_wal_replays_to_empty_state() {
580 let path = tmp_wal("empty");
581 std::fs::write(&path, "").expect("写空文件");
582 let replayed = WanningState::replay(&path).expect("空 WAL 是合法状态");
583 assert_eq!(
584 replayed.state_hash(),
585 WanningState::new(Arc::new(MockClock::new(0))).state_hash()
586 );
587 }
588
589 #[test]
594 fn live_resuming_carries_ledger_revocations_and_nonces() {
595 let path = tmp_wal("resume");
596 {
597 let clock = MockClock::new(1500);
598 let mut state = WanningState::with_wal(Arc::new(clock.clone()), &path).expect("开");
599 state
600 .register_delegation(long_lived_delegation())
601 .expect("注册");
602 state.decide(&intent(1, 500)).expect("放行");
603 state.decide(&intent(2, 100)).expect("再放行");
604 state.revoke("d1").expect("撤销");
605 } let resumed = WanningState::live_resuming(&path).expect("续跑");
608 assert_eq!(resumed.gate().spent_cents("d1"), Some(600));
610 assert!(resumed.gate().is_revoked("d1"), "撤销必须跨重启存活");
611 assert_eq!(
612 resumed.state_hash(),
613 WanningState::replay(&path).expect("回放").state_hash(),
614 "续跑态与回放态必须一致"
615 );
616 assert!(
618 resumed.gate().clock().now() > 1_700_000_000,
619 "续跑必须用系统时钟,得到 {}",
620 resumed.gate().clock().now()
621 );
622
623 let mut resumed = resumed;
626 let deny = resumed.decide(&intent(3, 100)).expect("判定");
627 assert_eq!(deny.deny_reason(), Some(DenyReason::Revoked));
628 let replay_deny = resumed.decide(&intent(1, 100)).expect("判定");
629 assert_eq!(replay_deny.deny_reason(), Some(DenyReason::Revoked));
630 assert_eq!(resumed.gate().spent_cents("d1"), Some(600), "账本不动");
631 let records = crate::wal::read_records(&path).expect("读回");
632 assert_eq!(records.len(), 6, "注册+2 放行+撤销+续跑后 2 条拒绝");
633 }
634
635 #[test]
636 fn live_resuming_on_fresh_wal_starts_empty() {
637 let path = tmp_wal("resume-fresh");
638 let mut state = WanningState::live_resuming(&path).expect("新 WAL 直接续跑=空账开张");
639 assert_eq!(
640 state.state_hash(),
641 WanningState::replay(&path).expect("回放").state_hash()
642 );
643 state
644 .register_delegation(long_lived_delegation())
645 .expect("注册");
646 assert!(state.decide(&intent(1, 100)).expect("判定").is_allow());
647 }
648
649 #[test]
650 fn live_resuming_fails_closed_on_corrupted_wal() {
651 let path = tmp_wal("resume-corrupt");
652 {
653 let mut state =
654 WanningState::with_wal(Arc::new(MockClock::new(1500)), &path).expect("开");
655 state.register_delegation(delegation()).expect("注册");
656 }
657 use std::io::Write;
658 let mut f = std::fs::OpenOptions::new()
659 .append(true)
660 .open(&path)
661 .expect("开");
662 f.write_all(b"{\"kind\":\"decide\",\"ts\":1,\"dele\n")
663 .expect("追加坏行");
664 drop(f);
665
666 match WanningState::live_resuming(&path) {
667 Err(CoreError::WalBadLine { line, .. }) => assert_eq!(line, 2),
668 other => panic!("审计损坏必须拒启,实际 {other:?}"),
669 }
670 }
671}