harmoniis-wallet 0.1.129

//! Daemon process management: start, stop, status, and the main mining loop.

use std::path::PathBuf;
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
use std::sync::Arc;

use super::protocol::MiningProtocol;
use super::sha256::Sha256Midstate;
use super::stats::{self, StatsTracker};
use super::work_unit::{NonceTable, WorkUnit};

use super::{select_backend, select_backend_for_devices, BackendChoice, MinerConfig};

fn default_wallet_root() -> PathBuf {
    if let Ok(path) = std::env::var("HARMONIIS_WALLET_ROOT") {
        let trimmed = path.trim();
        if !trimmed.is_empty() {
            return PathBuf::from(trimmed);
        }
    }
    let home = dirs_next::home_dir().unwrap_or_else(|| PathBuf::from("."));
    home.join(".harmoniis").join("wallet")
}

/// PID file path: wallet-root/miner.pid
pub fn pid_file_path() -> PathBuf {
    default_wallet_root().join("miner.pid")
}

/// Log file path: wallet-root/miner.log
pub fn log_file_path() -> PathBuf {
    default_wallet_root().join("miner.log")
}

/// Orphan log (solutions that were rejected): wallet-root/miner_orphans.log
pub fn orphan_log_path() -> PathBuf {
    default_wallet_root().join("miner_orphans.log")
}

/// Pending keep log (accepted on server but not persisted locally yet): wallet-root/miner_pending_keeps.log
pub fn pending_keep_log_path() -> PathBuf {
    default_wallet_root().join("miner_pending_keeps.log")
}

/// Pending solutions file — solutions persisted to disk BEFORE async submission.
/// If the miner crashes, these can be retried on next startup.
pub fn pending_solutions_path() -> PathBuf {
    default_wallet_root().join("miner_pending_solutions.log")
}

/// Decode the preimage's embedded `"timestamp": <number>` field.
///
/// The preimage is the base64 of a JSON prefix ending in
/// `... "timestamp": <f64>, "nonce": ...`. We base64-decode (just
/// once, on the rare error path) and regex-extract. Returns `None`
/// if anything along the way fails — the caller falls back to a
/// generic "Bad timestamp" log line.
pub(crate) fn extract_preimage_timestamp(preimage_b64: &str) -> Option<f64> {
    use base64::{engine::general_purpose::STANDARD, Engine};
    // Mining preimages can include trailing nonce+suffix bytes; only
    // the prefix block is base64 of the JSON. Decode the longest
    // prefix that's a multiple of 4 bytes — if some trailing chars
    // are non-base64 (the suffix) the decode will fail, in which case
    // we fall through and try the leading 64-char prefix block.
    let candidate = if preimage_b64.len() >= 64 {
        &preimage_b64[..(preimage_b64.len() / 4) * 4]
    } else {
        preimage_b64
    };
    let bytes = STANDARD
        .decode(candidate)
        .or_else(|_| STANDARD.decode(&preimage_b64[..64.min(preimage_b64.len())]))
        .ok()?;
    let s = std::str::from_utf8(&bytes).ok()?;
    let key = "\"timestamp\":";
    let idx = s.find(key)?;
    let rest = s[idx + key.len()..].trim_start();
    let end = rest
        .char_indices()
        .find(|(_, c)| !(c.is_ascii_digit() || *c == '.' || *c == '-' || *c == '+' || *c == 'e'))
        .map(|(i, _)| i)
        .unwrap_or(rest.len());
    rest[..end].parse::<f64>().ok()
}

#[cfg(test)]
mod extract_ts_tests {
    use super::*;

    #[test]
    fn parses_real_preimage() {
        use crate::miner::work_unit::WorkUnit;
        use webylib::Amount;
        let wu = WorkUnit::new(
            28,
            Amount::from_wats(20_000_000_000_000),
            Amount::from_wats(1_000_000_000_000),
        );
        // The miner stores the FULL base64 preimage (prefix block(s) +
        // nonces + suffix) in the persistence/orphan logs; we
        // reconstruct that here so the parser is exercised on the same
        // shape it sees in production.
        let table = crate::miner::work_unit::NonceTable::new();
        let preimage = wu.preimage_string(&table, 0, 0);
        let parsed = extract_preimage_timestamp(&preimage)
            .expect("should extract timestamp from valid preimage");
        assert!(
            (parsed - wu.timestamp).abs() < 1.0,
            "expected {}, got {parsed}",
            wu.timestamp
        );
    }
}

/// Overflow solutions — solutions that could not be submitted during burst
/// drain on shutdown. Can be retried with `hrmw webminer collect`.
pub fn overflow_solutions_path() -> PathBuf {
    default_wallet_root().join("miner_overflow_solutions.log")
}

/// Outcome of a retry pass over `miner_pending_solutions.log`.
///
/// Reads the file, checks each against `miner_pending_keeps.log`
/// (already accepted), submits unsubmitted ones to the server, and
/// inserts accepted keeps into the wallet.
#[derive(Debug, Clone, Copy, Default)]
pub struct RetryOutcome {
    pub submitted: usize,
    pub already_accepted: usize,
    /// Solutions whose embedded preimage timestamp landed outside the
    /// server's ±2 h window — typically a `pending_solutions.log`
    /// replayed long after the miner died, or a session whose local
    /// clock was wrong at mining time. These cannot be redeemed.
    pub stale_replay: usize,
    pub failed: usize,
}

/// Retry unsubmitted solutions from miner_pending_solutions.log.
/// Uses 4 parallel threads for faster submission (~28 solutions/min vs 7/min).
pub fn retry_pending_solutions(server_url: &str) -> anyhow::Result<RetryOutcome> {
    retry_pending_solutions_inner(server_url, false)
}

/// Like `retry_pending_solutions` but prints per-solution feedback to stdout.
pub fn retry_pending_solutions_verbose(server_url: &str) -> anyhow::Result<RetryOutcome> {
    retry_pending_solutions_inner(server_url, true)
}

fn retry_pending_solutions_inner(server_url: &str, verbose: bool) -> anyhow::Result<RetryOutcome> {
    use super::protocol::MiningProtocol;
    use std::collections::HashSet;
    use std::sync::atomic::{AtomicUsize, Ordering};

    let solutions_path = pending_solutions_path();
    if !solutions_path.exists() {
        return Ok(RetryOutcome::default());
    }

    let solutions_text = std::fs::read_to_string(&solutions_path)?;
    if solutions_text.trim().is_empty() {
        return Ok(RetryOutcome::default());
    }

    // Load already-accepted keep secrets to skip them.
    let keeps_path = pending_keep_log_path();
    let accepted_keeps: HashSet<String> = if keeps_path.exists() {
        std::fs::read_to_string(&keeps_path)
            .unwrap_or_default()
            .lines()
            .map(|l| l.trim().to_string())
            .filter(|l| !l.is_empty())
            .collect()
    } else {
        HashSet::new()
    };

    // Parse all entries, filter already-accepted.
    struct PendingEntry {
        preimage: String,
        hash: [u8; 32],
        keep_secret: String,
    }

    let mut pre_already = 0usize;
    let mut entries = Vec::new();
    for line in solutions_text.lines() {
        let line = line.trim();
        if line.is_empty() {
            continue;
        }
        let parts: Vec<&str> = line.split('\t').collect();
        if parts.len() < 3 {
            continue;
        }
        let keep_secret = parts[2].to_string();
        if accepted_keeps.contains(&keep_secret) {
            pre_already += 1;
            continue;
        }
        let hash_hex = parts[1].trim_start_matches("0x");
        if let Ok(b) = hex::decode(hash_hex) {
            if b.len() == 32 {
                let mut arr = [0u8; 32];
                arr.copy_from_slice(&b);
                entries.push(PendingEntry {
                    preimage: parts[0].to_string(),
                    hash: arr,
                    keep_secret,
                });
            }
        }
    }

    if entries.is_empty() {
        if pre_already > 0 {
            let _ = std::fs::write(&solutions_path, "");
        }
        return Ok(RetryOutcome {
            submitted: 0,
            already_accepted: pre_already,
            stale_replay: 0,
            failed: 0,
        });
    }

    let total = entries.len();

    // Both modes use 4 parallel threads.
    // Verbose mode additionally prints per-solution progress.
    const RETRY_THREADS: usize = 4;
    let submitted = Arc::new(AtomicUsize::new(0));
    let already = Arc::new(AtomicUsize::new(pre_already));
    let failed = Arc::new(AtomicUsize::new(0));
    let progress = Arc::new(AtomicUsize::new(0));
    let entries = Arc::new(entries);
    let keeps_path = Arc::new(keeps_path);
    let server_url = server_url.to_string();

    let stale_replay = Arc::new(AtomicUsize::new(0));
    let orphan_path = Arc::new(orphan_log_path());

    let mut handles = Vec::new();
    let chunk_size = total.div_ceil(RETRY_THREADS);

    for t in 0..RETRY_THREADS {
        let start = t * chunk_size;
        let end = (start + chunk_size).min(total);
        if start >= total {
            break;
        }
        let entries = entries.clone();
        let submitted = submitted.clone();
        let already = already.clone();
        let failed = failed.clone();
        let stale_replay = stale_replay.clone();
        let progress = progress.clone();
        let keeps_path = keeps_path.clone();
        let orphan_path = orphan_path.clone();
        let server_url = server_url.clone();

        handles.push(std::thread::spawn(move || {
            let proto = match MiningProtocol::new(&server_url) {
                Ok(p) => p,
                Err(_) => return,
            };
            for i in start..end {
                let entry = &entries[i];
                let n = progress.fetch_add(1, Ordering::Relaxed) + 1;
                match proto.submit_report_blocking(&entry.preimage, &entry.hash) {
                    Ok(_) => {
                        submitted.fetch_add(1, Ordering::Relaxed);
                        let _ = std::fs::OpenOptions::new()
                            .create(true)
                            .append(true)
                            .open(&*keeps_path)
                            .and_then(|mut f| {
                                use std::io::Write;
                                writeln!(f, "{}", entry.keep_secret)
                            });
                        if verbose {
                            let preview = &entry.keep_secret[..entry.keep_secret.len().min(24)];
                            println!("  [{n}/{total}] Submitted  {preview}...");
                        }
                    }
                    Err(e) => {
                        let msg = e.to_string();
                        if msg.contains("Didn't use a new secret") {
                            already.fetch_add(1, Ordering::Relaxed);
                            // Server confirms this keep_secret was already
                            // accepted on a prior submission (often from a
                            // cloud session where a 504 timeout made the
                            // client think it failed even though the server
                            // committed it). Persist it to the keeps log so
                            // future dedup recognizes it locally and we
                            // don't round-trip the server again next run.
                            let _ = std::fs::OpenOptions::new()
                                .create(true)
                                .append(true)
                                .open(&*keeps_path)
                                .and_then(|mut f| {
                                    use std::io::Write;
                                    writeln!(f, "{}", entry.keep_secret)
                                });
                            if verbose {
                                println!("  [{n}/{total}] Already accepted");
                            }
                        } else if msg.contains("Bad timestamp") {
                            // The preimage was mined more than ~2 h ago
                            // (or the mining-time clock differed >2 h
                            // from the server) — it cannot be re-mined,
                            // re-stamped, or recovered. Move it to the
                            // orphan log with a `stale_replay` reason
                            // so the operator has the diagnostic trail
                            // and stop counting it as a hard failure.
                            stale_replay.fetch_add(1, Ordering::Relaxed);
                            let _ = std::fs::OpenOptions::new()
                                .create(true)
                                .append(true)
                                .open(&*orphan_path)
                                .and_then(|mut f| {
                                    use std::io::Write;
                                    writeln!(
                                        f,
                                        "{}\t0x{}\t{}\treason=stale_replay",
                                        entry.preimage,
                                        hex::encode(entry.hash),
                                        entry.keep_secret
                                    )
                                });
                            if verbose {
                                println!(
                                    "  [{n}/{total}] Stale (timestamp outside server ±2h window) — moved to orphans"
                                );
                            }
                        } else {
                            failed.fetch_add(1, Ordering::Relaxed);
                            if verbose {
                                println!("  [{n}/{total}] Failed: {msg}");
                            }
                        }
                    }
                }
            }
        }));
    }

    for h in handles {
        let _ = h.join();
    }

    let s = submitted.load(Ordering::Relaxed);
    let a = already.load(Ordering::Relaxed);
    let r = stale_replay.load(Ordering::Relaxed);
    let f = failed.load(Ordering::Relaxed);

    // Stale-replay entries are unrecoverable — remove them from the
    // pending log so they don't get re-tried forever. Genuine failures
    // are kept for the next attempt.
    if s > 0 || a > 0 || r > 0 {
        let _ = std::fs::write(&solutions_path, "");
    }
    if r > 0 && verbose {
        println!(
            "  ({r} solution(s) had timestamps outside the server's ±2h window — \
             these were mined too long ago to redeem and were moved to {})",
            orphan_path.display()
        );
    }

    Ok(RetryOutcome {
        submitted: s,
        already_accepted: a,
        stale_replay: r,
        failed: f,
    })
}

/// Check if a miner process is already running.
pub fn is_running() -> Option<u32> {
    let pid_path = pid_file_path();
    if !pid_path.exists() {
        return None;
    }
    let pid_str = std::fs::read_to_string(&pid_path).ok()?;
    let pid: u32 = pid_str.trim().parse().ok()?;

    // Check if process is alive
    #[cfg(unix)]
    {
        // kill -0 checks if process exists without sending a signal
        let result = unsafe { libc::kill(pid as i32, 0) };
        if result == 0 {
            Some(pid)
        } else {
            // Stale PID file — clean up
            let _ = std::fs::remove_file(&pid_path);
            None
        }
    }
    #[cfg(not(unix))]
    {
        // Verify the process is actually alive via `tasklist`.
        let alive = std::process::Command::new("tasklist")
            .args(["/FI", &format!("PID eq {pid}"), "/NH"])
            .stdout(std::process::Stdio::piped())
            .stderr(std::process::Stdio::null())
            .output()
            .ok()
            .map(|o| String::from_utf8_lossy(&o.stdout).contains(&pid.to_string()))
            .unwrap_or(false);
        if alive {
            Some(pid)
        } else {
            let _ = std::fs::remove_file(&pid_path);
            None
        }
    }
}

/// Start the miner as a detached background process.
pub fn start(config: &MinerConfig) -> anyhow::Result<()> {
    if let Some(pid) = is_running() {
        anyhow::bail!("miner already running (PID {})", pid);
    }

    // Ensure ~/.harmoniis/ exists
    let dir = pid_file_path().parent().unwrap().to_path_buf();
    std::fs::create_dir_all(&dir)?;

    let log_file = std::fs::File::create(log_file_path())?;
    let log_err = log_file.try_clone()?;

    let exe = std::env::current_exe()?;
    let mut cmd = std::process::Command::new(exe);
    cmd.arg("webminer")
        .arg("run")
        .arg("--server")
        .arg(&config.server_url)
        .arg("--backend")
        .arg(config.backend.as_cli_str())
        .arg("--max-difficulty")
        .arg(config.max_difficulty.to_string());

    if let Some(cpu_threads) = config.cpu_threads {
        cmd.arg("--cpu-threads").arg(cpu_threads.to_string());
    }

    if config.backend == BackendChoice::Cpu {
        cmd.arg("--cpu-only");
    }
    if config.accept_terms {
        cmd.arg("--accept-terms");
    }
    if let Some(ref devices) = config.devices {
        let s: Vec<String> = devices.iter().map(|d| d.to_string()).collect();
        cmd.arg("--device").arg(s.join(","));
    }

    // Pass wallet paths
    cmd.arg("--wallet")
        .arg(&config.wallet_path)
        .arg("--webcash-wallet")
        .arg(&config.webcash_wallet_path);

    cmd.stdout(log_file)
        .stderr(log_err)
        .stdin(std::process::Stdio::null());

    // Detach child process from parent's console/session.
    #[cfg(unix)]
    {
        use std::os::unix::process::CommandExt;
        unsafe {
            cmd.pre_exec(|| {
                libc::setsid();
                Ok(())
            });
        }
    }
    #[cfg(windows)]
    {
        use std::os::windows::process::CommandExt;
        const CREATE_NO_WINDOW: u32 = 0x0800_0000;
        const DETACHED_PROCESS: u32 = 0x0000_0008;
        cmd.creation_flags(CREATE_NO_WINDOW | DETACHED_PROCESS);
    }

    let child = cmd.spawn()?;
    let pid = child.id();

    std::fs::write(pid_file_path(), pid.to_string())?;
    println!("Miner started (PID: {})", pid);
    println!("Log: {}", log_file_path().display());

    Ok(())
}

/// Stop the running miner.
pub fn stop() -> anyhow::Result<()> {
    let pid = match is_running() {
        Some(pid) => pid,
        None => {
            println!("No miner running.");
            return Ok(());
        }
    };

    #[cfg(unix)]
    {
        unsafe {
            libc::kill(pid as i32, libc::SIGTERM);
        }
    }
    #[cfg(not(unix))]
    {
        let _ = std::process::Command::new("taskkill")
            .args(["/PID", &pid.to_string()])
            .stdout(std::process::Stdio::null())
            .stderr(std::process::Stdio::null())
            .status();
    }

    // Wait for process to exit (up to 5 seconds)
    for _ in 0..50 {
        if is_running().is_none() {
            break;
        }
        std::thread::sleep(std::time::Duration::from_millis(100));
    }

    // Clean up PID file
    let _ = std::fs::remove_file(pid_file_path());

    if is_running().is_some() {
        println!(
            "Miner (PID {}) did not stop gracefully, may need manual kill.",
            pid
        );
    } else {
        println!("Miner stopped (PID {}).", pid);
    }

    Ok(())
}

/// Show miner status.
pub fn status() -> anyhow::Result<()> {
    match is_running() {
        Some(pid) => {
            println!("Miner running (PID: {})", pid);

            let status_path = stats::status_file_path();
            if status_path.exists() {
                let json = std::fs::read_to_string(&status_path)?;
                let s: stats::MinerStats = serde_json::from_str(&json)?;
                println!("  Backend:    {}", s.backend);
                println!(
                    "  Hash rate:  {}",
                    stats::format_hash_rate(s.hash_rate_mhs * 1_000_000.0)
                );
                println!("  Attempts:   {}", s.total_attempts);
                println!(
                    "  Solutions:  {} found, {} accepted",
                    s.solutions_found, s.solutions_accepted
                );
                println!("  Difficulty: {}", s.difficulty);
                println!("  Uptime:     {}s", s.uptime_secs);
                let hps = s.hash_rate_mhs * 1_000_000.0;
                println!("  ETA:        {}", stats::estimate_time(hps, s.difficulty));
            } else {
                println!("  (no stats available yet)");
            }
        }
        None => {
            println!("No miner running.");
        }
    }
    Ok(())
}

/// The actual mining loop (called by `hrmw webminer run`).
pub async fn run_mining_loop(config: MinerConfig) -> anyhow::Result<()> {
    println!("Webcash miner starting...");
    println!("Server: {}", config.server_url);

    // Set up SIGTERM handler
    let shutdown = Arc::new(AtomicBool::new(false));
    {
        let shutdown = shutdown.clone();
        signal_hook::flag::register(signal_hook::consts::SIGTERM, shutdown.clone())?;
        signal_hook::flag::register(signal_hook::consts::SIGINT, shutdown)?;
    }

    // Write PID file (for the `run` process itself)
    let pid = std::process::id();
    std::fs::write(pid_file_path(), pid.to_string())?;

    // Select backend
    let backend = if let Some(ref devices) = config.devices {
        select_backend_for_devices(devices).await?
    } else {
        select_backend(config.backend, config.cpu_threads).await?
    };
    let chunk_size = backend.max_batch_hint();
    let pipeline_depth = backend.recommended_pipeline_depth().clamp(1, 1024);
    println!("Mining setup:");
    println!("  backend_mode={}", config.backend.as_cli_str());
    println!("  backend_name={}", backend.name());
    println!("  nonce_chunk_size={}", chunk_size);
    println!("  workunit_pipeline_depth={}", pipeline_depth);
    println!(
        "  cpu_system_threads={}",
        std::thread::available_parallelism()
            .map(|n| n.get())
            .unwrap_or(1)
    );
    for line in backend.startup_summary() {
        println!("  {}", line);
    }

    // Initialize protocol client (Arc-shared with background submitter)
    let protocol = Arc::new(MiningProtocol::new(&config.server_url)?);

    // Initialize stats
    let tracker = Arc::new(StatsTracker::new(backend.name()));
    let status_path = stats::status_file_path();

    let pending_keep_path = pending_keep_log_path();
    println!("Webcash wallet: {}", config.webcash_wallet_path.display());
    println!("Pending keep log: {}", pending_keep_path.display());

    // Initialize nonce table (shared across all work units)
    let nonce_table = NonceTable::new();

    // Fetch initial target. `get_target()` also seeds `clock_skew` from
    // the response `Date:` header, so by the time we build the first
    // WorkUnit the embedded preimage timestamp is anchored to server
    // time — the user's local clock no longer has to be correct.
    println!("Fetching mining target...");
    let mut target = protocol.get_target().await?;
    tracker.set_difficulty(target.difficulty);
    println!(
        "  difficulty={} mining_amount={} subsidy={} epoch={}",
        target.difficulty, target.mining_amount, target.subsidy_amount, target.epoch
    );

    // Loud warning when local clock is far off the server. Mining still
    // works (we anchor the preimage timestamp), but the user almost
    // certainly wants to fix their NTP or RTC — and it explains why a
    // user on v0.1.128 or earlier saw every fresh solution rejected as
    // "Bad timestamp" before this anchor was added.
    {
        let skew = super::clock_skew::current_skew_secs();
        if super::clock_skew::has_observed() {
            if skew.abs() >= 30 * 60 {
                eprintln!(
                    "⚠ Local clock is {} vs server. Mining will work (timestamps \
                     anchored to server), but please sync your system clock — Date \
                     & Time → Set automatically, or `sudo ntpdate pool.ntp.org`.",
                    super::clock_skew::format_skew(skew),
                );
            } else if skew.abs() >= 30 {
                println!(
                    "Clock skew vs server: {} (within ±2h window)",
                    super::clock_skew::format_skew(skew)
                );
            }
        } else {
            eprintln!(
                "Note: server did not return a Date header — preimage timestamps \
                 will use the local clock. If solutions get rejected as \
                 \"Bad timestamp\", run `hrmw webminer doctor` to verify."
            );
        }
    }

    let mut last_stats_print = std::time::Instant::now();
    let target_refresh_interval = std::time::Duration::from_secs(15);
    let stats_print_interval = std::time::Duration::from_secs(5);
    let mut work_unit_timer;
    let mut pending_work_units: Option<Vec<WorkUnit>> = None;
    let shared_target = Arc::new(std::sync::RwLock::new(target.clone()));

    // ── N independent submitter clients, created EAGERLY in main process ──
    //
    // CRITICAL: reqwest::blocking::Client MUST be created here in the main
    // process context. Creating it in a subprocess or lazily in threads fails
    // (server hangs, zero bytes returned). This was the root cause of the
    // subprocess reporter failure — documented in git history (a9ade36, ab55786).
    //
    // Each client = own TCP connection. Server sees N independent miners,
    // processes them in parallel. N scales with GPU count.
    // Override with HRMW_REPORTER_CLIENTS env var.
    // Single reporter client — the webcash.org server processes mining reports
    // sequentially (~6s each, single-threaded Tornado). Multiple connections
    // provide zero throughput benefit and add threads that steal CPU from mining.
    let _num_clients: usize = 1;
    let reporter_client = Arc::new(
        reqwest::blocking::Client::builder()
            .timeout(std::time::Duration::from_secs(120))
            .build()
            .expect("failed to build blocking HTTP client"),
    );

    // Pre-warm: establish TCP+TLS connection before mining starts.
    {
        let url = format!("{}/api/v1/target", config.server_url);
        match reporter_client.get(&url).send() {
            Ok(_) => eprintln!("[reporter] connection warm"),
            Err(e) => eprintln!("[reporter] warmup failed: {e}"),
        }
    }

    struct SolutionMsg {
        preimage: String,
        hash: [u8; 32],
        keep_secret: String,
        difficulty_achieved: u32,
        // The difficulty value embedded in the preimage JSON at WorkUnit
        // creation time. Locked into the SHA256 input — cannot be refreshed
        // post-mining. The server rejects when `committed_difficulty <
        // current_target` (under the "Bad timestamp" wording), so the
        // reporter must short-circuit instead of paying ~27s per remote
        // rejection.
        committed_difficulty: u32,
    }
    let (solution_tx, solution_rx) = std::sync::mpsc::channel::<SolutionMsg>();
    let queue_depth = Arc::new(AtomicUsize::new(0));
    let drain_accepted = Arc::new(AtomicUsize::new(0));
    let drain_skipped = Arc::new(AtomicUsize::new(0));
    let drain_failed = Arc::new(AtomicUsize::new(0));
    // EWMA of recent accept latency in milliseconds. Seeded at 8000ms (the
    // documented healthy steady-state from SESSION-HANDLER), updated by the
    // reporter on each accept. Used by the mining loop to size the
    // forward-dated timestamp offset proportional to projected queue wait.
    let measured_accept_ms = Arc::new(std::sync::atomic::AtomicU64::new(8000));

    // Background wallet insertion — separate OS thread inserts keeps into wallet
    // via /api/v1/replace immediately after each report. Runs sequentially,
    // one insert at a time, does not block the reporter thread.
    let (wallet_tx, wallet_rx) = std::sync::mpsc::channel::<String>();
    let wallet_inserted = Arc::new(AtomicUsize::new(0));
    // Wallet insertion on separate OS thread with its OWN HTTP client.
    // /replace and /mining_report are different server endpoints — they
    // don't share a bottleneck. Both run in parallel, independently.
    let wallet_thread = {
        let wallet_path = config.webcash_wallet_path.clone();
        let inserted = wallet_inserted.clone();
        std::thread::Builder::new()
            .name("wallet-insert".into())
            .spawn(move || {
                let mut rt: Option<tokio::runtime::Runtime> = None;
                while let Ok(keep_secret) = wallet_rx.recv() {
                    let rt = rt.get_or_insert_with(|| {
                        tokio::runtime::Builder::new_current_thread()
                            .enable_all()
                            .build()
                            .expect("wallet runtime")
                    });
                    if let Ok(secret) = webylib::SecretWebcash::parse(&keep_secret) {
                        match rt.block_on(async {
                            let wallet = webylib::Wallet::open(&wallet_path).await?;
                            wallet.insert(secret).await
                        }) {
                            Ok(_) => {
                                let n = inserted.fetch_add(1, Ordering::Relaxed) + 1;
                                eprintln!("[wallet] inserted #{n}");
                            }
                            Err(e) if e.to_string().contains("UNIQUE constraint") => {
                                inserted.fetch_add(1, Ordering::Relaxed);
                            }
                            Err(e) => eprintln!("[wallet] insert failed: {e}"),
                        }
                    }
                }
                eprintln!("[wallet] exiting");
            })
            .expect("failed to spawn wallet thread")
    };

    let reporter_handle = {
        let client = reporter_client.clone();
        let server_url = config.server_url.clone();
        let sub_tracker = tracker.clone();
        let sub_pending_keep = pending_keep_path.clone();
        let sub_shared_target = shared_target.clone();
        let sub_queue_depth = queue_depth.clone();
        let sub_wallet_tx = wallet_tx.clone();
        let sub_drain_accepted = drain_accepted.clone();
        let sub_drain_skipped = drain_skipped.clone();
        let sub_drain_failed = drain_failed.clone();
        let sub_measured_accept_ms = measured_accept_ms.clone();

        std::thread::Builder::new()
            .name("reporter".into())
            .spawn(move || {
                eprintln!("[reporter] started");
                while let Ok(msg) = solution_rx.recv() {
                    sub_queue_depth.fetch_sub(1, Ordering::Relaxed);

                    let current_diff = sub_shared_target.read().unwrap().difficulty;
                    // (1) hash leading-zero count too low for current target.
                    if msg.difficulty_achieved < current_diff {
                        eprintln!(
                            "[reporter] skipping stale (achieved {} < {})",
                            msg.difficulty_achieved, current_diff
                        );
                        sub_drain_skipped.fetch_add(1, Ordering::Relaxed);
                        let _ = std::fs::OpenOptions::new()
                            .create(true)
                            .append(true)
                            .open(orphan_log_path())
                            .and_then(|mut f| {
                                use std::io::Write;
                                writeln!(
                                    f,
                                    "{}\t0x{}\t{}\tdifficulty={}\tcommitted={}",
                                    msg.preimage,
                                    hex::encode(msg.hash),
                                    msg.keep_secret,
                                    msg.difficulty_achieved,
                                    msg.committed_difficulty
                                )
                            });
                        continue;
                    }
                    // (2) preimage's committed difficulty value pre-dates the
                    // current target. The hash itself may be strong enough,
                    // but the server rejects on `committed < current` — and
                    // the value is part of the SHA256 input, so we cannot
                    // re-stamp it. Skip in microseconds instead of paying
                    // ~27s per server rejection.
                    if msg.committed_difficulty < current_diff {
                        eprintln!(
                            "[reporter] skipping stale (committed {} < {})",
                            msg.committed_difficulty, current_diff
                        );
                        sub_drain_skipped.fetch_add(1, Ordering::Relaxed);
                        let _ = std::fs::OpenOptions::new()
                            .create(true)
                            .append(true)
                            .open(orphan_log_path())
                            .and_then(|mut f| {
                                use std::io::Write;
                                writeln!(
                                    f,
                                    "{}\t0x{}\t{}\tdifficulty={}\tcommitted={}",
                                    msg.preimage,
                                    hex::encode(msg.hash),
                                    msg.keep_secret,
                                    msg.difficulty_achieved,
                                    msg.committed_difficulty
                                )
                            });
                        continue;
                    }

                    let t0 = std::time::Instant::now();
                    match MiningProtocol::submit_report_with_client(
                        &client,
                        &server_url,
                        &msg.preimage,
                        &msg.hash,
                    ) {
                        Ok(resp) => {
                            let ms = t0.elapsed().as_millis();
                            sub_tracker.record_accepted();
                            sub_drain_accepted.fetch_add(1, Ordering::Relaxed);
                            // EWMA(accept_ms): 25% weight on the new sample.
                            let prev = sub_measured_accept_ms.load(Ordering::Relaxed);
                            let next = ((prev as u128 * 3 + ms as u128) / 4) as u64;
                            sub_measured_accept_ms.store(next, Ordering::Relaxed);
                            eprintln!("[reporter] accepted in {ms}ms");

                            if let Some(new_diff) = resp.difficulty_target {
                                let mut t = sub_shared_target.write().unwrap();
                                if new_diff != t.difficulty {
                                    println!(
                                        "Difficulty adjustment: {} → {}",
                                        t.difficulty, new_diff
                                    );
                                    t.difficulty = new_diff;
                                    sub_tracker.set_difficulty(new_diff);
                                }
                            }

                            let _ = std::fs::OpenOptions::new()
                                .create(true)
                                .append(true)
                                .open(&sub_pending_keep)
                                .and_then(|mut f| {
                                    use std::io::Write;
                                    writeln!(f, "{}", msg.keep_secret)
                                });

                            // Send to wallet thread for immediate insert.
                            let _ = sub_wallet_tx.send(msg.keep_secret.clone());
                        }
                        Err(e) => {
                            let ms = t0.elapsed().as_millis();
                            let err = e.to_string();
                            if err.contains("Didn't use a new secret") {
                                sub_drain_accepted.fetch_add(1, Ordering::Relaxed);
                                let _ = std::fs::OpenOptions::new()
                                    .create(true)
                                    .append(true)
                                    .open(&sub_pending_keep)
                                    .and_then(|mut f| {
                                        use std::io::Write;
                                        writeln!(f, "{}", msg.keep_secret)
                                    });
                                let _ = sub_wallet_tx.send(msg.keep_secret.clone());
                            } else if err.contains("Bad timestamp") {
                                // The committed-difficulty branch above
                                // already short-circuits before submit,
                                // so a server-side "Bad timestamp" here
                                // means the embedded preimage timestamp
                                // is outside the server's ±2 h window.
                                // Decode it and tell the user *why* —
                                // either the local clock was off when
                                // this was mined (skew not yet seeded
                                // on first boot), or the solution sat
                                // in the queue past the 2 h cap.
                                let ts = extract_preimage_timestamp(&msg.preimage);
                                let server_now = super::clock_skew::server_now_secs_f64();
                                let skew = super::clock_skew::current_skew_secs();
                                match ts {
                                    Some(t) => {
                                        let delta = (t - server_now) as i64;
                                        eprintln!(
                                            "[reporter] FAILED in {ms}ms: Bad timestamp \
                                             (preimage_ts={t:.0}, server_now≈{server_now:.0}, \
                                             delta={}, clock_skew={}). \
                                             Solution lost — see {} for diagnosis.",
                                            super::clock_skew::format_skew(delta),
                                            super::clock_skew::format_skew(skew),
                                            orphan_log_path().display(),
                                        );
                                    }
                                    None => {
                                        eprintln!(
                                            "[reporter] FAILED in {ms}ms: Bad timestamp \
                                             (could not parse preimage; clock_skew={}). \
                                             Run `hrmw webminer doctor` to verify clock.",
                                            super::clock_skew::format_skew(skew),
                                        );
                                    }
                                }
                                sub_drain_failed.fetch_add(1, Ordering::Relaxed);
                                let _ = std::fs::OpenOptions::new()
                                    .create(true)
                                    .append(true)
                                    .open(orphan_log_path())
                                    .and_then(|mut f| {
                                        use std::io::Write;
                                        writeln!(
                                            f,
                                            "{}\t0x{}\t{}\tdifficulty={}\tcommitted={}\treason=bad_timestamp",
                                            msg.preimage,
                                            hex::encode(msg.hash),
                                            msg.keep_secret,
                                            msg.difficulty_achieved,
                                            msg.committed_difficulty
                                        )
                                    });
                            } else {
                                eprintln!("[reporter] FAILED in {ms}ms: {err}");
                                sub_drain_failed.fetch_add(1, Ordering::Relaxed);
                                let _ = std::fs::OpenOptions::new()
                                    .create(true)
                                    .append(true)
                                    .open(orphan_log_path())
                                    .and_then(|mut f| {
                                        use std::io::Write;
                                        writeln!(
                                            f,
                                            "{}\t0x{}\t{}\tdifficulty={}\tcommitted={}",
                                            msg.preimage,
                                            hex::encode(msg.hash),
                                            msg.keep_secret,
                                            msg.difficulty_achieved,
                                            msg.committed_difficulty
                                        )
                                    });
                            }
                        }
                    }
                }
                eprintln!("[reporter] exiting");
            })
            .expect("failed to spawn reporter thread")
    };
    // Background target refresher — polls server every 15s without ever
    // blocking the mining loop. Updates are picked up atomically next cycle.
    {
        let refresh_target = shared_target.clone();
        let refresh_protocol = protocol.clone();
        let refresh_tracker = tracker.clone();
        let refresh_status = status_path.clone();
        let refresh_shutdown = shutdown.clone();
        tokio::spawn(async move {
            let mut interval = tokio::time::interval(target_refresh_interval);
            interval.tick().await; // skip first immediate tick
            loop {
                interval.tick().await;
                if refresh_shutdown.load(Ordering::Relaxed) {
                    break;
                }
                match refresh_protocol.get_target().await {
                    Ok(new_target) => {
                        let mut t = refresh_target.write().unwrap();
                        if new_target.difficulty != t.difficulty {
                            println!(
                                "Difficulty changed: {} -> {}",
                                t.difficulty, new_target.difficulty
                            );
                        }
                        *t = new_target;
                        refresh_tracker.set_difficulty(t.difficulty);
                    }
                    Err(e) => eprintln!("Warning: failed to fetch target: {e}"),
                }
                let _ = refresh_tracker.write_to_file(&refresh_status);
            }
        });
    }

    // Main mining loop — ZERO network I/O, never blocks.
    while !shutdown.load(Ordering::Relaxed) {
        // Read latest target atomically (non-blocking RwLock read).
        target = shared_target.read().unwrap().clone();

        // Skip if difficulty exceeds our max
        if target.difficulty > config.max_difficulty {
            println!(
                "Difficulty {} exceeds max {}, waiting...",
                target.difficulty, config.max_difficulty
            );
            tokio::time::sleep(std::time::Duration::from_secs(5)).await;
            continue;
        }

        let t_cycle = std::time::Instant::now();

        // Use pre-built batch from previous cycle if available and difficulty
        // hasn't changed, otherwise create fresh with rayon parallelism.
        // ── Adaptive forward-dating ─────────────────────────────────────
        // The preimage timestamp is locked into the SHA256 input at WorkUnit
        // creation. To absorb queue pressure inside the server's ±2h timestamp
        // window (per maaku/webminer server.cc:1314-1319), we forward-date
        // each new batch by the projected wait time of solutions reaching the
        // reporter — but ONLY when mining outpaces reporting. With ETA per
        // solution ≥ accept latency, the queue stays naturally bounded and
        // forward-dating is unnecessary; we keep the embedded timestamp at
        // wall-clock now (no behavior change vs pre-0.1.125).
        //
        // Offset cap: 115 minutes — just under the +2h forward bound, ~5 min
        // safety margin. Combined with the symmetric backward window, this
        // gives a 230-minute total in-flight validity from mining to submit.
        const MAX_FORWARD_OFFSET_SECS: f64 = 115.0 * 60.0;
        const THROTTLE_TRIGGER_PCT: f64 = 0.95;
        let qd = queue_depth.load(Ordering::Relaxed) as f64;
        let accept_secs = (measured_accept_ms.load(Ordering::Relaxed) as f64) / 1000.0;
        let snapshot_for_eta = tracker.snapshot();
        let hash_rate_hs = snapshot_for_eta.hash_rate_mhs * 1_000_000.0;
        let mining_eta_secs = if hash_rate_hs > 1.0 && target.difficulty > 0 {
            2.0_f64.powi(target.difficulty as i32) / hash_rate_hs
        } else {
            f64::INFINITY
        };
        // Only forward-date when mining outpaces reporting — the regime where
        // the queue would otherwise grow unbounded. At ETA ≥ accept latency,
        // the system is at or below balance and `now` is fine.
        let imbalanced = mining_eta_secs < accept_secs;
        let projected_wait_secs = if imbalanced { qd * accept_secs } else { 0.0 };
        let actual_offset_secs = projected_wait_secs.min(MAX_FORWARD_OFFSET_SECS);
        let throttle_deficit_secs = (projected_wait_secs - MAX_FORWARD_OFFSET_SECS).max(0.0);

        // GPU throttle (energy saver, last resort). When projected wait
        // exceeds 95% of cap, sleep up to one accept cycle so the queue can
        // drain before we mine more solutions we cannot redeem in time.
        if projected_wait_secs > MAX_FORWARD_OFFSET_SECS * THROTTLE_TRIGGER_PCT {
            let sleep_secs = throttle_deficit_secs.min(accept_secs).max(1.0);
            eprintln!(
                "[throttle] queue={} offset_cap_reached, sleeping {:.1}s (accept_ewma={:.1}s, eta={:.1}s)",
                qd as usize, sleep_secs, accept_secs, mining_eta_secs
            );
            tokio::time::sleep(std::time::Duration::from_secs_f64(sleep_secs)).await;
            continue;
        }

        // Server-anchored "now" + forward-dating offset. Reading the
        // skew is a single relaxed atomic load (lock-free, ~1ns), so
        // this stays zero-cost on the mining hot path. The skew is
        // refreshed by the target poll task and the reporter thread —
        // never by the mining loop.
        let timestamp_secs = super::clock_skew::server_now_secs_f64() + actual_offset_secs;

        let work_units = match pending_work_units.take() {
            Some(pending) if !pending.is_empty() && pending[0].difficulty == target.difficulty => {
                pending
            }
            _ => {
                let d = target.difficulty;
                let a = target.mining_amount;
                let s = target.subsidy_amount;
                let n = pipeline_depth;
                let ts = timestamp_secs;
                tokio::task::spawn_blocking(move || {
                    use rayon::prelude::*;
                    (0..n)
                        .into_par_iter()
                        .map(|_| WorkUnit::new_with_timestamp(d, a, s, Some(ts)))
                        .collect()
                })
                .await?
            }
        };

        // Build midstate refs — Arc avoids cloning 32 Sha256Midstates per cycle.
        let midstates: Arc<Vec<Sha256Midstate>> =
            Arc::new(work_units.iter().map(|wu| wu.midstate.clone()).collect());

        // Start creating NEXT batch in background with rayon (overlapped with
        // GPU mining). On a 122-thread machine, 32 WUs finish in ~30μs.
        let next_d = target.difficulty;
        let next_a = target.mining_amount;
        let next_s = target.subsidy_amount;
        let next_n = pipeline_depth;
        let next_ts = timestamp_secs;
        let next_batch_handle = tokio::task::spawn_blocking(move || {
            use rayon::prelude::*;
            (0..next_n)
                .into_par_iter()
                .map(|_| WorkUnit::new_with_timestamp(next_d, next_a, next_s, Some(next_ts)))
                .collect::<Vec<_>>()
        });

        // Mine current batch on GPUs (overlapped with next batch creation).
        work_unit_timer = std::time::Instant::now();
        let chunks = backend
            .mine_work_units(&midstates, &nonce_table, target.difficulty, None)
            .await?;
        let mine_us = work_unit_timer.elapsed().as_micros();

        // Collect pre-built next batch (should be ready by now).
        pending_work_units = next_batch_handle.await.ok();

        let cycle_us = t_cycle.elapsed().as_micros();
        let mut attempts_this_work_unit = 0u64;
        for chunk in &chunks {
            attempts_this_work_unit = attempts_this_work_unit.saturating_add(chunk.attempted);
        }

        tracker.add_attempts(attempts_this_work_unit);

        // Print stats periodically using rolling average (not per-cycle).
        if last_stats_print.elapsed() >= stats_print_interval {
            let snapshot = tracker.snapshot();
            let hps = snapshot.hash_rate_mhs * 1_000_000.0;
            let expected_solutions = if target.difficulty > 0 {
                let denom = 2.0_f64.powi(target.difficulty as i32);
                snapshot.total_attempts as f64 / denom
            } else {
                0.0
            };
            let p_zero_pct = (-expected_solutions).exp() * 100.0;
            let qd = queue_depth.load(Ordering::Relaxed);
            let offset_min = actual_offset_secs / 60.0;
            let accept_ewma_s = (measured_accept_ms.load(Ordering::Relaxed) as f64) / 1000.0;
            let skew_label = super::clock_skew::format_skew(super::clock_skew::current_skew_secs());
            println!(
                "speed={} difficulty={} solutions={}/{} pending={} eta={} expected={:.2} p0={:.2}% offset={:.1}min accept_ewma={:.1}s clock_skew={} (mine={}μs cycle={}μs)",
                stats::format_hash_rate(hps),
                target.difficulty,
                snapshot.solutions_accepted,
                snapshot.solutions_found,
                qd,
                stats::estimate_time(hps, target.difficulty),
                expected_solutions,
                p_zero_pct,
                offset_min,
                accept_ewma_s,
                skew_label,
                mine_us,
                cycle_us,
            );
            if offset_min > MAX_FORWARD_OFFSET_SECS / 60.0 * 0.8 {
                eprintln!(
                    "⚠ Forward offset at {:.0}% of {:.0}-min cap — GPU throttle imminent",
                    offset_min / (MAX_FORWARD_OFFSET_SECS / 60.0) * 100.0,
                    MAX_FORWARD_OFFSET_SECS / 60.0
                );
            }
            last_stats_print = std::time::Instant::now();
        }

        for (wu, chunk) in work_units.into_iter().zip(chunks.into_iter()) {
            if let Some(solution) = chunk.result {
                tracker.record_solution();
                let preimage =
                    wu.preimage_string(&nonce_table, solution.nonce1_idx, solution.nonce2_idx);

                println!(
                    "SOLUTION FOUND! difficulty={} hash=0x{}",
                    solution.difficulty_achieved,
                    hex::encode(solution.hash)
                );

                // Persist to disk FIRST (crash safety), then queue for
                // background submission on the dedicated OS thread.
                let pending_line = format!(
                    "{}\t0x{}\t{}\tdifficulty={}\n",
                    preimage,
                    hex::encode(solution.hash),
                    wu.keep_secret,
                    solution.difficulty_achieved,
                );
                let _ = std::fs::OpenOptions::new()
                    .create(true)
                    .append(true)
                    .open(pending_solutions_path())
                    .and_then(|mut f| {
                        use std::io::Write;
                        f.write_all(pending_line.as_bytes())
                    });

                // Dispatch to shared work-stealing queue.
                queue_depth.fetch_add(1, Ordering::Relaxed);
                let _ = solution_tx.send(SolutionMsg {
                    preimage,
                    hash: solution.hash,
                    keep_secret: wu.keep_secret.to_string(),
                    difficulty_achieved: solution.difficulty_achieved,
                    committed_difficulty: wu.difficulty,
                });
            }
        }
    }

    // Graceful shutdown: close the channel, let reporter threads drain remaining
    // solutions, then join all threads. Monitor drain progress every second.
    let pending = queue_depth.load(Ordering::Relaxed);
    let snapshot = tracker.snapshot();
    println!(
        "Miner shutting down — draining {pending} pending (stale skip in μs, fresh ~6s each, 600s drain timeout)"
    );
    println!(
        "  Session totals: found={} accepted={} pending={}",
        snapshot.solutions_found, snapshot.solutions_accepted, pending
    );
    let baseline_accepted = drain_accepted.load(Ordering::Relaxed);
    let baseline_skipped = drain_skipped.load(Ordering::Relaxed);
    let baseline_failed = drain_failed.load(Ordering::Relaxed);
    let drain_start = std::time::Instant::now();
    drop(solution_tx); // Signal EOF — threads will drain remaining and exit.

    let drain_timeout = std::time::Duration::from_secs(600);
    loop {
        let remaining = queue_depth.load(Ordering::Relaxed);
        if remaining == 0 {
            break;
        }
        if drain_start.elapsed() > drain_timeout {
            eprintln!("⚠ Drain timeout (600s) — {remaining} solutions LOST (server too slow)");
            break;
        }
        let drained_accepted = drain_accepted
            .load(Ordering::Relaxed)
            .saturating_sub(baseline_accepted);
        let drained_skipped = drain_skipped
            .load(Ordering::Relaxed)
            .saturating_sub(baseline_skipped);
        let drained_failed = drain_failed
            .load(Ordering::Relaxed)
            .saturating_sub(baseline_failed);
        println!(
            "  Draining: {remaining} remaining ({drained_accepted} accepted, {drained_skipped} skipped past-target, {drained_failed} other failures)"
        );
        std::thread::sleep(std::time::Duration::from_secs(2));
    }

    let _ = reporter_handle.join();
    let drain_secs = drain_start.elapsed().as_secs();
    let final_snap = tracker.snapshot();
    println!(
        "  Drain complete in {drain_secs}s — accepted={} found={}",
        final_snap.solutions_accepted, final_snap.solutions_found
    );

    // Wait for wallet thread to finish remaining inserts.
    drop(wallet_tx);
    println!(
        "Waiting for wallet insertions ({} done so far)...",
        wallet_inserted.load(Ordering::Relaxed)
    );
    let _ = wallet_thread.join();
    println!(
        "  Wallet complete: {} inserted",
        wallet_inserted.load(Ordering::Relaxed)
    );

    // Clear pending_solutions.log — solutions are either accepted (in wallet)
    // or lost (stale timestamps). Keeping them causes "Bad timestamp" errors
    // on future collect attempts. The keeps log is the source of truth.
    let _ = std::fs::write(pending_solutions_path(), "");

    let _ = std::fs::remove_file(pid_file_path());
    let _ = tracker.write_to_file(&status_path);
    println!("Miner stopped.");

    Ok(())
}