diff --git a/crates/leader-election/src/elector.rs b/crates/leader-election/src/elector.rs
index 37e3e99..1d23bde 100644
--- a/crates/leader-election/src/elector.rs
+++ b/crates/leader-election/src/elector.rs
@@ -269,7 +269,7 @@ impl<L: Lock> LeaderElector<L> {
             let record = self.build_record(now).await;
             match self.lock.update(record.clone()).await {
                 Ok(()) => {
-                    self.update_observed(record, now).await;
+                    self.set_observed_record(record, now).await;
                     return true;
                 }
                 Err(e) => {
@@ -282,7 +282,7 @@ impl<L: Lock> LeaderElector<L> {
         // Slow path: Get current record
         let old_record = match self.lock.get().await {
             Ok(Some(r)) => {
-                self.update_observed(r.clone(), now).await;
+                self.observe_record(r.clone(), now).await;
                 r
             }
             Ok(None) => {
@@ -291,7 +291,7 @@ impl<L: Lock> LeaderElector<L> {
                 return match self.lock.create(record.clone()).await {
                     Ok(()) => {
                         debug!(identity = %self.config.identity, "created new lease");
-                        self.update_observed(record, now).await;
+                        self.set_observed_record(record, now).await;
                         true
                     }
                     Err(e) => {
@@ -308,7 +308,7 @@ impl<L: Lock> LeaderElector<L> {
 
         // Check if we can take over
         if !old_record.holder_identity.is_empty()
-            && self.is_lease_valid_with(&old_record, now)
+            && self.is_observed_lease_valid(&old_record, now).await
             && !self.is_leader_with(&old_record)
         {
             // Lease is held by someone else and still valid — give up this attempt
@@ -328,7 +328,7 @@ impl<L: Lock> LeaderElector<L> {
 
         match self.lock.update(record.clone()).await {
             Ok(()) => {
-                self.update_observed(record, now).await;
+                self.set_observed_record(record, now).await;
                 true
             }
             Err(e) => {
@@ -403,11 +403,24 @@ impl<L: Lock> LeaderElector<L> {
         }
     }
 
-    /// Check if a specific record's lease is still valid relative to now.
-    fn is_lease_valid_with(&self, record: &LeaderElectionRecord, now: DateTime<Utc>) -> bool {
-        let duration = Duration::from_secs(record.lease_duration_seconds as u64);
-        // Use renew_time as the basis for validity (the last time the lease was refreshed)
-        now < record.renew_time + duration
+    /// Check if a specific record's lease is still valid based on local observation time.
+    async fn is_observed_lease_valid(
+        &self,
+        record: &LeaderElectionRecord,
+        now: DateTime<Utc>,
+    ) -> bool {
+        let observed = self.observed.read().await;
+        if observed.record.as_ref() != Some(record) {
+            return false;
+        }
+
+        match observed.observed_time {
+            Some(observed_time) => {
+                let duration = Duration::from_secs(record.lease_duration_seconds as u64);
+                now < observed_time + duration
+            }
+            None => false,
+        }
     }
 
     /// Build a new election record for this identity.
@@ -446,8 +459,19 @@ impl<L: Lock> LeaderElector<L> {
             .unwrap_or(false)
     }
 
-    /// Update observed state after a successful get or update.
-    async fn update_observed(&self, record: LeaderElectionRecord, now: DateTime<Utc>) {
+    /// Observe a record fetched from the lock backend.
+    async fn observe_record(&self, record: LeaderElectionRecord, now: DateTime<Utc>) {
+        let mut observed = self.observed.write().await;
+        if observed.record.as_ref() != Some(&record) {
+            observed.record = Some(record);
+            observed.observed_time = Some(now);
+        } else if observed.observed_time.is_none() {
+            observed.observed_time = Some(now);
+        }
+    }
+
+    /// Record a successful local create/update acknowledged by the lock backend.
+    async fn set_observed_record(&self, record: LeaderElectionRecord, now: DateTime<Utc>) {
         let mut observed = self.observed.write().await;
         observed.record = Some(record);
         observed.observed_time = Some(now);
@@ -608,25 +632,57 @@ mod tests {
         assert!(!elector.is_leader_with(&record_other));
     }
 
-    #[test]
-    fn test_is_lease_valid_with() {
+    #[tokio::test]
+    async fn test_observed_lease_valid_uses_local_observed_time() {
         let config = test_config("node-1");
         let elector = LeaderElector::new(config, DummyLock::new("node-1"), SystemClock).unwrap();
         let now = Utc::now();
 
-        // Valid: renew_time + duration is in the future
         let record = LeaderElectionRecord {
-            holder_identity: "node-1".to_string(),
+            holder_identity: "node-2".to_string(),
+            lease_duration_seconds: 15,
+            acquire_time: now - chrono::Duration::seconds(30),
+            renew_time: now - chrono::Duration::seconds(30),
+            leader_transitions: 0,
+        };
+        elector.observe_record(record.clone(), now).await;
+
+        assert!(elector.is_observed_lease_valid(&record, now).await);
+        assert!(
+            elector
+                .is_observed_lease_valid(&record, now + chrono::Duration::seconds(14))
+                .await
+        );
+        assert!(
+            !elector
+                .is_observed_lease_valid(&record, now + chrono::Duration::seconds(16))
+                .await
+        );
+    }
+
+    #[tokio::test]
+    async fn test_same_observed_record_does_not_refresh_observed_time() {
+        let config = test_config("node-1");
+        let elector = LeaderElector::new(config, DummyLock::new("node-1"), SystemClock).unwrap();
+        let now = Utc::now();
+
+        let record = LeaderElectionRecord {
+            holder_identity: "node-2".to_string(),
             lease_duration_seconds: 15,
             acquire_time: now,
             renew_time: now,
             leader_transitions: 0,
         };
-        assert!(elector.is_lease_valid_with(&record, now));
-
-        // Expired: check time is past renew_time + duration
-        let future = now + chrono::Duration::seconds(20);
-        assert!(!elector.is_lease_valid_with(&record, future));
+        elector.observe_record(record.clone(), now).await;
+        elector
+            .observe_record(record.clone(), now + chrono::Duration::seconds(10))
+            .await;
+
+        assert!(
+            !elector
+                .is_observed_lease_valid(&record, now + chrono::Duration::seconds(16))
+                .await
+        );
     }
 
     // ─── Dummy lock for unit tests ────────────────────────────────────
diff --git a/crates/leader-election/src/record.rs b/crates/leader-election/src/record.rs
index a1fac29..8f5a50c 100644
--- a/crates/leader-election/src/record.rs
+++ b/crates/leader-election/src/record.rs
@@ -17,7 +17,7 @@
 use chrono::{DateTime, Utc};
 
 /// Record stored in the Lease object, representing the current election state.
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, PartialEq, Eq)]
 pub struct LeaderElectionRecord {
     /// Identity of the current leader (empty if released).
     pub holder_identity: String,
diff --git a/crates/leader-election/tests/integration_tests.rs b/crates/leader-election/tests/integration_tests.rs
index e856a67..bee4e1c 100644
--- a/crates/leader-election/tests/integration_tests.rs
+++ b/crates/leader-election/tests/integration_tests.rs
@@ -30,6 +30,7 @@ use kube_leader_election::{
 // ─── Test Helpers ───────────────────────────────────────────────────────────
 
 /// Fake lock with controllable behavior for testing.
+#[derive(Clone)]
 struct FakeLock {
     identity: String,
     record: Arc<RwLock<Option<LeaderElectionRecord>>>,
@@ -116,6 +117,10 @@ impl MockClock {
             now: Arc::new(RwLock::new(time)),
         }
     }
+
+    async fn set(&self, time: DateTime<Utc>) {
+        *self.now.write().await = time;
+    }
 }
 
 impl Clock for MockClock {
@@ -318,13 +323,15 @@ async fn test_acquire_no_holder() {
     );
 }
 
-/// Test 3: Acquire when lease is expired.
+/// Test 3: Acquire only after the locally observed lease window expires.
 #[tokio::test]
 async fn test_acquire_expired_lease() {
     let lock = FakeLock::new("node-2");
+    let now = Utc::now();
 
-    // Set an expired lease held by another node
-    let expired_time = Utc::now() - chrono::Duration::seconds(30);
+    // The remote renew_time is old, but a candidate must not use that
+    // remote wall clock to take over immediately.
+    let expired_time = now - chrono::Duration::seconds(30);
     let expired_record = LeaderElectionRecord {
         holder_identity: "node-1".to_string(),
         lease_duration_seconds: 15,
@@ -334,7 +341,8 @@ async fn test_acquire_expired_lease() {
     };
     lock.set_record(expired_record).await;
 
-    let clock = MockClock::new(Utc::now());
+    let clock = MockClock::new(now);
+    let test_clock = clock.clone();
     let config = test_config("node-2");
     let callbacks = Arc::new(TestCallbacks::new());
 
@@ -345,13 +353,21 @@ async fn test_acquire_expired_lease() {
     let cb = callbacks.clone();
     let handle = tokio::spawn(async move { elector.run(SharedCallbacks(cb), cancel_clone).await });
 
-    tokio::time::sleep(Duration::from_millis(200)).await;
+    tokio::time::sleep(Duration::from_millis(150)).await;
+    assert_eq!(
+        callbacks.started_count().await,
+        0,
+        "should not acquire immediately from stale remote renew_time"
+    );
+
+    test_clock.set(now + chrono::Duration::seconds(16)).await;
+    tokio::time::sleep(Duration::from_millis(250)).await;
     cancel.cancel();
 
     let result = tokio::time::timeout(Duration::from_secs(2), handle).await;
     assert!(result.is_ok(), "elector should complete");
 
-    // node-2 should have acquired the expired lease
+    // node-2 should acquire after the local observed lease duration elapses.
     assert_eq!(
         callbacks.started_count().await,
         1,
@@ -400,6 +416,79 @@ async fn test_acquire_active_lease() {
     let _ = tokio::time::timeout(Duration::from_secs(2), handle).await;
 }
 
+#[tokio::test]
+async fn test_clock_skewed_candidate_waits_while_remote_record_changes() {
+    let lock = FakeLock::new("shared");
+    let remote_start = Utc::now();
+    let candidate_start = remote_start + chrono::Duration::seconds(60);
+    let leader_record = LeaderElectionRecord {
+        holder_identity: "node-1".to_string(),
+        lease_duration_seconds: 15,
+        acquire_time: remote_start,
+        renew_time: remote_start,
+        leader_transitions: 0,
+    };
+    lock.set_record(leader_record.clone()).await;
+
+    let clock = MockClock::new(candidate_start);
+    let test_clock = clock.clone();
+    let mut config = test_config("node-2");
+    config.release_on_cancel = false;
+    let callbacks = Arc::new(TestCallbacks::new());
+
+    let elector = LeaderElector::new(config, lock.clone(), clock).unwrap();
+    let cancel = CancellationToken::new();
+    let cancel_clone = cancel.clone();
+
+    let cb = callbacks.clone();
+    let handle = tokio::spawn(async move { elector.run(SharedCallbacks(cb), cancel_clone).await });
+
+    tokio::time::sleep(Duration::from_millis(150)).await;
+    assert_eq!(
+        callbacks.started_count().await,
+        0,
+        "skewed candidate should not acquire immediately"
+    );
+
+    for step in 1..=3 {
+        test_clock
+            .set(candidate_start + chrono::Duration::seconds(step * 10))
+            .await;
+        lock.set_record(LeaderElectionRecord {
+            renew_time: remote_start + chrono::Duration::seconds(step * 10),
+            ..leader_record.clone()
+        })
+        .await;
+        tokio::time::sleep(Duration::from_millis(150)).await;
+        assert_eq!(
+            callbacks.started_count().await,
+            0,
+            "candidate should not acquire while renewals keep changing the record"
+        );
+    }
+
+    test_clock
+        .set(candidate_start + chrono::Duration::seconds(46))
+        .await;
+    tokio::time::sleep(Duration::from_millis(250)).await;
+    cancel.cancel();
+
+    let result = tokio::time::timeout(Duration::from_secs(2), handle).await;
+    assert!(result.is_ok(), "elector should complete");
+    assert_eq!(
+        callbacks.started_count().await,
+        1,
+        "candidate should acquire after renewals stop and local observation expires"
+    );
+
+    let final_record = lock
+        .get()
+        .await
+        .expect("fake lock get should succeed")
+        .expect("record should exist");
+    assert_eq!(final_record.holder_identity, "node-2");
+}
+
 /// Test 5: Successful renewal.
 #[tokio::test]
 async fn test_renew_success() {
diff --git a/src/status.rs b/src/status.rs
index 52860f5..8c30b4b 100644
--- a/src/status.rs
+++ b/src/status.rs
@@ -606,32 +606,48 @@ fn redact_sensitive_pairs(message: &str) -> String {
     }
 
     fn parse_value_end(input: &str, start: usize) -> usize {
-        let bytes = input.as_bytes();
-        if start >= bytes.len() {
+        if start >= input.len() {
             return start;
         }
-        let first = bytes[start];
-        if first == b'"' || first == b'\'' {
-            let quote = first;
-            let mut index = start + 1;
-            while index < bytes.len() {
-                if bytes[index] == quote && bytes.get(index.wrapping_sub(1)) != Some(&b'\\') {
-                    return index + 1;
+
+        let mut chars = input[start..].char_indices();
+        let Some((_, first)) = chars.next() else {
+            return start;
+        };
+        if first == '"' || first == '\'' {
+            let mut previous = first;
+            for (offset, ch) in chars {
+                if ch == first && previous != '\\' {
+                    return start + offset + ch.len_utf8();
                 }
-                index += 1;
+                previous = ch;
             }
-            return bytes.len();
+            return input.len();
         }
 
-        let mut index = start;
-        while index < bytes.len() {
-            let ch = bytes[index] as char;
+        for (offset, ch) in input[start..].char_indices() {
             if ch.is_whitespace() || matches!(ch, ',' | ';' | '}' | ']' | ')') {
-                break;
+                return start + offset;
             }
-            index += 1;
         }
-        index
+        input.len()
+    }
+
+    fn skip_whitespace(input: &str, start: usize) -> usize {
+        for (offset, ch) in input[start..].char_indices() {
+            if !ch.is_whitespace() {
+                return start + offset;
+            }
+        }
+        input.len()
+    }
+
+    fn matches_key_at(message: &str, start: usize, key: &str) -> bool {
+        let end = start + key.len();
+        end <= message.len()
+            && message.is_char_boundary(start)
+            && message.is_char_boundary(end)
+            && message[start..end].eq_ignore_ascii_case(key)
     }
 
     fn redacted_value(original: &str) -> String {
@@ -657,12 +673,11 @@ fn redact_sensitive_pairs(message: &str) -> String {
         for key in SENSITIVE_KEYS {
             let key_len = key.len();
 
-            let unquoted_match = cursor + key_len <= bytes.len()
-                && message[cursor..cursor + key_len].eq_ignore_ascii_case(key);
+            let unquoted_match = matches_key_at(message, cursor, key);
             let quoted_match = cursor + key_len + 2 <= bytes.len()
                 && matches!(bytes[cursor] as char, '"' | '\'')
                 && bytes[cursor + key_len + 1] == bytes[cursor]
-                && message[cursor + 1..cursor + 1 + key_len].eq_ignore_ascii_case(key);
+                && matches_key_at(message, cursor + 1, key);
 
             let (key_start, key_end, cursor_after_key) = if unquoted_match {
                 if cursor > 0 {
@@ -681,18 +696,12 @@ fn redact_sensitive_pairs(message: &str) -> String {
 
             let candidate = &message[key_start..key_end];
 
-            let mut sep_index = cursor_after_key;
-            while sep_index < bytes.len() && (bytes[sep_index] as char).is_whitespace() {
-                sep_index += 1;
-            }
+            let sep_index = skip_whitespace(message, cursor_after_key);
             if sep_index >= bytes.len() || !matches!(bytes[sep_index] as char, '=' | ':') {
                 continue;
             }
 
-            let mut value_start = sep_index + 1;
-            while value_start < bytes.len() && (bytes[value_start] as char).is_whitespace() {
-                value_start += 1;
-            }
+            let value_start = skip_whitespace(message, sep_index + 1);
             let value_end = parse_value_end(message, value_start);
             if value_end <= value_start {
                 continue;
@@ -711,8 +720,11 @@ fn redact_sensitive_pairs(message: &str) -> String {
         }
 
         if !matched {
-            output.push(bytes[cursor] as char);
-            cursor += 1;
+            let Some(ch) = message[cursor..].chars().next() else {
+                break;
+            };
+            output.push(ch);
+            cursor += ch.len_utf8();
         }
     }
 
@@ -1032,6 +1044,44 @@ mod tests {
         assert!(!sanitized.contains("SECRET_JSON"));
     }
 
+    #[test]
+    fn sanitize_message_handles_unicode_without_panicking() {
+        let message = "错误🔐 token: tok_123 用户=测试 secretkey: SK_TEST 完成";
+
+        let sanitized = sanitize_message(message);
+
+        assert!(sanitized.contains("错误🔐"));
+        assert!(sanitized.contains("用户=测试"));
+        assert!(sanitized.contains("完成"));
+        assert!(sanitized.contains("token: <redacted>"));
+        assert!(sanitized.contains("secretkey: <redacted>"));
+        assert!(!sanitized.contains("tok_123"));
+        assert!(!sanitized.contains("SK_TEST"));
+    }
+
+    #[test]
+    fn sanitize_message_redacts_unicode_quoted_values() {
+        let message = "{\"说明\":\"🔐\",\"secretkey\":\"秘密值\"}";
+
+        let sanitized = sanitize_message(message);
+
+        assert!(sanitized.contains("\"说明\":\"🔐\""));
+        assert!(sanitized.contains("\"secretkey\":\"<redacted>\""));
+        assert!(!sanitized.contains("秘密值"));
+    }
+
+    #[test]
+    fn sanitize_message_redacts_after_unicode_whitespace() {
+        let message = "token:\u{3000}tok_123 secretkey:\u{2003}SK_TEST";
+
+        let sanitized = sanitize_message(message);
+
+        assert!(sanitized.contains("token:\u{3000}<redacted>"));
+        assert!(sanitized.contains("secretkey:\u{2003}<redacted>"));
+        assert!(!sanitized.contains("tok_123"));
+        assert!(!sanitized.contains("SK_TEST"));
+    }
+
     fn condition(type_: &str, status: &str, reason: &str) -> Condition {
         Condition {
             type_: type_.to_string(),