Use current Kimi Desktop session and recheck it before each request

CHANGELOG.md

@@ -9,6 +9,7 @@
 
 ## [Windows] Unreleased
 
+- Kimi (China): use the current Kimi Desktop session from its Local Storage (`%APPDATA%\kimi-desktop`) for web usage and monthly membership enrichment, and check that session again before each request, so a signed-out, replaced, or expired Desktop token is never sent or recovered from an older record (upstream #4416, #3536).
 - Codex CLI: add opt-in `cost --identified-only` request subtotals across local and SSH (`--remote`) histories, deduplicating strong request identities and keeping partial results without estimating a combined price, plus `--group-by project` reports that show the partial-history notice even when no projects are listed (upstream #4447, #4472).
 - OpenCode Go: reuse unchanged local history scans (keyed by `opencode.db` and its `-wal` sidecar size and modification time, at most one 4 MiB payload kept), and in Auto overlay Console quota from a stored or automatically discovered browser session before falling back to the local estimate; cookie source Off never reads a browser (upstream #4410, #4445).
 - Add **Parallel** prepaid balance: reads the account balance and any pending debit from the platform.parallel.ai console using your browser session or a pasted Cookie header. A signed-out console moves on to the next browser session and reports an expired session when none remain (upstream #4479, #4508).

rust/src/browser/leveldb/mod.rs

@@ -16,6 +16,10 @@
 //! - Files larger than [`MAX_FILE_BYTES`], scans larger than [`MAX_TOTAL_SCAN_BYTES`], and blocks
 //!   that decode beyond [`MAX_BLOCK_BYTES`] are skipped so a corrupt profile cannot exhaust
 //!   memory or monopolize a refresh.
+//!
+//! [`read_current_value_strict`] is the stricter single-key variant for credentials whose stale
+//! value must never be resurrected: it fails instead of skipping any file or block, and it
+//! rejects symlinked entries and a directory that changes while it is read.
 
 pub mod local_storage;
 mod log;
@@ -66,6 +70,8 @@ pub enum LevelDbError {
     Io(#[from] std::io::Error),
     #[error("LevelDB directory exceeds safe scan limits")]
     ResourceLimit,
+    #[error("LevelDB directory could not be read completely and consistently")]
+    Incomplete,
 }
 
 /// A put (`value: Some`) or delete (`value: None`) with the sequence number it was written at.
@@ -225,6 +231,115 @@ pub(crate) fn read_entries_with_budget(
     Ok((entries.into_entries(), scanned_bytes))
 }
 
+/// Newest live value of exactly `key` in the LevelDB directory `dir`, or `None` when the key is
+/// absent or its newest record is a delete.
+///
+/// A skipped file or block could hold a newer record or tombstone for `key`, so unlike
+/// [`read_entries`] every problem fails the whole read with [`LevelDbError::Incomplete`]: a
+/// missing `CURRENT`, a symlinked entry, an oversized or unreadable file, a malformed table or
+/// data block, a torn log record, or a directory listing whose names, sizes or modification
+/// times differ after the read. The `MANIFEST` is still not consulted and checksums are not
+/// verified, so this narrows rather than eliminates the window for a concurrent writer.
+pub(crate) fn read_current_value_strict(
+    dir: &Path,
+    key: &[u8],
+) -> Result<Option<Vec<u8>>, LevelDbError> {
+    let before = directory_fingerprint(dir)?;
+    if !before.iter().any(|file| file.name == "CURRENT") {
+        return Err(LevelDbError::Incomplete);
+    }
+    let mut entries = EntryAccumulator {
+        key_prefix: Some(key),
+        ..EntryAccumulator::default()
+    };
+    let mut scanned_bytes = 0u64;
+    for file in &before {
+        let path = dir.join(&file.name);
+        let Some(kind) = FileKind::of(&path) else {
+            continue;
+        };
+        let remaining_bytes = MAX_TOTAL_SCAN_BYTES.saturating_sub(scanned_bytes);
+        let data = match open_no_follow(&path)
+            .map_err(BoundedReadError::Io)
+            .and_then(|file| read_bounded_open_file(file, remaining_bytes))
+        {
+            Ok(data) => data,
+            Err(BoundedReadError::BudgetExceeded) => return Err(LevelDbError::ResourceLimit),
+            Err(BoundedReadError::FileTooLarge | BoundedReadError::Io(_)) => {
+                return Err(LevelDbError::Incomplete);
+            }
+        };
+        let file_bytes = u64::try_from(data.len()).map_err(|_| LevelDbError::ResourceLimit)?;
+        scanned_bytes = scanned_bytes
+            .checked_add(file_bytes)
+            .ok_or(LevelDbError::ResourceLimit)?;
+        let complete = match kind {
+            FileKind::Log => log::read_log_until(&data, &mut |record| entries.absorb(record)),
+            FileKind::Table => matches!(
+                table::read_table_until(&data, &mut |record| entries.absorb(record)),
+                Ok((0, true))
+            ),
+        };
+        if entries.resource_limit_exceeded {
+            return Err(LevelDbError::ResourceLimit);
+        }
+        if !complete {
+            return Err(LevelDbError::Incomplete);
+        }
+    }
+    if directory_fingerprint(dir)? != before {
+        return Err(LevelDbError::Incomplete);
+    }
+    Ok(entries.newest.remove(key).and_then(|(_, value)| value))
+}
+
+#[derive(PartialEq, Eq)]
+struct FileFingerprint {
+    name: std::ffi::OsString,
+    len: u64,
+    modified: Option<std::time::SystemTime>,
+}
+
+fn directory_fingerprint(dir: &Path) -> Result<Vec<FileFingerprint>, LevelDbError> {
+    let mut files = Vec::new();
+    for (count, dir_entry) in std::fs::read_dir(dir)?.enumerate() {
+        if count == MAX_DIRECTORY_ENTRIES {
+            return Err(LevelDbError::ResourceLimit);
+        }
+        let dir_entry = dir_entry.map_err(|_| LevelDbError::Incomplete)?;
+        // `DirEntry::metadata` does not traverse links; junctions count as symlinks on Windows.
+        let metadata = dir_entry.metadata().map_err(|_| LevelDbError::Incomplete)?;
+        if metadata.file_type().is_symlink() {
+            return Err(LevelDbError::Incomplete);
+        }
+        files.push(FileFingerprint {
+            name: dir_entry.file_name(),
+            len: metadata.len(),
+            modified: metadata.modified().ok(),
+        });
+    }
+    files.sort_by(|left, right| left.name.cmp(&right.name));
+    Ok(files)
+}
+
+/// Open `path` for reading without following a link swapped in after the directory listing.
+fn open_no_follow(path: &Path) -> std::io::Result<std::fs::File> {
+    let mut options = std::fs::OpenOptions::new();
+    options.read(true);
+    #[cfg(windows)]
+    {
+        use std::os::windows::fs::OpenOptionsExt;
+        const FILE_FLAG_OPEN_REPARSE_POINT: u32 = 0x0020_0000;
+        options.custom_flags(FILE_FLAG_OPEN_REPARSE_POINT);
+    }
+    let file = options.open(path)?;
+    let metadata = file.metadata()?;
+    if metadata.file_type().is_symlink() || !metadata.is_file() {
+        return Err(std::io::Error::other("not a regular file"));
+    }
+    Ok(file)
+}
+
 #[derive(Clone, Copy)]
 enum FileKind {
     Log,
@@ -252,6 +367,13 @@ enum BoundedReadError {
 
 fn read_bounded_file(path: &Path, remaining_bytes: u64) -> Result<Vec<u8>, BoundedReadError> {
     let file = std::fs::File::open(path).map_err(BoundedReadError::Io)?;
+    read_bounded_open_file(file, remaining_bytes)
+}
+
+fn read_bounded_open_file(
+    file: std::fs::File,
+    remaining_bytes: u64,
+) -> Result<Vec<u8>, BoundedReadError> {
     let size = file.metadata().map_err(BoundedReadError::Io)?.len();
     if size > MAX_FILE_BYTES {
         return Err(BoundedReadError::FileTooLarge);

rust/src/browser/leveldb/tests.rs

@@ -11,7 +11,7 @@ use super::local_storage::{
 };
 use super::snappy::{self, SnappyError};
 use super::table::{TableError, read_table};
-use super::{Entry, Record, log, read_entries};
+use super::{Entry, LevelDbError, Record, log, read_current_value_strict, read_entries};
 
 // ---------------------------------------------------------------------------------------------
 // Fixture builders
@@ -519,6 +519,87 @@ fn read_entries_reports_a_missing_directory() {
     assert!(read_entries(&dir.path().join("absent")).is_err());
 }
 
+fn strict_dir(files: &[(&str, Vec<u8>)]) -> tempfile::TempDir {
+    let dir = tempfile::tempdir().unwrap();
+    std::fs::write(dir.path().join("CURRENT"), b"MANIFEST-000001\n").unwrap();
+    for (name, data) in files {
+        std::fs::write(dir.path().join(name), data).unwrap();
+    }
+    dir
+}
+
+#[test]
+fn strict_read_returns_the_newest_value_of_the_exact_key() {
+    let table: &[TableRecord] = &[(b"token", 1, 1, b"old"), (b"token-other", 2, 1, b"x")];
+    let dir = strict_dir(&[
+        ("000002.ldb", write_table(&[(table, Compression::None)])),
+        (
+            "000003.log",
+            write_log(&[write_batch(5, &[Op::Put(b"token", b"new")])]),
+        ),
+    ]);
+    assert_eq!(
+        read_current_value_strict(dir.path(), b"token").unwrap(),
+        Some(b"new".to_vec())
+    );
+    assert_eq!(
+        read_current_value_strict(dir.path(), b"absent").unwrap(),
+        None
+    );
+}
+
+#[test]
+fn strict_read_honors_a_newer_delete() {
+    let dir = strict_dir(&[(
+        "000003.log",
+        write_log(&[
+            write_batch(1, &[Op::Put(b"token", b"old")]),
+            write_batch(2, &[Op::Delete(b"token")]),
+        ]),
+    )]);
+    assert_eq!(
+        read_current_value_strict(dir.path(), b"token").unwrap(),
+        None
+    );
+}
+
+#[test]
+fn strict_read_fails_instead_of_skipping_a_file() {
+    let good = write_log(&[write_batch(1, &[Op::Put(b"token", b"old")])]);
+    let mut torn = write_log(&[write_batch(2, &[Op::Delete(b"token")])]);
+    torn.truncate(torn.len() - 1);
+    let cases = [
+        vec![
+            ("000003.log", good.clone()),
+            ("000004.ldb", b"not a table".to_vec()),
+        ],
+        vec![("000003.log", good.clone()), ("000004.log", torn)],
+    ];
+    for files in cases {
+        let dir = strict_dir(&files);
+        assert!(matches!(
+            read_current_value_strict(dir.path(), b"token"),
+            Err(LevelDbError::Incomplete)
+        ));
+        // The best-effort reader still returns the stale value from the readable file.
+        assert_eq!(read_entries(dir.path()).unwrap().len(), 1);
+    }
+}
+
+#[test]
+fn strict_read_requires_current() {
+    let dir = tempfile::tempdir().unwrap();
+    std::fs::write(
+        dir.path().join("000003.log"),
+        write_log(&[write_batch(1, &[Op::Put(b"token", b"v")])]),
+    )
+    .unwrap();
+    assert!(matches!(
+        read_current_value_strict(dir.path(), b"token"),
+        Err(LevelDbError::Incomplete)
+    ));
+}
+
 // ---------------------------------------------------------------------------------------------
 // Chromium Local Storage layer
 // ---------------------------------------------------------------------------------------------

rust/src/providers/kimi/code_api.rs

@@ -2,7 +2,8 @@
 //! upstream 0.48.0 enrichment (#2622): Code API / CLI usage snapshots are
 //! merged with the monthly membership pool + Code 7-day limit when a
 //! `kimi.com` web token is available (manual cookie → Kimi Desktop session →
-//! browser import; gated by Cookie Source Off).
+//! browser import; gated by Cookie Source Off). A Desktop session is checked
+//! again before each enrichment request (upstream #4416).
 
 use reqwest::Url;
 use std::path::{Path, PathBuf};
@@ -85,16 +86,19 @@ pub(crate) async fn fetch_via_code_api(
     snapshot.login_method = Some(plan_name.unwrap_or_else(|| login_method.to_string()));
 
     // Upstream #2622: enrich Code API + CLI usage with the monthly membership
-    // pool from a signed-in Kimi Desktop (or browser/manual) session.
-    for web_token in web::web_auth_tokens(ctx.manual_cookie_header.as_deref(), region) {
-        match web::fetch_subscription_for_enrichment_result(&client, &web_token, region).await {
+    // pool from a signed-in Kimi Desktop (or browser/manual) session. A
+    // Desktop session that changed before dispatch ends enrichment but keeps
+    // the Code quota snapshot (upstream #4416).
+    for web_credential in web::web_auth_credentials(ctx.manual_cookie_header.as_deref(), region) {
+        match web::fetch_subscription_for_enrichment_result(&client, &web_credential, region).await
+        {
             Ok(subscription) => {
                 if let Some(subscription) = subscription {
                     snapshot = super::apply_subscription_windows(snapshot, &subscription);
                 }
                 if !has_plan_name
                     && let Some(plan) =
-                        web::fetch_subscription_plan(&client, &web_token, region).await
+                        web::fetch_subscription_plan(&client, &web_credential, region).await
                 {
                     snapshot.login_method = Some(plan);
                 }