use std::collections::{BTreeMap, BTreeSet, BinaryHeap, HashMap, HashSet};
use std::fmt::Write as _; use std::fs::{self, File, OpenOptions};
use std::io::{BufRead, BufReader, BufWriter, Seek, SeekFrom, Write};
use std::path::{Path, PathBuf};
use anyhow::{Context, Result};
use clap::{Parser, Subcommand};
use serde::{Deserialize, Serialize};
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::{Arc, Mutex, mpsc};
use shogiesa_core::{
Board, GameOutcome, GamePhase, GameResultInfo, Observation, PieceType, PositionRecord,
PositionTags, QualityConfig, QualityDecision, SCHEMA_VERSION, Score, ScorePerspective,
SearchLimitKind, SideToMove, SourceInfo, UsiMove, bestmove_agreement,
cp_from_black_perspective, effective_bestmove_kind, engine_bestmove_agreement,
evaluate_quality, has_special_bestmove, parse_usi_move, phase_from_ply,
requested_depth_underreached, score_swing, sfen::Sfen, zobrist_from_sfen,
};
use shogiesa_pack as pack;
use shogiesa_stratify::{EvalBucket, bucket_key, eval_bucket_of};
use shogiesa_usi::{SearchLimit, UsiEngine, UsiError};
use stratifykit_core::{
BucketStatus, HeapEntry, QuotaSpec, TotalF32, bucket_floor, classify_bucket, group_aware_fill,
mean_of, push_bounded, reservoir_sample, seeded_hash,
};
use tracing::info;
#[derive(Parser)]
#[command(
name = "shogiesa",
version,
about = "Shogi training data feed for NNUE engines."
)]
struct Cli {
#[command(subcommand)]
command: Commands,
}
#[derive(Subcommand)]
enum Commands {
Extract(ExtractArgs),
Label(Box<LabelArgs>),
Stability(StabilityArgs),
Pack(PackArgs),
Unpack(UnpackArgs),
Split(SplitArgs),
Sample(SampleArgs),
Mine(MineArgs),
Balance(BalanceArgs),
Stratify(StratifyArgs),
Select(SelectArgs),
Filter(FilterArgs),
Calibrate(CalibrateArgs),
Audit(AuditArgs),
Tune(TuneArgs),
Report(ReportArgs),
Distribution(DistributionArgs),
Validate(ValidateArgs),
Cache(CacheArgs),
FromMatch(FromMatchArgs),
MergeObservations(MergeObservationsArgs),
MakeGateOpenings(MakeGateOpeningsArgs),
Shuffle(ShuffleArgs),
#[command(name = "lineprior")]
LinePrior(LinePriorArgs),
}
#[derive(clap::Args)]
struct ExtractArgs {
#[arg(short, long)]
input: PathBuf,
#[arg(short, long)]
out: PathBuf,
#[arg(long, default_value = "1")]
min_ply: u32,
#[arg(long)]
max_ply: Option<u32>,
#[arg(long, default_value = "1", name = "every-n-plies")]
every_n_plies: u32,
#[arg(long)]
dedup: bool,
#[arg(long)]
dedup_zobrist: bool,
}
#[derive(clap::Args)]
struct FromMatchArgs {
#[arg(short, long)]
input: PathBuf,
#[arg(short, long)]
out: PathBuf,
#[arg(long, value_parser = ["engine1", "engine2"])]
losing_side: Option<String>,
#[arg(long, default_value = "1")]
min_ply: u32,
#[arg(long)]
max_ply: Option<u32>,
#[arg(long, default_value = "1", name = "every-n-plies")]
every_n_plies: u32,
#[arg(long)]
dedup: bool,
}
#[derive(clap::Args)]
struct MergeObservationsArgs {
#[arg(long)]
primary: PathBuf,
#[arg(long)]
secondary: PathBuf,
#[arg(short, long)]
out: PathBuf,
#[arg(
long,
default_value = "keep-both",
value_parser = ["keep-both", "prefer-primary", "prefer-secondary"]
)]
on_collision: String,
}
#[derive(clap::Args)]
struct ReportArgs {
#[arg(short, long)]
input: PathBuf,
}
#[derive(clap::Args)]
struct DistributionArgs {
#[arg(short, long)]
input: PathBuf,
#[arg(long, default_value = "20")]
ply_bucket_size: u32,
#[arg(long, default_value = "0.5")]
under_ratio: f64,
#[arg(long, default_value = "2.0")]
over_ratio: f64,
}
#[derive(clap::Args)]
struct ValidateArgs {
#[arg(short, long)]
input: PathBuf,
#[arg(long)]
strict: bool,
}
#[derive(clap::Args)]
struct LabelArgs {
#[arg(short, long)]
input: PathBuf,
#[arg(long)]
engine: PathBuf,
#[arg(long)]
engine_name: Option<String>,
#[arg(long, required_unless_present = "nodes", conflicts_with = "nodes")]
depths: Option<String>,
#[arg(long, required_unless_present = "depths", conflicts_with = "depths")]
nodes: Option<String>,
#[arg(long, default_value = "10000")]
timeout_ms: u64,
#[arg(long, default_value = "1")]
jobs: usize,
#[arg(long = "engine-option", value_name = "KEY=VALUE")]
engine_options: Vec<String>,
#[arg(long, default_value = "1")]
multipv: u32,
#[arg(long, conflicts_with = "replace_existing")]
skip_existing: bool,
#[arg(long)]
replace_existing: bool,
#[arg(long)]
resume_from: Option<PathBuf>,
#[arg(short, long)]
out: PathBuf,
#[arg(long)]
manifest: Option<PathBuf>,
#[arg(long)]
preserve_order: bool,
#[arg(long)]
cache_dir: Option<PathBuf>,
#[arg(
long,
default_value = "content",
value_parser = ["content", "metadata", "none"]
)]
engine_fingerprint_mode: String,
#[arg(long)]
weight_file: Option<PathBuf>,
#[arg(long)]
experiment_id: Option<String>,
#[arg(long)]
candidate_id: Option<String>,
#[arg(long)]
baseline_id: Option<String>,
#[arg(long)]
lineage_id: Option<String>,
#[arg(long)]
teacher_manifest_sha256: Option<String>,
#[arg(long)]
init_seed: Option<u64>,
#[arg(long)]
split_seed: Option<u64>,
#[arg(long)]
split_sha256: Option<String>,
#[arg(long)]
validity: Option<String>,
#[arg(long)]
usi_strict: bool,
#[arg(long)]
restart_engine_every: Option<u32>,
#[arg(long)]
restart_on_protocol_error: bool,
#[arg(long)]
transcript_on_error: Option<PathBuf>,
}
#[derive(Clone, Copy)]
enum ExistingPolicy {
Append,
Skip,
Replace,
}
#[derive(clap::Args)]
struct StabilityArgs {
#[arg(short, long)]
input: PathBuf,
#[arg(short, long)]
out: PathBuf,
}
#[derive(clap::Args)]
struct SplitArgs {
#[arg(short, long)]
input: PathBuf,
#[arg(long)]
by_source: bool,
#[arg(long = "out-dir")]
out_dir: Option<PathBuf>,
#[arg(long, default_value = "256")]
max_open_writers: usize,
#[arg(long)]
train: Option<PathBuf>,
#[arg(long)]
valid: Option<PathBuf>,
#[arg(long)]
test: Option<PathBuf>,
#[arg(long, default_value = "0.1")]
valid_frac: f64,
#[arg(long, default_value = "0.1")]
test_frac: f64,
#[arg(long, default_value = "0")]
seed: u64,
}
#[derive(clap::Args)]
struct SampleArgs {
#[arg(short, long)]
input: PathBuf,
#[arg(short, long)]
out: PathBuf,
#[arg(long)]
count: usize,
#[arg(long, default_value = "0")]
seed: u64,
#[arg(long)]
manifest: Option<PathBuf>,
}
#[derive(clap::Args)]
struct ShuffleArgs {
#[arg(short, long)]
input: PathBuf,
#[arg(short, long)]
out: PathBuf,
#[arg(long, default_value = "0")]
seed: u64,
#[arg(long, default_value = "32")]
block_size: u64,
#[arg(long = "order-manifest")]
order_manifest: Option<PathBuf>,
#[arg(long, requires = "teacher_depth")]
teacher_engine: Option<String>,
#[arg(long, requires = "teacher_engine")]
teacher_depth: Option<u32>,
#[arg(long)]
split_id: Option<String>,
#[arg(long)]
experiment_id: Option<String>,
#[arg(long)]
manifest: Option<PathBuf>,
}
#[derive(clap::Args)]
struct MineArgs {
#[arg(short, long)]
input: PathBuf,
#[arg(short, long)]
out: PathBuf,
#[arg(long, default_value = "200")]
blunder_threshold: i32,
#[arg(long, default_value = "1")]
blunder_window: usize,
#[arg(long)]
losing_threshold: Option<i32>,
}
#[derive(clap::Args)]
struct BalanceArgs {
#[arg(short, long)]
input: PathBuf,
#[arg(short, long)]
out: PathBuf,
#[arg(long = "by", value_name = "DIMENSION")]
by: Vec<String>,
#[arg(long)]
target: Option<usize>,
#[arg(long)]
manifest: Option<PathBuf>,
}
#[derive(clap::Args)]
struct StratifyArgs {
#[arg(short, long)]
input: PathBuf,
#[arg(long, conflicts_with_all = ["quota", "out", "manifest"])]
write_template: Option<PathBuf>,
#[arg(long = "by", value_name = "DIMENSION", conflicts_with = "quota")]
by: Vec<String>,
#[arg(long, conflicts_with = "write_template")]
quota: Option<PathBuf>,
#[arg(short, long)]
out: Option<PathBuf>,
#[arg(long, default_value = "0")]
seed: u64,
#[arg(long)]
manifest: Option<PathBuf>,
}
#[derive(clap::Args)]
struct MakeGateOpeningsArgs {
#[arg(short, long)]
input: PathBuf,
#[arg(short, long)]
out: PathBuf,
#[arg(long)]
count: usize,
#[arg(long, default_value = "8")]
min_ply: u32,
#[arg(long)]
max_ply: Option<u32>,
#[arg(long, default_value = "0")]
seed: u64,
#[arg(long)]
manifest: Option<PathBuf>,
#[arg(long)]
allow_unplayable: bool,
}
#[derive(clap::Args)]
struct LinePriorArgs {
#[command(subcommand)]
action: LinePriorAction,
}
#[derive(clap::Subcommand)]
enum LinePriorAction {
Export(LinePriorExportArgs),
}
#[derive(clap::Args)]
struct LinePriorExportArgs {
#[arg(long)]
input: PathBuf,
#[arg(long)]
out: PathBuf,
#[arg(long, default_value = "sfen", value_parser = ["sfen"])]
state_format: String,
#[arg(long, default_value = "usi", value_parser = ["usi"])]
action_format: String,
#[arg(long)]
max_ply: Option<u32>,
#[arg(long)]
source: String,
#[arg(long, default_value = "game-result", value_parser = ["game-result"])]
outcome_mode: String,
#[arg(long, default_value = "none", value_parser = ["none"])]
score_mode: String,
#[arg(long)]
manifest: Option<PathBuf>,
}
#[derive(clap::Args)]
struct SelectArgs {
#[arg(short, long)]
input: PathBuf,
#[arg(long, value_parser = ["uncertain", "hard", "coverage"])]
strategy: String,
#[arg(long)]
count: usize,
#[arg(long, default_value = "0")]
seed: u64,
#[arg(short, long)]
out: PathBuf,
#[arg(long, allow_hyphen_values = true)]
min_policy_margin_cp: Option<i32>,
#[arg(long, default_value = "200")]
blunder_threshold: i32,
#[arg(long, default_value = "1")]
blunder_window: usize,
#[arg(long)]
assume_grouped_by_source: bool,
}
#[derive(clap::Args)]
struct PackArgs {
#[arg(short, long)]
input: PathBuf,
#[arg(short, long)]
out: PathBuf,
#[arg(long)]
manifest: Option<PathBuf>,
}
#[derive(clap::Args)]
struct UnpackArgs {
#[arg(short, long)]
input: PathBuf,
#[arg(short, long)]
out: PathBuf,
}
#[derive(clap::Args)]
struct FilterArgs {
#[arg(short, long)]
input: PathBuf,
#[arg(short, long, required_unless_present = "dry_run")]
out: Option<PathBuf>,
#[arg(long)]
dry_run: bool,
#[arg(long)]
require_bestmove_agreement: bool,
#[arg(long)]
exclude_mate: bool,
#[arg(long)]
exclude_in_check: bool,
#[arg(long)]
exclude_capture: bool,
#[arg(long)]
exclude_timeout_salvaged: bool,
#[arg(long)]
max_score_swing_cp: Option<i32>,
#[arg(long, allow_hyphen_values = true)]
eval_min: Option<i32>,
#[arg(long, allow_hyphen_values = true)]
eval_max: Option<i32>,
#[arg(long, default_value = "1")]
min_observations: u32,
#[arg(long)]
phase: Option<String>,
#[arg(long, allow_hyphen_values = true)]
min_policy_margin_cp: Option<i32>,
#[arg(long)]
require_exact_score: bool,
#[arg(long)]
require_policy_margin: bool,
#[arg(long)]
min_depth_reached: Option<u32>,
#[arg(long)]
require_requested_depth_reached: bool,
#[arg(long)]
allow_timeout_salvaged_mate: bool,
#[arg(long)]
require_engine_agreement: bool,
#[arg(long)]
max_engine_score_swing_cp: Option<i32>,
#[arg(long)]
manifest: Option<PathBuf>,
#[arg(long)]
explain_out: Option<PathBuf>,
#[arg(long, conflicts_with_all = [
"min_observations", "phase", "exclude_mate", "exclude_in_check", "exclude_capture",
"exclude_timeout_salvaged", "max_score_swing_cp", "eval_min", "eval_max",
"min_policy_margin_cp", "require_exact_score", "require_policy_margin",
"min_depth_reached", "require_requested_depth_reached", "allow_timeout_salvaged_mate",
"require_engine_agreement", "max_engine_score_swing_cp", "require_bestmove_agreement",
])]
preset: Option<String>,
}
#[derive(clap::Args)]
struct CalibrateArgs {
#[arg(short, long)]
input: PathBuf,
#[arg(short, long)]
out: PathBuf,
#[arg(long, conflicts_with = "min_policy_margin_cp")]
sweep_policy_margin: Option<String>,
#[arg(long, conflicts_with = "max_score_swing_cp")]
sweep_score_swing: Option<String>,
#[arg(long, allow_hyphen_values = true)]
min_policy_margin_cp: Option<i32>,
#[arg(long)]
max_score_swing_cp: Option<i32>,
#[arg(long, default_value = "1")]
min_observations: u32,
#[arg(long)]
phase: Option<String>,
#[arg(long)]
exclude_mate: bool,
#[arg(long)]
exclude_in_check: bool,
#[arg(long)]
exclude_capture: bool,
#[arg(long)]
exclude_timeout_salvaged: bool,
#[arg(long, allow_hyphen_values = true)]
eval_min: Option<i32>,
#[arg(long, allow_hyphen_values = true)]
eval_max: Option<i32>,
#[arg(long)]
require_bestmove_agreement: bool,
#[arg(long)]
require_engine_agreement: bool,
#[arg(long)]
max_engine_score_swing_cp: Option<i32>,
#[arg(long)]
require_exact_score: bool,
#[arg(long)]
require_policy_margin: bool,
#[arg(long)]
min_depth_reached: Option<u32>,
#[arg(long)]
require_requested_depth_reached: bool,
#[arg(long)]
allow_timeout_salvaged_mate: bool,
}
#[derive(clap::Args)]
struct AuditArgs {
#[arg(short, long)]
input: PathBuf,
#[arg(long)]
teacher_depth: u32,
#[arg(long)]
student_depths: String,
#[arg(short, long)]
out: PathBuf,
}
#[derive(clap::Args)]
struct TuneArgs {
#[arg(short, long)]
input: PathBuf,
#[arg(long)]
teacher_depth: u32,
#[arg(long)]
student_depths: String,
#[arg(long, conflicts_with = "min_policy_margin_cp")]
sweep_policy_margin: Option<String>,
#[arg(long, conflicts_with = "max_score_swing_cp")]
sweep_score_swing: Option<String>,
#[arg(long, allow_hyphen_values = true)]
min_policy_margin_cp: Option<i32>,
#[arg(long)]
max_score_swing_cp: Option<i32>,
#[arg(long, default_value = "1")]
min_observations: u32,
#[arg(long)]
phase: Option<String>,
#[arg(long)]
exclude_mate: bool,
#[arg(long)]
exclude_in_check: bool,
#[arg(long)]
exclude_capture: bool,
#[arg(long)]
exclude_timeout_salvaged: bool,
#[arg(long, allow_hyphen_values = true)]
eval_min: Option<i32>,
#[arg(long, allow_hyphen_values = true)]
eval_max: Option<i32>,
#[arg(long)]
require_bestmove_agreement: bool,
#[arg(long)]
require_engine_agreement: bool,
#[arg(long)]
max_engine_score_swing_cp: Option<i32>,
#[arg(long)]
require_exact_score: bool,
#[arg(long)]
require_policy_margin: bool,
#[arg(long)]
min_depth_reached: Option<u32>,
#[arg(long)]
require_requested_depth_reached: bool,
#[arg(long)]
allow_timeout_salvaged_mate: bool,
#[arg(short, long)]
out: PathBuf,
#[arg(long)]
report: Option<PathBuf>,
#[arg(long)]
preset_out: Option<PathBuf>,
}
#[derive(clap::Args)]
struct CacheArgs {
#[command(subcommand)]
action: CacheAction,
}
#[derive(clap::Subcommand)]
enum CacheAction {
Stats(CacheStatsArgs),
Verify(CacheVerifyArgs),
Prune(CachePruneArgs),
}
#[derive(clap::Args)]
struct CacheStatsArgs {
#[arg(long)]
cache_dir: PathBuf,
}
#[derive(clap::Args)]
struct CacheVerifyArgs {
#[arg(long)]
cache_dir: PathBuf,
}
#[derive(clap::Args)]
struct CachePruneArgs {
#[arg(long)]
cache_dir: PathBuf,
#[arg(long)]
corrupted_only: bool,
#[arg(long)]
legacy_only: bool,
#[arg(long)]
older_than_days: Option<u64>,
#[arg(long)]
yes: bool,
}
fn main() -> Result<()> {
tracing_subscriber::fmt()
.with_env_filter(tracing_subscriber::EnvFilter::from_default_env())
.with_writer(std::io::stderr)
.init();
let cli = Cli::parse();
match cli.command {
Commands::Extract(args) => cmd_extract(args),
Commands::Label(args) => cmd_label(*args),
Commands::Stability(args) => cmd_stability(args),
Commands::Split(args) => cmd_split(args),
Commands::Sample(args) => cmd_sample(args),
Commands::Mine(args) => cmd_mine(args),
Commands::Balance(args) => cmd_balance(args),
Commands::Stratify(args) => cmd_stratify(args),
Commands::Select(args) => cmd_select(args),
Commands::Pack(args) => cmd_pack(args),
Commands::Unpack(args) => cmd_unpack(args),
Commands::Filter(args) => cmd_filter(args),
Commands::Calibrate(args) => cmd_calibrate(args),
Commands::Audit(args) => cmd_audit(args),
Commands::Tune(args) => cmd_tune(args),
Commands::Report(args) => cmd_report(args),
Commands::Distribution(args) => cmd_distribution(args),
Commands::Validate(args) => cmd_validate(args),
Commands::Cache(args) => match args.action {
CacheAction::Stats(a) => cmd_cache_stats(a),
CacheAction::Verify(a) => cmd_cache_verify(a),
CacheAction::Prune(a) => cmd_cache_prune(a),
},
Commands::FromMatch(args) => cmd_from_match(args),
Commands::MergeObservations(args) => cmd_merge_observations(args),
Commands::MakeGateOpenings(args) => cmd_make_gate_openings(args),
Commands::Shuffle(args) => cmd_shuffle(args),
Commands::LinePrior(args) => match args.action {
LinePriorAction::Export(a) => cmd_lineprior_export(a),
},
}
}
fn zobrist_dedup_keep(
rec: &PositionRecord,
seen_hashes: &mut HashSet<u64>,
skipped: &mut usize,
) -> bool {
match zobrist_from_sfen(&rec.sfen) {
Some(hash) => seen_hashes.insert(hash),
None => {
tracing::warn!(sfen = %rec.sfen, "cannot zobrist-hash SFEN, skipping");
*skipped += 1;
false
}
}
}
fn cmd_extract(args: ExtractArgs) -> Result<()> {
let config = shogiesa_core::ExtractConfig {
min_ply: args.min_ply,
max_ply: args.max_ply,
every_n: args.every_n_plies,
dedup: args.dedup,
};
let paths = collect_game_paths(&args.input)?;
if paths.is_empty() {
anyhow::bail!("no .csa or .kif files found in {:?}", args.input);
}
let out_file =
File::create(&args.out).with_context(|| format!("cannot create {:?}", args.out))?;
let mut writer = BufWriter::new(out_file);
let use_zobrist = args.dedup_zobrist;
let extract_config = if use_zobrist {
shogiesa_core::ExtractConfig {
dedup: false,
..config
}
} else {
config
};
let mut seen: HashSet<String> = HashSet::new();
let mut seen_hashes: HashSet<u64> = HashSet::new();
let mut total_games = 0usize;
let mut total_positions = 0usize;
let mut skipped = 0usize;
for path in &paths {
total_games += 1;
let ext = path.extension().and_then(|e| e.to_str()).unwrap_or("");
let result = match ext {
"kif" | "ki2" => shogiesa_kif::extract_from_path(path, &extract_config, &mut seen)
.map_err(|e| e.to_string()),
_ => shogiesa_csa::extract_from_path(path, &extract_config, &mut seen)
.map_err(|e| e.to_string()),
};
match result {
Ok(records) => {
for rec in &records {
if use_zobrist && !zobrist_dedup_keep(rec, &mut seen_hashes, &mut skipped) {
continue;
}
serde_json::to_writer(&mut writer, rec)?;
writer.write_all(b"\n")?;
total_positions += 1;
}
}
Err(e) => {
tracing::warn!(path = %path.display(), "skipped: {e}");
skipped += 1;
}
}
info!(
games = total_games,
positions = total_positions,
"processed {}",
path.display()
);
}
writer.flush()?;
eprintln!(
"done: {} games, {} positions extracted, {} skipped → {:?}",
total_games, total_positions, skipped, args.out
);
Ok(())
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum MatchResult {
Engine1Win,
Engine2Win,
Draw,
}
fn collect_match_kifu_paths(input: &Path) -> Result<Vec<PathBuf>> {
if input.is_file() {
return Ok(vec![input.to_path_buf()]);
}
let mut paths = Vec::new();
for entry in fs::read_dir(input).with_context(|| format!("cannot read directory {input:?}"))? {
let p = entry?.path();
if p.extension().and_then(|e| e.to_str()) == Some("txt") {
paths.push(p);
}
}
paths.sort();
Ok(paths)
}
fn parse_match_kifu_header(
line: &str,
result: &mut Option<MatchResult>,
engine1_color: &mut Option<SideToMove>,
engine2_color: &mut Option<SideToMove>,
) {
let line = line.trim();
if let Some(rest) = line.strip_prefix("# Engine1:") {
if rest.trim_end().ends_with("(Black)") {
*engine1_color = Some(SideToMove::Black);
} else if rest.trim_end().ends_with("(White)") {
*engine1_color = Some(SideToMove::White);
}
return;
}
if let Some(rest) = line.strip_prefix("# Engine2:") {
if rest.trim_end().ends_with("(Black)") {
*engine2_color = Some(SideToMove::Black);
} else if rest.trim_end().ends_with("(White)") {
*engine2_color = Some(SideToMove::White);
}
return;
}
match line {
"# Result: Engine1 Win" => *result = Some(MatchResult::Engine1Win),
"# Result: Engine2 Win" => *result = Some(MatchResult::Engine2Win),
"# Result: Draw" => *result = Some(MatchResult::Draw),
_ => {}
}
}
fn match_result_to_game_outcome(
result: MatchResult,
engine1_color: Option<SideToMove>,
engine2_color: Option<SideToMove>,
) -> Option<GameOutcome> {
let wins = |color: SideToMove| match color {
SideToMove::Black => GameOutcome::BlackWins,
SideToMove::White => GameOutcome::WhiteWins,
};
match result {
MatchResult::Draw => Some(GameOutcome::Draw),
MatchResult::Engine1Win => engine1_color.map(wins),
MatchResult::Engine2Win => engine2_color.map(wins),
}
}
#[cfg(test)]
mod match_result_to_game_outcome_tests {
use super::{GameOutcome, MatchResult, SideToMove, match_result_to_game_outcome};
#[test]
fn engine1_win_black_is_black_wins() {
assert_eq!(
match_result_to_game_outcome(
MatchResult::Engine1Win,
Some(SideToMove::Black),
Some(SideToMove::White)
),
Some(GameOutcome::BlackWins)
);
}
#[test]
fn engine2_win_white_is_black_wins() {
assert_eq!(
match_result_to_game_outcome(
MatchResult::Engine2Win,
Some(SideToMove::Black),
Some(SideToMove::White)
),
Some(GameOutcome::WhiteWins)
);
}
#[test]
fn draw_is_draw_regardless_of_colors() {
assert_eq!(
match_result_to_game_outcome(MatchResult::Draw, None, None),
Some(GameOutcome::Draw)
);
}
#[test]
fn missing_color_is_none() {
assert_eq!(
match_result_to_game_outcome(MatchResult::Engine1Win, None, Some(SideToMove::White)),
None
);
assert_eq!(
match_result_to_game_outcome(MatchResult::Engine2Win, Some(SideToMove::Black), None),
None
);
}
}
fn match_qualifies(losing_side: Option<&str>, result: Option<MatchResult>) -> bool {
match losing_side {
None => true,
Some("engine1") => result == Some(MatchResult::Engine2Win),
Some("engine2") => result == Some(MatchResult::Engine1Win),
Some(_) => false, }
}
fn extract_from_match_kifu(
content: &str,
source_path: &str,
losing_side: Option<&str>,
config: &shogiesa_core::ExtractConfig,
seen: &mut HashSet<String>,
) -> Vec<PositionRecord> {
let mut result = None;
let mut engine1_color = None;
let mut engine2_color = None;
let mut position_line = None;
for line in content.lines() {
if line.starts_with("position ") {
position_line = Some(line);
break;
}
parse_match_kifu_header(line, &mut result, &mut engine1_color, &mut engine2_color);
}
if !match_qualifies(losing_side, result) {
return Vec::new();
}
let game_result = result
.and_then(|r| match_result_to_game_outcome(r, engine1_color, engine2_color))
.map(|outcome| GameResultInfo {
outcome,
result_source: "match_header".to_string(),
});
let Some(position_line) = position_line else {
tracing::warn!(
path = source_path,
"no `position ...` line found, skipping game"
);
return Vec::new();
};
let tokens: Vec<&str> = position_line.split_whitespace().collect();
let (mut board, move_tokens): (Board, &[&str]) = match tokens.as_slice() {
["position", "startpos", "moves", rest @ ..] => (Board::initial(SideToMove::Black), rest),
["position", "startpos"] => (Board::initial(SideToMove::Black), &[]),
[
"position",
"sfen",
b,
side,
hand,
move_count,
"moves",
rest @ ..,
] => {
let sfen = format!("{b} {side} {hand} {move_count}");
match Board::from_sfen(&sfen) {
Ok(board) => (board, rest),
Err(e) => {
tracing::warn!(path = source_path, "bad sfen {sfen:?}: {e}");
return Vec::new();
}
}
}
["position", "sfen", b, side, hand, move_count] => {
let sfen = format!("{b} {side} {hand} {move_count}");
match Board::from_sfen(&sfen) {
Ok(board) => (board, &[] as &[&str]),
Err(e) => {
tracing::warn!(path = source_path, "bad sfen {sfen:?}: {e}");
return Vec::new();
}
}
}
_ => {
tracing::warn!(
path = source_path,
"unrecognized `position` line, skipping game"
);
return Vec::new();
}
};
let mut out = Vec::new();
let mut ply: u32 = board.move_count - 1;
for token in move_tokens {
let mv = match parse_usi_move(token) {
Ok(mv) => mv,
Err(e) => {
tracing::warn!(path = source_path, ply, "malformed move {token:?}: {e}");
break;
}
};
let mover = board.side;
let has_capture = match mv {
UsiMove::Normal {
to_file, to_rank, ..
} => board.is_capture(to_file, to_rank, mover),
UsiMove::Drop { .. } => false,
};
if let Err(e) = board.apply_usi_move(mover, &mv) {
tracing::warn!(path = source_path, ply, "board error: {e}");
break;
}
ply += 1;
if config.max_ply.is_some_and(|max| ply > max) {
break;
}
if ply < config.min_ply {
continue;
}
if !(ply - config.min_ply).is_multiple_of(config.every_n) {
continue;
}
let sfen = board.to_sfen();
if config.dedup && !seen.insert(sfen.clone()) {
continue;
}
let tags = PositionTags {
phase: phase_from_ply(ply),
side_to_move: board.side,
in_check: board.is_in_check(),
has_capture,
};
let source = SourceInfo {
kind: "from_match".to_string(),
path: source_path.to_string(),
ply,
root_id: None,
variation_id: None,
branch_from_ply: None,
};
let mut rec = PositionRecord::new(sfen, source, tags);
rec.game_result = game_result.clone();
out.push(rec);
}
out
}
fn cmd_from_match(args: FromMatchArgs) -> Result<()> {
let config = shogiesa_core::ExtractConfig {
min_ply: args.min_ply,
max_ply: args.max_ply,
every_n: args.every_n_plies,
dedup: args.dedup,
};
let paths = collect_match_kifu_paths(&args.input)?;
if paths.is_empty() {
anyhow::bail!("no .txt kifu files found in {:?}", args.input);
}
let out_file =
File::create(&args.out).with_context(|| format!("cannot create {:?}", args.out))?;
let mut writer = BufWriter::new(out_file);
let mut seen: HashSet<String> = HashSet::new();
let mut total_games = 0usize;
let mut total_positions = 0usize;
for path in &paths {
total_games += 1;
let content = fs::read_to_string(path).with_context(|| format!("cannot read {path:?}"))?;
let source = path.to_string_lossy().into_owned();
let records = extract_from_match_kifu(
&content,
&source,
args.losing_side.as_deref(),
&config,
&mut seen,
);
for rec in &records {
serde_json::to_writer(&mut writer, rec)?;
writer.write_all(b"\n")?;
total_positions += 1;
}
}
writer.flush()?;
eprintln!(
"done: {total_games} games read, {total_positions} positions extracted → {:?}",
args.out
);
Ok(())
}
#[derive(Clone, Copy)]
enum MergeObservationPolicy {
KeepBoth,
PreferPrimary,
PreferSecondary,
}
fn merge_observations_into(
base: &mut Vec<Observation>,
incoming: Vec<Observation>,
policy: MergeObservationPolicy,
) -> usize {
let key = |o: &Observation| {
(
o.engine.clone(),
o.engine_version.clone(),
o.depth,
o.requested_depth,
o.search_limit_kind,
o.requested_nodes,
)
};
let mut collisions = 0usize;
match policy {
MergeObservationPolicy::KeepBoth => base.extend(incoming),
MergeObservationPolicy::PreferPrimary => {
for obs in incoming {
let k = key(&obs);
if base.iter().any(|o| key(o) == k) {
collisions += 1;
continue;
}
base.push(obs);
}
}
MergeObservationPolicy::PreferSecondary => {
for obs in incoming {
let k = key(&obs);
if base.iter().any(|o| key(o) == k) {
collisions += 1;
base.retain(|o| key(o) != k);
}
base.push(obs);
}
}
}
collisions
}
fn merge_alignment_key(rec: &PositionRecord) -> (String, String, u32) {
(rec.sfen.clone(), rec.source.path.clone(), rec.source.ply)
}
type ResumeIndex = HashMap<(String, String, u32), u64>;
fn build_resume_index(path: &Path) -> Result<ResumeIndex> {
#[derive(Deserialize)]
struct KeyOnly {
sfen: String,
source: KeyOnlySource,
}
#[derive(Deserialize)]
struct KeyOnlySource {
path: String,
ply: u32,
}
let file = File::open(path).with_context(|| format!("cannot open {path:?}"))?;
let mut reader = BufReader::new(file);
let mut index = HashMap::new();
let mut offset = 0u64;
let mut buf = Vec::new();
loop {
buf.clear();
let n = reader
.read_until(b'\n', &mut buf)
.with_context(|| format!("cannot read {path:?}"))?;
if n == 0 {
break;
}
let line_offset = offset;
offset += n as u64;
let line = String::from_utf8_lossy(&buf);
let line = line.trim();
if line.is_empty() {
continue;
}
match serde_json::from_str::<KeyOnly>(line) {
Ok(rec) => {
index.insert((rec.sfen, rec.source.path, rec.source.ply), line_offset);
}
Err(e) => {
tracing::warn!("resume-from parse error at offset {line_offset}: {e}");
}
}
}
Ok(index)
}
fn read_resume_observations_at(file: &mut File, offset: u64) -> Result<Vec<Observation>> {
file.seek(SeekFrom::Start(offset))
.context("cannot seek resume-from file")?;
let mut line = String::new();
BufReader::new(file)
.read_line(&mut line)
.context("cannot read resume-from file")?;
let record: PositionRecord =
serde_json::from_str(line.trim()).context("cannot parse resume-from record")?;
Ok(record.observations)
}
fn cmd_merge_observations(args: MergeObservationsArgs) -> Result<()> {
let policy = match args.on_collision.as_str() {
"keep-both" => MergeObservationPolicy::KeepBoth,
"prefer-primary" => MergeObservationPolicy::PreferPrimary,
"prefer-secondary" => MergeObservationPolicy::PreferSecondary,
_ => unreachable!("clap's value_parser already restricts --on-collision"),
};
let (secondary_records, _) = load_records(&args.secondary)?;
let secondary_by_key: HashMap<(String, String, u32), usize> = secondary_records
.iter()
.enumerate()
.map(|(i, r)| (merge_alignment_key(r), i))
.collect();
let mut consumed = vec![false; secondary_records.len()];
let (primary_records, _) = load_records(&args.primary)?;
let out_file =
File::create(&args.out).with_context(|| format!("cannot create {:?}", args.out))?;
let mut writer = BufWriter::new(out_file);
let (mut both, mut primary_only, mut collisions) = (0usize, 0usize, 0usize);
for mut rec in primary_records {
if let Some(&idx) = secondary_by_key.get(&merge_alignment_key(&rec)) {
consumed[idx] = true;
both += 1;
collisions += merge_observations_into(
&mut rec.observations,
secondary_records[idx].observations.clone(),
policy,
);
rec.stability = None;
} else {
primary_only += 1;
}
serde_json::to_writer(&mut writer, &rec)?;
writer.write_all(b"\n")?;
}
let mut secondary_only = 0usize;
for (idx, rec) in secondary_records.into_iter().enumerate() {
if !consumed[idx] {
serde_json::to_writer(&mut writer, &rec)?;
writer.write_all(b"\n")?;
secondary_only += 1;
}
}
writer.flush()?;
eprintln!(
"done: {} written ({both} merged, {primary_only} primary-only, {secondary_only} \
secondary-only, {collisions} colliding observation keys resolved by --on-collision {}) \
→ {:?}",
both + primary_only + secondary_only,
args.on_collision,
args.out
);
Ok(())
}
struct LabelCache {
dir: PathBuf,
engine_options_hash: u64,
multipv: u32,
engine_fingerprint: Option<u64>,
engine_fingerprint_mode: EngineFingerprintMode,
hits: Arc<AtomicUsize>,
misses: Arc<AtomicUsize>,
}
#[derive(Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
enum EngineFingerprintMode {
Content,
Metadata,
None,
}
impl EngineFingerprintMode {
fn parse(s: &str) -> Self {
match s {
"content" => Self::Content,
"metadata" => Self::Metadata,
_ => Self::None,
}
}
fn as_str(self) -> &'static str {
match self {
Self::Content => "content",
Self::Metadata => "metadata",
Self::None => "none",
}
}
}
fn compute_engine_fingerprint(mode: EngineFingerprintMode, engine_path: &Path) -> Option<u64> {
match mode {
EngineFingerprintMode::None => None,
EngineFingerprintMode::Content => match fs::read(engine_path) {
Ok(bytes) => Some(hash_parts_u64(&[&bytes])),
Err(e) => {
tracing::warn!(
engine = %engine_path.display(),
"cannot read engine binary for fingerprinting ({e}) -- is --engine a bare \
name resolved via PATH? falling back to no fingerprint for this run"
);
None
}
},
EngineFingerprintMode::Metadata => {
let stat = fs::canonicalize(engine_path).and_then(|p| fs::metadata(&p).map(|m| (p, m)));
match stat {
Ok((canonical, meta)) => {
let mtime_nanos = meta
.modified()
.ok()
.and_then(|t| t.duration_since(std::time::UNIX_EPOCH).ok())
.map(|d| d.as_nanos() as u64)
.unwrap_or(0);
let path_bytes = canonical.to_string_lossy().into_owned();
Some(hash_parts_u64(&[
path_bytes.as_bytes(),
&meta.len().to_le_bytes(),
&mtime_nanos.to_le_bytes(),
]))
}
Err(e) => {
tracing::warn!(
engine = %engine_path.display(),
"cannot stat engine binary for fingerprinting ({e}) -- is --engine a bare \
name resolved via PATH? falling back to no fingerprint for this run"
);
None
}
}
}
}
}
fn hash_parts(parts: &[&[u8]]) -> blake3::Hash {
let mut h = blake3::Hasher::new();
for p in parts {
h.update(&(p.len() as u64).to_le_bytes());
h.update(p);
}
h.finalize()
}
fn hash_parts_u64(parts: &[&[u8]]) -> u64 {
u64::from_le_bytes(hash_parts(parts).as_bytes()[..8].try_into().unwrap())
}
fn sorted_engine_option_bytes(options: &[(String, String)]) -> Vec<&[u8]> {
let mut sorted: Vec<&(String, String)> = options.iter().collect();
sorted.sort();
let mut parts: Vec<&[u8]> = Vec::with_capacity(sorted.len() * 2);
for (k, v) in &sorted {
parts.push(k.as_bytes());
parts.push(v.as_bytes());
}
parts
}
fn engine_options_hash(options: &[(String, String)]) -> u64 {
hash_parts_u64(&sorted_engine_option_bytes(options))
}
fn engine_options_hash_hex(options: &[(String, String)]) -> String {
hash_parts(&sorted_engine_option_bytes(options))
.to_hex()
.to_string()
}
fn sha256_hex(bytes: &[u8]) -> String {
use sha2::{Digest, Sha256};
let digest = Sha256::digest(bytes);
digest.iter().map(|b| format!("{b:02x}")).collect()
}
fn hash_file_sha256(path: &Path) -> Result<String> {
use sha2::{Digest, Sha256};
let mut file = File::open(path).with_context(|| format!("cannot open {path:?}"))?;
let mut hasher = Sha256::new();
std::io::copy(&mut file, &mut hasher).with_context(|| format!("cannot hash {path:?}"))?;
Ok(hasher
.finalize()
.iter()
.map(|b| format!("{b:02x}"))
.collect())
}
fn compute_weight_sha256(path: &Path) -> Result<String> {
if path.is_dir() {
anyhow::bail!("weight_sha256 requires a single file, got a directory: {path:?}");
}
hash_file_sha256(path)
}
fn compute_binary_sha256(engine_path: &Path) -> Option<String> {
match fs::read(engine_path) {
Ok(bytes) => Some(sha256_hex(&bytes)),
Err(e) => {
tracing::warn!(
engine = %engine_path.display(),
"cannot read engine binary for binary_sha256 ({e})"
);
None
}
}
}
fn engine_option_u32(options: &[(String, String)], key: &str) -> Option<u32> {
options
.iter()
.find(|(k, _)| k.eq_ignore_ascii_case(key))
.and_then(|(_, v)| v.parse().ok())
}
const CACHE_SCHEMA_VERSION: u32 = 1;
#[derive(Serialize, Deserialize)]
struct CacheEntry {
cache_schema_version: u32,
created_at: u64,
schema_version: u32,
engine_name: String,
engine_version: Option<String>,
engine_fingerprint: Option<u64>,
engine_fingerprint_mode: EngineFingerprintMode,
engine_options_hash: u64,
requested_depth: u32,
#[serde(default)]
search_limit_kind: SearchLimitKind,
#[serde(default)]
search_limit_value: u64,
multipv: u32,
observation: Observation,
}
enum CacheRead {
V2(CacheEntry),
V1(Observation),
}
impl CacheRead {
fn observation(&self) -> &Observation {
match self {
Self::V2(entry) => &entry.observation,
Self::V1(obs) => obs,
}
}
fn into_observation(self) -> Observation {
match self {
Self::V2(entry) => entry.observation,
Self::V1(obs) => obs,
}
}
}
fn parse_cache_entry(json: &str) -> Option<CacheRead> {
if let Ok(entry) = serde_json::from_str::<CacheEntry>(json) {
return Some(CacheRead::V2(entry));
}
serde_json::from_str::<Observation>(json)
.ok()
.map(CacheRead::V1)
}
fn label_cache_path(
cache: &LabelCache,
sfen: &str,
engine_name: &str,
engine_version: Option<&str>,
limit: SearchLimit,
) -> PathBuf {
let (version_tag, version_bytes): (&[u8], &[u8]) = match engine_version {
Some(v) => (&[1], v.as_bytes()),
None => (&[0], &[]),
};
let engine_options_hash_bytes = cache.engine_options_hash.to_le_bytes();
let (limit_kind_tag, limit_value_bytes): (&[u8], [u8; 8]) = match limit {
SearchLimit::Depth(d) => (&[0], (d as u64).to_le_bytes()),
SearchLimit::Nodes(n) => (&[1], n.to_le_bytes()),
};
let multipv_bytes = cache.multipv.to_le_bytes();
let schema_version_bytes = SCHEMA_VERSION.to_le_bytes();
let fingerprint_tag: &[u8] = if cache.engine_fingerprint.is_some() {
&[1]
} else {
&[0]
};
let fingerprint_bytes = cache.engine_fingerprint.unwrap_or(0).to_le_bytes();
let key = hash_parts(&[
sfen.as_bytes(),
engine_name.as_bytes(),
version_tag,
version_bytes,
&engine_options_hash_bytes,
limit_kind_tag,
&limit_value_bytes,
&multipv_bytes,
&schema_version_bytes,
fingerprint_tag,
&fingerprint_bytes,
])
.to_hex()
.to_string();
cache.dir.join(&key[0..2]).join(format!("{key}.json"))
}
fn write_cache_entry_atomically(path: &Path, json: &str) -> std::io::Result<()> {
let tmp = path.with_file_name(format!(
"{}.tmp.{}.{:?}",
path.file_name().unwrap_or_default().to_string_lossy(),
std::process::id(),
std::thread::current().id(),
));
fs::write(&tmp, json)?;
match fs::rename(&tmp, path) {
Ok(()) => Ok(()),
Err(_) if path.exists() => {
let _ = fs::remove_file(&tmp);
Ok(())
}
Err(e) => Err(e),
}
}
#[derive(Clone, Default)]
struct UsiStrictConfig {
strict: bool,
restart_on_protocol_error: bool,
restart_engine_every: Option<u32>,
transcript_dir: Option<PathBuf>,
}
#[derive(Clone)]
struct LabelContext {
weight_sha256: Option<String>,
engine_options_hash_hex: String,
timeout_ms: u64,
existing_policy: ExistingPolicy,
strict_cfg: UsiStrictConfig,
}
struct WorkerState {
worker_id: usize,
searches_since_restart: u32,
}
struct RunCounters {
timeout_salvaged: AtomicUsize,
protocol_violations: AtomicUsize,
engine_restarts: AtomicUsize,
}
fn error_kind_label(e: &UsiError) -> &'static str {
match e {
UsiError::Io(_) => "io",
UsiError::Timeout => "timeout",
UsiError::InvalidResponse => "invalid_response",
UsiError::NoBestmove => "no_bestmove",
UsiError::BestmoveWithoutGo => "bestmove_without_go",
UsiError::DuplicateBestmove => "duplicate_bestmove",
UsiError::DelayedBestmoveAfterTimeout => "delayed_bestmove_after_timeout",
UsiError::IllegalBestmove { .. } => "illegal_bestmove",
}
}
fn dump_transcript(dir: &Path, engine: &UsiEngine, worker_id: usize, e: &UsiError) {
let nanos = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_nanos())
.unwrap_or(0);
let kind = error_kind_label(e);
let _ = fs::create_dir_all(dir);
let _ = fs::write(
dir.join(format!("{nanos}_worker{worker_id}_{kind}.log")),
engine.transcript().join("\n"),
);
}
fn handle_analyse_error(
e: &UsiError,
limit: SearchLimit,
engine: &UsiEngine,
strict_cfg: &UsiStrictConfig,
worker_id: usize,
counters: &RunCounters,
) -> bool {
tracing::warn!(worker_id, ?limit, "analysis error: {e}");
let dead = matches!(e, UsiError::Io(_));
let is_protocol_violation = matches!(
e,
UsiError::BestmoveWithoutGo
| UsiError::DuplicateBestmove
| UsiError::DelayedBestmoveAfterTimeout
| UsiError::IllegalBestmove { .. }
);
if is_protocol_violation {
counters.protocol_violations.fetch_add(1, Ordering::Relaxed);
}
if let Some(dir) = &strict_cfg.transcript_dir {
dump_transcript(dir, engine, worker_id, e);
}
dead || strict_cfg.restart_on_protocol_error
}
fn observation_covers_limit(o: &Observation, engine_name: &str, limit: SearchLimit) -> bool {
if o.engine != engine_name {
return false;
}
match limit {
SearchLimit::Depth(d) => o.search_limit_kind == SearchLimitKind::Depth && o.depth >= d,
SearchLimit::Nodes(n) => {
o.search_limit_kind == SearchLimitKind::Nodes
&& o.nodes.is_some_and(|actual| actual >= n)
}
}
}
fn same_search_request(o: &Observation, new: &Observation) -> bool {
o.engine == new.engine
&& o.search_limit_kind == new.search_limit_kind
&& match new.search_limit_kind {
SearchLimitKind::Depth => {
o.depth == new.depth
&& (o.requested_depth.is_none() || o.requested_depth == new.requested_depth)
}
SearchLimitKind::Nodes => o.requested_nodes == new.requested_nodes,
}
}
fn analyze_record(
rec: &mut PositionRecord,
engine: &mut UsiEngine,
limits: &[SearchLimit],
cache: Option<&LabelCache>,
counters: &RunCounters,
ctx: &LabelContext,
worker: &mut WorkerState,
) -> bool {
let mut restart_needed = false;
for &limit in limits {
if matches!(ctx.existing_policy, ExistingPolicy::Skip)
&& rec
.observations
.iter()
.any(|o| observation_covers_limit(o, &engine.engine_name, limit))
{
continue;
}
let cache_path = cache.map(|c| {
label_cache_path(
c,
&rec.sfen,
&engine.engine_name,
engine.engine_version.as_deref(),
limit,
)
});
let cached: Option<Observation> = cache_path.as_ref().and_then(|path| {
let hit = fs::read_to_string(path)
.ok()
.and_then(|s| parse_cache_entry(&s))
.map(CacheRead::into_observation);
if let Some(c) = cache {
c.hits.fetch_add(hit.is_some() as usize, Ordering::Relaxed);
c.misses
.fetch_add(hit.is_none() as usize, Ordering::Relaxed);
}
hit
});
let observation = match cached {
Some(obs) => Some(obs),
None => {
worker.searches_since_restart += 1;
match engine.analyse(&rec.sfen, limit, ctx.timeout_ms) {
Ok(result)
if ctx.strict_cfg.strict
&& result.bestmove_kind.is_none()
&& !is_legal_bestmove_lite(&rec.sfen, &result.bestmove)
.unwrap_or(true) =>
{
let err = UsiError::IllegalBestmove {
sfen: rec.sfen.clone(),
mv: result.bestmove.clone(),
};
if handle_analyse_error(
&err,
limit,
engine,
&ctx.strict_cfg,
worker.worker_id,
counters,
) {
restart_needed = true;
}
None
}
Ok(result) => {
if result.timed_out {
counters.timeout_salvaged.fetch_add(1, Ordering::Relaxed);
}
let obs = Observation {
engine: engine.engine_name.clone(),
engine_version: engine.engine_version.clone(),
depth: result.depth,
requested_depth: match limit {
SearchLimit::Depth(d) => Some(d),
SearchLimit::Nodes(_) => None,
},
requested_nodes: match limit {
SearchLimit::Depth(_) => None,
SearchLimit::Nodes(n) => Some(n),
},
search_limit_kind: limit.kind(),
score: result.score,
score_perspective: ScorePerspective::SideToMove,
score_bound: result.score_bound,
bestmove: result.bestmove,
bestmove_kind: result.bestmove_kind,
nodes: result.nodes,
time_ms: result.time_ms,
seldepth: result.seldepth,
nps: result.nps,
hashfull: result.hashfull,
pv: result.pv,
policy_margin_cp: result.policy_margin_cp,
candidates: result.candidates,
engine_options_hash: Some(ctx.engine_options_hash_hex.clone()),
weight_sha256: ctx.weight_sha256.clone(),
was_timeout_salvaged: result.timed_out,
};
if let Some(path) = &cache_path {
if let Some(parent) = path.parent() {
let _ = fs::create_dir_all(parent);
}
if let Some(c) = cache {
let (cache_requested_depth, search_limit_value) = match limit {
SearchLimit::Depth(d) => (d, d as u64),
SearchLimit::Nodes(n) => (0, n),
};
let entry = CacheEntry {
cache_schema_version: CACHE_SCHEMA_VERSION,
created_at: std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_secs())
.unwrap_or(0),
schema_version: SCHEMA_VERSION,
engine_name: obs.engine.clone(),
engine_version: obs.engine_version.clone(),
engine_fingerprint: c.engine_fingerprint,
engine_fingerprint_mode: c.engine_fingerprint_mode,
engine_options_hash: c.engine_options_hash,
requested_depth: cache_requested_depth,
search_limit_kind: limit.kind(),
search_limit_value,
multipv: c.multipv,
observation: obs.clone(),
};
if let Ok(json) = serde_json::to_string(&entry) {
let _ = write_cache_entry_atomically(path, &json);
}
}
}
Some(obs)
}
Err(e) => {
if handle_analyse_error(
&e,
limit,
engine,
&ctx.strict_cfg,
worker.worker_id,
counters,
) {
restart_needed = true;
}
None
}
}
}
};
if let Some(obs) = observation {
if !matches!(ctx.existing_policy, ExistingPolicy::Append) {
rec.observations.retain(|o| !same_search_request(o, &obs));
}
rec.observations.push(obs);
}
if restart_needed {
break;
}
}
if let Some(every) = ctx.strict_cfg.restart_engine_every
&& worker.searches_since_restart >= every
{
restart_needed = true;
}
restart_needed
}
struct Job {
id: u64,
record: PositionRecord,
}
fn reorder_push(
pending: &mut BTreeMap<u64, PositionRecord>,
next_id: &mut u64,
id: u64,
record: PositionRecord,
) -> Vec<PositionRecord> {
pending.insert(id, record);
let mut ready = Vec::new();
while let Some(record) = pending.remove(next_id) {
ready.push(record);
*next_id += 1;
}
ready
}
fn cmd_label(args: LabelArgs) -> Result<()> {
let run_start = std::time::Instant::now();
let limits: Vec<SearchLimit> = if let Some(depths) = &args.depths {
let depths: Vec<u32> = depths
.split(',')
.filter_map(|s| s.trim().parse().ok())
.collect();
if depths.is_empty() {
anyhow::bail!("--depths must contain at least one valid integer, e.g. '4,6,8'");
}
depths.into_iter().map(SearchLimit::Depth).collect()
} else {
let nodes: Vec<u64> = args
.nodes
.as_ref()
.expect("clap guarantees --depths or --nodes")
.split(',')
.filter_map(|s| s.trim().parse().ok())
.collect();
if nodes.is_empty() {
anyhow::bail!("--nodes must contain at least one valid integer, e.g. '50000,200000'");
}
nodes.into_iter().map(SearchLimit::Nodes).collect()
};
let weight_sha256 = args
.weight_file
.as_ref()
.map(|path| compute_weight_sha256(path))
.transpose()?;
if args.resume_from.as_ref() == Some(&args.out) {
anyhow::bail!(
"--resume-from must not be the same path as --out (that would just replay --out's \
own current contents back into itself instead of resuming from a separate prior \
run's output -- almost certainly not what was intended)"
);
}
let mut resume_source: Option<(ResumeIndex, File)> = match &args.resume_from {
Some(path) if path.exists() => {
let index = build_resume_index(path)?;
let file = File::open(path).with_context(|| format!("cannot open {path:?}"))?;
Some((index, file))
}
_ => None,
};
let engine_path = args.engine.clone();
let engine_name = args.engine_name.clone().unwrap_or_default();
let timeout_ms = args.timeout_ms;
let jobs = args.jobs.max(1);
let mut engine_options: Vec<(String, String)> = args
.engine_options
.iter()
.filter_map(|s| {
let (k, v) = s.split_once('=')?;
Some((k.to_string(), v.to_string()))
})
.collect();
if args.multipv > 1 {
engine_options.push(("MultiPV".to_string(), args.multipv.to_string()));
}
let existing_policy = if args.replace_existing {
ExistingPolicy::Replace
} else if args.skip_existing || args.resume_from.is_some() {
ExistingPolicy::Skip
} else {
ExistingPolicy::Append
};
let strict_cfg = UsiStrictConfig {
strict: args.usi_strict,
restart_on_protocol_error: args.restart_on_protocol_error,
restart_engine_every: args.restart_engine_every,
transcript_dir: args.transcript_on_error.clone(),
};
let label_ctx = LabelContext {
weight_sha256: weight_sha256.clone(),
engine_options_hash_hex: engine_options_hash_hex(&engine_options),
timeout_ms,
existing_policy,
strict_cfg,
};
let engine_fingerprint_mode = EngineFingerprintMode::parse(&args.engine_fingerprint_mode);
let cache: Option<Arc<LabelCache>> = match &args.cache_dir {
Some(dir) => {
let engine_fingerprint =
compute_engine_fingerprint(engine_fingerprint_mode, &engine_path);
Some(Arc::new(LabelCache {
dir: dir.clone(),
engine_options_hash: engine_options_hash(&engine_options),
multipv: args.multipv,
engine_fingerprint,
engine_fingerprint_mode,
hits: Arc::new(AtomicUsize::new(0)),
misses: Arc::new(AtomicUsize::new(0)),
}))
}
None => None,
};
let probe = UsiEngine::launch(
&engine_path,
engine_name.clone(),
timeout_ms,
&engine_options,
)
.with_context(|| format!("failed to launch engine {engine_path:?}"))?;
let engine_display_name = probe.engine_name.clone();
drop(probe);
info!(jobs, "labeling started");
let queue_depth = jobs * 4;
let (permit_tx, permit_rx) = mpsc::sync_channel::<()>(queue_depth);
for _ in 0..queue_depth {
permit_tx.send(()).expect("permit channel just created");
}
let writer_permit_tx = permit_tx.clone();
drop(permit_tx);
let (job_tx, job_rx) = mpsc::sync_channel::<Job>(jobs);
let job_rx = Arc::new(Mutex::new(job_rx));
let (result_tx, result_rx) = mpsc::channel::<Job>();
let engine_launch_failures = Arc::new(AtomicUsize::new(0));
let counters = Arc::new(RunCounters {
timeout_salvaged: AtomicUsize::new(0),
protocol_violations: AtomicUsize::new(0),
engine_restarts: AtomicUsize::new(0),
});
let done = Arc::new(AtomicUsize::new(0));
let input_path = args.input.clone();
let track_input_hash = args.manifest.is_some();
let reader_handle = std::thread::spawn(move || -> Result<(u64, usize, u64, String)> {
let mut input_hasher = blake3::Hasher::new();
let file =
File::open(&input_path).with_context(|| format!("cannot open {input_path:?}"))?;
let reader = BufReader::new(file);
let mut job_id = 0u64;
let mut skipped = 0usize;
let mut resumed_count = 0u64;
for (i, line) in reader.lines().enumerate() {
let line = line.with_context(|| format!("cannot read {input_path:?}"))?;
if track_input_hash {
input_hasher.update(line.as_bytes());
input_hasher.update(b"\n");
}
if line.trim().is_empty() {
continue;
}
match serde_json::from_str::<PositionRecord>(&line) {
Ok(mut record) if Sfen::parse(&record.sfen).is_ok() => {
if let Some((index, file)) = resume_source.as_mut()
&& let Some(&offset) = index.get(&merge_alignment_key(&record))
{
match read_resume_observations_at(file, offset) {
Ok(resume_obs) => {
merge_observations_into(
&mut record.observations,
resume_obs,
MergeObservationPolicy::KeepBoth,
);
resumed_count += 1;
}
Err(e) => {
tracing::warn!(
"resume-from record unreadable at offset {offset}: {e}"
);
}
}
}
if permit_rx.recv().is_err() || job_tx.send(Job { id: job_id, record }).is_err()
{
break; }
job_id += 1;
}
Ok(_) => {
tracing::warn!(line = i + 1, "invalid SFEN, skipping");
skipped += 1;
}
Err(e) => {
tracing::warn!(line = i + 1, "JSON parse error: {e}");
skipped += 1;
}
}
}
Ok((
job_id,
skipped,
resumed_count,
input_hasher.finalize().to_hex().to_string(),
))
});
let worker_handles: Vec<_> = (0..jobs)
.map(|worker_id| {
let job_rx = Arc::clone(&job_rx);
let result_tx = result_tx.clone();
let engine_path = engine_path.clone();
let engine_name = engine_name.clone();
let engine_options = engine_options.clone();
let engine_launch_failures = Arc::clone(&engine_launch_failures);
let counters = Arc::clone(&counters);
let done = Arc::clone(&done);
let limits = limits.clone();
let label_ctx = label_ctx.clone();
let cache = cache.clone();
std::thread::spawn(move || {
let mut engine: Option<UsiEngine> = None;
let mut worker = WorkerState {
worker_id,
searches_since_restart: 0,
};
loop {
let job = {
let rx = job_rx.lock().expect("job queue mutex poisoned");
rx.recv()
};
let Ok(Job { id, mut record }) = job else {
break;
};
if engine.is_none()
&& let Ok(e) = UsiEngine::launch(
&engine_path,
engine_name.clone(),
timeout_ms,
&engine_options,
)
{
engine = Some(
e.with_strict(label_ctx.strict_cfg.strict)
.with_transcript_capture(
label_ctx.strict_cfg.transcript_dir.is_some(),
),
);
worker.searches_since_restart = 0;
}
if let Some(eng) = engine.as_mut() {
let restart_needed = analyze_record(
&mut record,
eng,
&limits,
cache.as_deref(),
&counters,
&label_ctx,
&mut worker,
);
if restart_needed {
engine = None;
counters.engine_restarts.fetch_add(1, Ordering::Relaxed);
}
} else {
engine_launch_failures.fetch_add(1, Ordering::Relaxed);
tracing::warn!(sfen = %record.sfen, "engine unavailable, position left unlabeled");
}
let n = done.fetch_add(1, Ordering::Relaxed) + 1;
if n.is_multiple_of(200) {
eprint!("\r {n} done");
}
if result_tx.send(Job { id, record }).is_err() {
break;
}
}
})
})
.collect();
drop(result_tx);
let out_file =
File::create(&args.out).with_context(|| format!("cannot create {:?}", args.out))?;
let mut out_writer = BufWriter::new(out_file);
let mut manifest = args
.manifest
.as_ref()
.map(|_| RunManifest::new("label", &args.input));
let mut next_id = 0u64;
let mut pending: BTreeMap<u64, PositionRecord> = BTreeMap::new();
let mut written = 0usize;
let mut time_ms_sum = 0u64;
let mut time_ms_count = 0usize;
let mut write_one =
|record: &PositionRecord, manifest: &mut Option<RunManifest>| -> Result<()> {
serde_json::to_writer(&mut out_writer, record)?;
out_writer.write_all(b"\n")?;
out_writer.flush()?;
if let Some(m) = manifest.as_mut() {
accumulate_coverage(m, std::slice::from_ref(record));
}
for obs in &record.observations {
if let Some(t) = obs.time_ms {
time_ms_sum += t;
time_ms_count += 1;
}
}
let _ = writer_permit_tx.send(());
Ok(())
};
for job in result_rx {
if args.preserve_order {
for record in reorder_push(&mut pending, &mut next_id, job.id, job.record) {
write_one(&record, &mut manifest)?;
written += 1;
}
debug_assert!(pending.len() <= queue_depth);
} else {
write_one(&job.record, &mut manifest)?;
written += 1;
}
}
out_writer.flush()?;
eprintln!();
let (total, skipped, resumed_count, input_hash) = reader_handle
.join()
.map_err(|_| anyhow::anyhow!("label reader thread panicked"))??;
for handle in worker_handles {
handle
.join()
.map_err(|_| anyhow::anyhow!("label worker thread panicked"))?;
}
let engine_launch_failures = engine_launch_failures.load(Ordering::Relaxed);
let timeout_salvaged_count = counters.timeout_salvaged.load(Ordering::Relaxed);
let protocol_violations_count = counters.protocol_violations.load(Ordering::Relaxed);
let engine_restarts_count = counters.engine_restarts.load(Ordering::Relaxed);
let cache_counts = cache.as_ref().map(|c| {
(
c.hits.load(Ordering::Relaxed),
c.misses.load(Ordering::Relaxed),
)
});
let elapsed_secs = run_start.elapsed().as_secs_f64();
let records_per_sec = (elapsed_secs > 0.0).then(|| written as f64 / elapsed_secs);
let cache_hit_rate = cache_counts.and_then(|(hits, misses)| {
(hits + misses > 0).then(|| hits as f64 / (hits + misses) as f64)
});
let average_engine_time_ms =
(time_ms_count > 0).then(|| time_ms_sum as f64 / time_ms_count as f64);
let cache_suffix = cache_counts
.map(|(hits, misses)| format!(", {hits} cache hits, {misses} cache misses"))
.unwrap_or_default();
let throughput_suffix = records_per_sec
.map(|r| format!(", {r:.1} rec/s"))
.unwrap_or_default();
eprintln!(
"done [{engine_display_name}, jobs={jobs}]: {total} in, {written} labeled, {skipped} skipped, {engine_launch_failures} engine launch failures{cache_suffix}{throughput_suffix} → {:?}",
args.out
);
if let Some(mut manifest) = manifest {
manifest.input_hash = input_hash;
manifest.records_read = total as usize;
manifest.records_kept = written;
manifest.records_dropped = skipped;
manifest.engine_name = Some(engine_display_name);
let depth_values: Vec<u32> = limits
.iter()
.filter_map(|l| match l {
SearchLimit::Depth(d) => Some(*d),
SearchLimit::Nodes(_) => None,
})
.collect();
let node_values: Vec<u64> = limits
.iter()
.filter_map(|l| match l {
SearchLimit::Nodes(n) => Some(*n),
SearchLimit::Depth(_) => None,
})
.collect();
manifest.depths = (!depth_values.is_empty()).then_some(depth_values);
manifest.nodes = (!node_values.is_empty()).then_some(node_values);
manifest.multipv = (args.multipv > 1).then_some(args.multipv);
manifest.engine_options = Some(args.engine_options.clone());
manifest.jobs = Some(jobs);
manifest.engine_launch_failures = Some(engine_launch_failures);
manifest.timeout_salvaged_count = Some(timeout_salvaged_count);
manifest.protocol_violations_count = args.usi_strict.then_some(protocol_violations_count);
manifest.engine_restarts = Some(engine_restarts_count);
manifest.records_per_sec = records_per_sec;
manifest.average_engine_time_ms = average_engine_time_ms;
manifest.preserve_order = Some(args.preserve_order);
manifest.resume_from = args.resume_from.clone();
manifest.resumed_count = args.resume_from.is_some().then_some(resumed_count);
if let Some((hits, misses)) = cache_counts {
manifest.cache_hits = Some(hits);
manifest.cache_hit_rate = cache_hit_rate;
manifest.cache_misses = Some(misses);
manifest.engine_fingerprint_mode = Some(engine_fingerprint_mode.as_str());
}
manifest.dataset_sha256 = Some(hash_file_sha256(&args.input)?);
manifest.binary_sha256 = compute_binary_sha256(&engine_path);
manifest.weight_sha256 = weight_sha256;
manifest.engine_threads = engine_option_u32(&engine_options, "Threads");
manifest.engine_hash_mb = engine_option_u32(&engine_options, "Hash");
manifest.experiment_id.clone_from(&args.experiment_id);
manifest.candidate_id.clone_from(&args.candidate_id);
manifest.baseline_id.clone_from(&args.baseline_id);
manifest.lineage_id.clone_from(&args.lineage_id);
manifest
.teacher_manifest_sha256
.clone_from(&args.teacher_manifest_sha256);
manifest.init_seed = args.init_seed;
manifest.split_seed = args.split_seed;
manifest.split_sha256.clone_from(&args.split_sha256);
manifest.validity.clone_from(&args.validity);
write_manifest(args.manifest.as_ref().unwrap(), &manifest)?;
}
Ok(())
}
fn cmd_stability(args: StabilityArgs) -> Result<()> {
let reader = BufReader::new(
File::open(&args.input).with_context(|| format!("cannot open {:?}", args.input))?,
);
let out_file =
File::create(&args.out).with_context(|| format!("cannot create {:?}", args.out))?;
let mut writer = BufWriter::new(out_file);
let mut total = 0usize;
let mut enriched = 0usize;
let mut skipped = 0usize;
for (i, line) in reader.lines().enumerate() {
let line = line?;
if line.trim().is_empty() {
continue;
}
total += 1;
let mut rec: PositionRecord = match serde_json::from_str(&line) {
Ok(r) => r,
Err(e) => {
tracing::warn!(line = i + 1, "JSON parse error: {e}");
skipped += 1;
continue;
}
};
rec.fill_stability();
if rec.stability.is_some() {
enriched += 1;
}
serde_json::to_writer(&mut writer, &rec)?;
writer.write_all(b"\n")?;
}
writer.flush()?;
eprintln!(
"done: {total} read, {enriched} enriched with stability, {skipped} skipped → {:?}",
args.out
);
Ok(())
}
struct WriterPoolEntry {
writer: BufWriter<File>,
last_used: u64,
}
struct WriterPool {
max_open: usize,
writers: HashMap<String, WriterPoolEntry>,
opened_this_run: HashSet<String>,
tick: u64,
}
impl WriterPool {
fn new(max_open: usize) -> Self {
Self {
max_open,
writers: HashMap::new(),
opened_this_run: HashSet::new(),
tick: 0,
}
}
fn write_line(&mut self, out_path: &Path, key: &str, rec: &PositionRecord) -> Result<()> {
self.tick += 1;
let tick = self.tick;
if let Some(entry) = self.writers.get_mut(key) {
serde_json::to_writer(&mut entry.writer, rec)?;
entry.writer.write_all(b"\n")?;
entry.last_used = tick;
return Ok(());
}
if self.writers.len() >= self.max_open {
let lru_key = self
.writers
.iter()
.min_by_key(|(_, e)| e.last_used)
.map(|(k, _)| k.clone())
.expect("writers is non-empty when at capacity");
let mut evicted = self.writers.remove(&lru_key).unwrap();
evicted.writer.flush()?;
}
let first_touch = self.opened_this_run.insert(key.to_string());
let file = OpenOptions::new()
.create(true)
.write(true)
.truncate(first_touch)
.append(!first_touch)
.open(out_path)
.with_context(|| format!("cannot open {out_path:?}"))?;
let mut writer = BufWriter::new(file);
serde_json::to_writer(&mut writer, rec)?;
writer.write_all(b"\n")?;
self.writers.insert(
key.to_string(),
WriterPoolEntry {
writer,
last_used: tick,
},
);
Ok(())
}
fn flush_all(&mut self) -> Result<()> {
for entry in self.writers.values_mut() {
entry.writer.flush()?;
}
Ok(())
}
}
fn cmd_split(args: SplitArgs) -> Result<()> {
if args.train.is_some() || args.valid.is_some() || args.test.is_some() {
return cmd_split_train_valid_test(args);
}
if !args.by_source {
anyhow::bail!("either --by-source --out-dir, or --train/--valid/--test, is required");
}
let out_dir = args
.out_dir
.as_ref()
.ok_or_else(|| anyhow::anyhow!("--out-dir is required with --by-source"))?;
fs::create_dir_all(out_dir).with_context(|| format!("cannot create {out_dir:?}"))?;
let reader = BufReader::new(
File::open(&args.input).with_context(|| format!("cannot open {:?}", args.input))?,
);
let mut pool = WriterPool::new(args.max_open_writers.max(1));
let mut file_names: HashMap<String, String> = HashMap::new();
let mut file_counts: BTreeMap<String, usize> = BTreeMap::new();
let mut total = 0usize;
for (i, line) in reader.lines().enumerate() {
let line = line?;
if line.trim().is_empty() {
continue;
}
let rec: PositionRecord = match serde_json::from_str(&line) {
Ok(r) => r,
Err(e) => {
tracing::warn!(line = i + 1, "JSON parse error: {e}");
continue;
}
};
let key = rec.source.path.clone();
let file_name = file_names
.entry(key.clone())
.or_insert_with(|| {
let safe: String = key
.chars()
.map(|c| {
if c.is_alphanumeric() || c == '.' || c == '-' {
c
} else {
'_'
}
})
.collect();
format!("{safe}.jsonl")
})
.clone();
let out_path = out_dir.join(&file_name);
pool.write_line(&out_path, &key, &rec)?;
*file_counts.entry(file_name).or_default() += 1;
total += 1;
}
pool.flush_all()?;
let manifest = serde_json::json!({
"shogiesa_version": env!("CARGO_PKG_VERSION"),
"schema_version": SCHEMA_VERSION,
"input": args.input,
"by_source": args.by_source,
"total_positions": total,
"files": file_counts,
});
let manifest_path = out_dir.join("manifest.json");
fs::write(&manifest_path, serde_json::to_string_pretty(&manifest)?)
.with_context(|| format!("cannot write {manifest_path:?}"))?;
eprintln!(
"done: {total} positions split into {} files → {:?}",
file_counts.len(),
out_dir
);
Ok(())
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
enum SplitBucket {
Test,
Valid,
Train,
}
fn split_root_path(source_path: &str) -> &str {
source_path.split("#var").next().unwrap_or(source_path)
}
fn split_root_key(source: &SourceInfo) -> &str {
source
.root_id
.as_deref()
.unwrap_or_else(|| split_root_path(&source.path))
}
fn assign_split_bucket(seed: u64, root_key: &str, valid_frac: f64, test_frac: f64) -> SplitBucket {
let seed_bytes = seed.to_le_bytes();
let unit = hash_parts_u64(&[&seed_bytes, root_key.as_bytes()]) as f64 / u64::MAX as f64;
if unit < test_frac {
SplitBucket::Test
} else if unit < test_frac + valid_frac {
SplitBucket::Valid
} else {
SplitBucket::Train
}
}
fn cmd_split_train_valid_test(args: SplitArgs) -> Result<()> {
let (Some(train_path), Some(valid_path), Some(test_path)) =
(&args.train, &args.valid, &args.test)
else {
anyhow::bail!("--train, --valid, and --test must all be provided together");
};
if args.valid_frac + args.test_frac >= 1.0 {
anyhow::bail!("--valid-frac + --test-frac must be < 1.0");
}
let reader = BufReader::new(
File::open(&args.input).with_context(|| format!("cannot open {:?}", args.input))?,
);
let mut train_writer = BufWriter::new(
File::create(train_path).with_context(|| format!("cannot create {train_path:?}"))?,
);
let mut valid_writer = BufWriter::new(
File::create(valid_path).with_context(|| format!("cannot create {valid_path:?}"))?,
);
let mut test_writer = BufWriter::new(
File::create(test_path).with_context(|| format!("cannot create {test_path:?}"))?,
);
let mut counts: BTreeMap<&'static str, usize> = BTreeMap::new();
let mut sources: HashMap<&'static str, HashSet<String>> = HashMap::new();
let mut total = 0usize;
for (i, line) in reader.lines().enumerate() {
let line = line?;
if line.trim().is_empty() {
continue;
}
let rec: PositionRecord = match serde_json::from_str(&line) {
Ok(r) => r,
Err(e) => {
tracing::warn!(line = i + 1, "JSON parse error: {e}");
continue;
}
};
let bucket = assign_split_bucket(
args.seed,
split_root_key(&rec.source),
args.valid_frac,
args.test_frac,
);
let (name, writer) = match bucket {
SplitBucket::Train => ("train", &mut train_writer),
SplitBucket::Valid => ("valid", &mut valid_writer),
SplitBucket::Test => ("test", &mut test_writer),
};
serde_json::to_writer(&mut *writer, &rec)?;
writer.write_all(b"\n")?;
*counts.entry(name).or_default() += 1;
sources.entry(name).or_default().insert(rec.source.path);
total += 1;
}
train_writer.flush()?;
valid_writer.flush()?;
test_writer.flush()?;
let split_manifest = |name: &str, path: &PathBuf| {
serde_json::json!({
"path": path,
"positions": counts.get(name).copied().unwrap_or(0),
"sources": sources.get(name).map(HashSet::len).unwrap_or(0),
})
};
let manifest = serde_json::json!({
"shogiesa_version": env!("CARGO_PKG_VERSION"),
"schema_version": SCHEMA_VERSION,
"input": args.input,
"seed": args.seed,
"requested": { "valid_frac": args.valid_frac, "test_frac": args.test_frac },
"total_positions": total,
"splits": {
"train": split_manifest("train", train_path),
"valid": split_manifest("valid", valid_path),
"test": split_manifest("test", test_path),
},
});
let manifest_path = train_path
.parent()
.unwrap_or(Path::new("."))
.join("manifest.json");
fs::write(&manifest_path, serde_json::to_string_pretty(&manifest)?)
.with_context(|| format!("cannot write {manifest_path:?}"))?;
eprintln!(
"done: {total} positions split (seed={}) → train={}, valid={}, test={}",
args.seed,
counts.get("train").copied().unwrap_or(0),
counts.get("valid").copied().unwrap_or(0),
counts.get("test").copied().unwrap_or(0),
);
Ok(())
}
#[derive(serde::Serialize)]
struct RunManifest {
shogiesa_version: &'static str,
git_sha: &'static str,
schema_version: u32,
pack_format_version: u16,
command: &'static str,
args: Vec<String>,
input_path: String,
input_hash: String,
fingerprint_algorithm: &'static str,
records_read: usize,
records_kept: usize,
records_dropped: usize,
#[serde(skip_serializing_if = "BTreeMap::is_empty")]
drop_reasons: BTreeMap<&'static str, usize>,
labeled_records: usize,
unlabeled_records: usize,
observations_with_candidates: usize,
observations_total: usize,
#[serde(skip_serializing_if = "BTreeMap::is_empty")]
score_bound_distribution: BTreeMap<&'static str, usize>,
requested_depth_total: usize,
requested_depth_underreach: usize,
#[serde(skip_serializing_if = "Option::is_none")]
filter_config: Option<QualityConfig>,
#[serde(skip_serializing_if = "Option::is_none")]
engine_name: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
depths: Option<Vec<u32>>,
#[serde(skip_serializing_if = "Option::is_none")]
multipv: Option<u32>,
#[serde(skip_serializing_if = "Option::is_none")]
engine_options: Option<Vec<String>>,
#[serde(skip_serializing_if = "Option::is_none")]
jobs: Option<usize>,
#[serde(skip_serializing_if = "Option::is_none")]
engine_launch_failures: Option<usize>,
#[serde(skip_serializing_if = "Option::is_none")]
timeout_salvaged_count: Option<usize>,
#[serde(skip_serializing_if = "Option::is_none")]
cache_hits: Option<usize>,
#[serde(skip_serializing_if = "Option::is_none")]
cache_misses: Option<usize>,
#[serde(skip_serializing_if = "Option::is_none")]
engine_fingerprint_mode: Option<&'static str>,
#[serde(skip_serializing_if = "Option::is_none")]
records_per_sec: Option<f64>,
#[serde(skip_serializing_if = "Option::is_none")]
cache_hit_rate: Option<f64>,
#[serde(skip_serializing_if = "Option::is_none")]
average_engine_time_ms: Option<f64>,
#[serde(skip_serializing_if = "Option::is_none")]
preserve_order: Option<bool>,
#[serde(skip_serializing_if = "Option::is_none")]
resume_from: Option<PathBuf>,
#[serde(skip_serializing_if = "Option::is_none")]
resumed_count: Option<u64>,
#[serde(skip_serializing_if = "Option::is_none")]
max_root_share_in_any_bucket: Option<f64>,
#[serde(skip_serializing_if = "Option::is_none")]
distinct_roots_kept: Option<usize>,
#[serde(skip_serializing_if = "Option::is_none")]
order_hash: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
experiment_id: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
nodes: Option<Vec<u64>>,
#[serde(skip_serializing_if = "Option::is_none")]
candidate_id: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
baseline_id: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
lineage_id: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
dataset_sha256: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
split_sha256: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
teacher_manifest_sha256: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
binary_sha256: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
weight_sha256: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
init_seed: Option<u64>,
#[serde(skip_serializing_if = "Option::is_none")]
split_seed: Option<u64>,
#[serde(skip_serializing_if = "Option::is_none")]
shuffle_seed: Option<u64>,
#[serde(skip_serializing_if = "Option::is_none")]
validity: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
engine_threads: Option<u32>,
#[serde(skip_serializing_if = "Option::is_none")]
engine_hash_mb: Option<u32>,
#[serde(skip_serializing_if = "Option::is_none")]
protocol_violations_count: Option<usize>,
#[serde(skip_serializing_if = "Option::is_none")]
engine_restarts: Option<usize>,
#[serde(skip_serializing_if = "Option::is_none")]
selection_seed: Option<u64>,
#[serde(skip_serializing_if = "Option::is_none")]
canonical_valid_count: Option<usize>,
#[serde(skip_serializing_if = "Option::is_none")]
output_sha256: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
source_distribution: Option<BTreeMap<String, usize>>,
#[serde(skip_serializing_if = "Option::is_none")]
phase_distribution: Option<BTreeMap<GamePhase, usize>>,
#[serde(skip_serializing_if = "Option::is_none")]
material_distribution: Option<BTreeMap<u32, usize>>,
#[serde(skip_serializing_if = "Option::is_none")]
selection_algorithm_version: Option<&'static str>,
}
impl RunManifest {
fn new(command: &'static str, input_path: &Path) -> Self {
Self {
shogiesa_version: env!("CARGO_PKG_VERSION"),
git_sha: env!("SHOGIESA_GIT_SHA"),
schema_version: SCHEMA_VERSION,
pack_format_version: pack::FORMAT_VERSION,
command,
args: std::env::args().collect(),
input_path: input_path.display().to_string(),
input_hash: String::new(),
fingerprint_algorithm: "blake3",
records_read: 0,
records_kept: 0,
records_dropped: 0,
drop_reasons: BTreeMap::new(),
labeled_records: 0,
unlabeled_records: 0,
observations_with_candidates: 0,
observations_total: 0,
score_bound_distribution: BTreeMap::new(),
requested_depth_total: 0,
requested_depth_underreach: 0,
filter_config: None,
engine_name: None,
depths: None,
multipv: None,
engine_options: None,
jobs: None,
engine_launch_failures: None,
timeout_salvaged_count: None,
cache_hits: None,
cache_misses: None,
engine_fingerprint_mode: None,
records_per_sec: None,
cache_hit_rate: None,
average_engine_time_ms: None,
preserve_order: None,
resume_from: None,
resumed_count: None,
max_root_share_in_any_bucket: None,
distinct_roots_kept: None,
order_hash: None,
experiment_id: None,
nodes: None,
candidate_id: None,
baseline_id: None,
lineage_id: None,
dataset_sha256: None,
split_sha256: None,
teacher_manifest_sha256: None,
binary_sha256: None,
weight_sha256: None,
init_seed: None,
split_seed: None,
shuffle_seed: None,
validity: None,
engine_threads: None,
engine_hash_mb: None,
protocol_violations_count: None,
engine_restarts: None,
selection_seed: None,
canonical_valid_count: None,
output_sha256: None,
source_distribution: None,
phase_distribution: None,
material_distribution: None,
selection_algorithm_version: None,
}
}
}
fn write_manifest(path: &Path, manifest: &RunManifest) -> Result<()> {
fs::write(path, serde_json::to_string_pretty(manifest)?)
.with_context(|| format!("cannot write {path:?}"))
}
fn hash_file(path: &Path) -> Result<String> {
let reader = BufReader::new(File::open(path).with_context(|| format!("cannot open {path:?}"))?);
let mut h = blake3::Hasher::new();
for line in reader.lines() {
h.update(line?.as_bytes());
h.update(b"\n");
}
Ok(h.finalize().to_hex().to_string())
}
fn fold_file_hash(hasher: &mut blake3::Hasher, content: &str) {
hasher.update(blake3::hash(content.as_bytes()).as_bytes());
}
fn score_bound_str(bound: shogiesa_core::ScoreBound) -> &'static str {
match bound {
shogiesa_core::ScoreBound::Exact => "exact",
shogiesa_core::ScoreBound::Lowerbound => "lowerbound",
shogiesa_core::ScoreBound::Upperbound => "upperbound",
}
}
fn accumulate_candidate_coverage(
rec: &PositionRecord,
with_candidates: &mut usize,
total: &mut usize,
score_bound_distribution: &mut BTreeMap<&'static str, usize>,
) {
for obs in &rec.observations {
*total += 1;
if !obs.candidates.is_empty() {
*with_candidates += 1;
for c in &obs.candidates {
let key = score_bound_str(c.score_bound);
*score_bound_distribution.entry(key).or_default() += 1;
}
}
}
}
fn candidate_coverage_stats(
records: &[PositionRecord],
) -> (usize, usize, BTreeMap<&'static str, usize>) {
let mut with_candidates = 0;
let mut total = 0;
let mut score_bound_distribution: BTreeMap<&'static str, usize> = BTreeMap::new();
for rec in records {
accumulate_candidate_coverage(
rec,
&mut with_candidates,
&mut total,
&mut score_bound_distribution,
);
}
(with_candidates, total, score_bound_distribution)
}
fn accumulate_requested_depth(
rec: &PositionRecord,
total_with_requested: &mut usize,
underreach: &mut usize,
) {
for obs in &rec.observations {
if obs.requested_depth.is_some() {
*total_with_requested += 1;
if requested_depth_underreached(obs) {
*underreach += 1;
}
}
}
}
fn requested_depth_stats(records: &[PositionRecord]) -> (usize, usize) {
let mut total_with_requested = 0usize;
let mut underreach = 0usize;
for rec in records {
accumulate_requested_depth(rec, &mut total_with_requested, &mut underreach);
}
(total_with_requested, underreach)
}
fn accumulate_coverage(manifest: &mut RunManifest, records: &[PositionRecord]) {
for rec in records {
if rec.observations.is_empty() {
manifest.unlabeled_records += 1;
} else {
manifest.labeled_records += 1;
}
}
let (with_candidates, total, distribution) = candidate_coverage_stats(records);
manifest.observations_with_candidates += with_candidates;
manifest.observations_total += total;
for (key, count) in distribution {
*manifest.score_bound_distribution.entry(key).or_default() += count;
}
let (requested_total, underreach) = requested_depth_stats(records);
manifest.requested_depth_total += requested_total;
manifest.requested_depth_underreach += underreach;
}
fn cmd_sample(args: SampleArgs) -> Result<()> {
let seed = args.seed;
let count = args.count;
let reader = BufReader::new(
File::open(&args.input).with_context(|| format!("cannot open {:?}", args.input))?,
);
let mut io_error: Option<std::io::Error> = None;
let records = reader
.lines()
.enumerate()
.filter_map(|(i, line)| match line {
Ok(line) if line.trim().is_empty() => None,
Ok(line) => match serde_json::from_str::<PositionRecord>(&line) {
Ok(r) => Some(r),
Err(e) => {
tracing::warn!(line = i + 1, "parse error: {e}");
None
}
},
Err(e) => {
io_error.get_or_insert(e);
None
}
});
let (kept, total) = reservoir_sample(records, count, seed, |r| r.sfen.as_str());
if let Some(e) = io_error {
return Err(e.into());
}
let out_file =
File::create(&args.out).with_context(|| format!("cannot create {:?}", args.out))?;
let mut writer = BufWriter::new(out_file);
for record in &kept {
serde_json::to_writer(&mut writer, record)?;
writer.write_all(b"\n")?;
}
writer.flush()?;
eprintln!(
"done: {}/{total} sampled (seed={seed}) → {:?}",
kept.len(),
args.out
);
if let Some(manifest_path) = &args.manifest {
let mut manifest = RunManifest::new("sample", &args.input);
manifest.input_hash = hash_file(&args.input)?;
manifest.records_read = total;
manifest.records_kept = kept.len();
manifest.records_dropped = total - kept.len();
accumulate_coverage(&mut manifest, &kept);
write_manifest(manifest_path, &manifest)?;
}
Ok(())
}
#[derive(serde::Serialize)]
struct OrderManifestLine {
training_position: u64,
sample_id: String,
game_id: String,
game_position_index: u32,
outcome: Option<GameOutcome>,
teacher_cp: Option<i32>,
teacher_depth: Option<u32>,
source_file: String,
split_id: Option<String>,
shuffle_seed: u64,
block_id: u64,
opening_id: Option<String>,
}
fn sample_id(rec: &PositionRecord) -> String {
hash_parts(&[
rec.sfen.as_bytes(),
rec.source.path.as_bytes(),
&rec.source.ply.to_le_bytes(),
])
.to_hex()
.to_string()
}
fn select_teacher<'a>(
rec: &'a PositionRecord,
engine: &str,
depth: u32,
) -> Option<&'a Observation> {
let matching: Vec<&Observation> = rec
.observations
.iter()
.filter(|o| o.engine == engine)
.collect();
find_at_depth(&matching, depth)
}
fn cmd_shuffle(args: ShuffleArgs) -> Result<()> {
if args.block_size == 0 {
anyhow::bail!("--block-size must be > 0");
}
let (mut ordered, _broken) = load_records(&args.input)?;
let total = ordered.len();
ordered.sort_by_cached_key(|r| {
let id = sample_id(r);
(seeded_hash(args.seed, &id), id)
});
let sample_ids: Vec<String> = ordered.iter().map(sample_id).collect();
let out_file =
File::create(&args.out).with_context(|| format!("cannot create {:?}", args.out))?;
let mut writer = BufWriter::new(out_file);
for record in &ordered {
serde_json::to_writer(&mut writer, record)?;
writer.write_all(b"\n")?;
}
writer.flush()?;
if let Some(order_manifest_path) = &args.order_manifest {
let order_file = File::create(order_manifest_path)
.with_context(|| format!("cannot create {order_manifest_path:?}"))?;
let mut order_writer = BufWriter::new(order_file);
for (i, rec) in ordered.iter().enumerate() {
let teacher = match (&args.teacher_engine, args.teacher_depth) {
(Some(engine), Some(depth)) => select_teacher(rec, engine, depth),
_ => None,
};
let (teacher_cp, teacher_depth) = match teacher {
Some(obs) => (
match obs.score {
Score::Cp { value } => Some(value),
Score::Mate { .. } => None,
},
Some(obs.depth),
),
None => (None, None),
};
let line = OrderManifestLine {
training_position: i as u64 + 1,
sample_id: sample_ids[i].clone(),
game_id: shogiesa_stratify::group_key(&rec.source),
game_position_index: rec.source.ply,
outcome: rec.game_result.as_ref().map(|g| g.outcome),
teacher_cp,
teacher_depth,
source_file: rec.source.path.clone(),
split_id: args.split_id.clone(),
shuffle_seed: args.seed,
block_id: i as u64 / args.block_size,
opening_id: None,
};
serde_json::to_writer(&mut order_writer, &line)?;
order_writer.write_all(b"\n")?;
}
order_writer.flush()?;
}
eprintln!(
"done: {total} shuffled (seed={}) -> {:?}",
args.seed, args.out
);
if let Some(manifest_path) = &args.manifest {
let mut manifest = RunManifest::new("shuffle", &args.input);
manifest.input_hash = hash_file(&args.input)?;
manifest.records_read = total;
manifest.records_kept = total;
let id_refs: Vec<&[u8]> = sample_ids.iter().map(|s| s.as_bytes()).collect();
manifest.order_hash = Some(hash_parts(&id_refs).to_hex().to_string());
manifest.experiment_id.clone_from(&args.experiment_id);
manifest.shuffle_seed = Some(args.seed);
write_manifest(manifest_path, &manifest)?;
}
Ok(())
}
fn eval_black(rec: &PositionRecord) -> Option<i32> {
rec.observations
.iter()
.max_by_key(|o| o.depth)
.and_then(|o| match o.score {
Score::Cp { value } => Some(cp_from_black_perspective(
value,
o.score_perspective,
rec.tags.side_to_move,
)),
Score::Mate { .. } => None,
})
}
fn blunder_adjacent_indices(
records: &[PositionRecord],
blunder_threshold: i32,
blunder_window: usize,
) -> HashSet<usize> {
let mut by_game: HashMap<String, Vec<usize>> = HashMap::new();
for (i, rec) in records.iter().enumerate() {
by_game.entry(rec.source.path.clone()).or_default().push(i);
}
for indices in by_game.values_mut() {
indices.sort_by_key(|&i| records[i].source.ply);
}
let mut keep = HashSet::<usize>::new();
for indices in by_game.values() {
let evals: Vec<Option<i32>> = indices.iter().map(|&i| eval_black(&records[i])).collect();
for j in 1..indices.len() {
if let (Some(e0), Some(e1)) = (evals[j - 1], evals[j])
&& (e1 - e0).abs() >= blunder_threshold
{
let lo = j.saturating_sub(blunder_window);
let hi = (j + blunder_window + 1).min(indices.len());
for k in lo..hi {
if !records[indices[k]].observations.is_empty() {
keep.insert(indices[k]);
}
}
}
}
}
keep
}
fn cmd_mine(args: MineArgs) -> Result<()> {
let (records, _) = load_records(&args.input)?;
let total = records.len();
let mut keep = blunder_adjacent_indices(&records, args.blunder_threshold, args.blunder_window);
let mut by_game: HashMap<String, Vec<usize>> = HashMap::new();
for (i, rec) in records.iter().enumerate() {
by_game.entry(rec.source.path.clone()).or_default().push(i);
}
for indices in by_game.values_mut() {
indices.sort_by_key(|&i| records[i].source.ply);
}
for indices in by_game.values() {
let evals: Vec<Option<i32>> = indices.iter().map(|&i| eval_black(&records[i])).collect();
if let Some(threshold) = args.losing_threshold {
for (j, &idx) in indices.iter().enumerate() {
if let Some(eval) = evals[j] {
let side_eval = match records[idx].tags.side_to_move {
SideToMove::Black => eval,
SideToMove::White => -eval,
};
if side_eval < -threshold && !records[idx].observations.is_empty() {
keep.insert(idx);
}
}
}
}
}
let out_file =
File::create(&args.out).with_context(|| format!("cannot create {:?}", args.out))?;
let mut writer = BufWriter::new(out_file);
let mut mined = 0usize;
for (i, rec) in records.iter().enumerate() {
if keep.contains(&i) {
serde_json::to_writer(&mut writer, rec)?;
writer.write_all(b"\n")?;
mined += 1;
}
}
writer.flush()?;
eprintln!(
"done: {mined}/{total} hard positions mined → {:?}",
args.out
);
Ok(())
}
fn cmd_balance(args: BalanceArgs) -> Result<()> {
if args.by.is_empty() {
anyhow::bail!("--by requires at least one of: phase, side, eval-bucket, wdl");
}
let by_phase = args.by.iter().any(|s| s == "phase");
let by_side = args.by.iter().any(|s| s == "side");
let by_eval = args.by.iter().any(|s| s == "eval-bucket");
let by_wdl = args.by.iter().any(|s| s == "wdl");
let mut bucket_sizes: HashMap<String, usize> = HashMap::new();
let pass1 = BufReader::new(
File::open(&args.input).with_context(|| format!("cannot open {:?}", args.input))?,
);
for line in pass1.lines() {
let line = line?;
if line.trim().is_empty() {
continue;
}
if let Ok(record) = serde_json::from_str::<PositionRecord>(&line) {
*bucket_sizes
.entry(bucket_key(&record, by_phase, by_side, by_eval, by_wdl))
.or_default() += 1;
}
}
let min_size = bucket_sizes.values().copied().min().unwrap_or(0);
let target = args.target.unwrap_or(min_size);
let pass2 = BufReader::new(
File::open(&args.input).with_context(|| format!("cannot open {:?}", args.input))?,
);
let mut total = 0usize;
let mut heaps: HashMap<String, BinaryHeap<HeapEntry<String, PositionRecord>>> = HashMap::new();
for (i, line) in pass2.lines().enumerate() {
let line = line?;
if line.trim().is_empty() {
continue;
}
let record: PositionRecord = match serde_json::from_str(&line) {
Ok(r) => r,
Err(e) => {
tracing::warn!(line = i + 1, "parse error: {e}");
continue;
}
};
let bucket = bucket_key(&record, by_phase, by_side, by_eval, by_wdl);
let key = record.sfen.clone();
let index = total;
total += 1;
push_bounded(
heaps.entry(bucket).or_default(),
target,
HeapEntry { key, index, record },
);
}
let mut kept: Vec<HeapEntry<String, PositionRecord>> =
heaps.into_values().flat_map(|h| h.into_vec()).collect();
kept.sort_by_key(|e| e.index);
let out_file =
File::create(&args.out).with_context(|| format!("cannot create {:?}", args.out))?;
let mut writer = BufWriter::new(out_file);
for entry in &kept {
serde_json::to_writer(&mut writer, &entry.record)?;
writer.write_all(b"\n")?;
}
writer.flush()?;
eprintln!(
"done: {}/{total} selected (target {target}/bucket, {} buckets) → {:?}",
kept.len(),
bucket_sizes.len(),
args.out
);
if let Some(manifest_path) = &args.manifest {
let mut manifest = RunManifest::new("balance", &args.input);
manifest.input_hash = hash_file(&args.input)?;
manifest.records_read = total;
manifest.records_kept = kept.len();
manifest.records_dropped = total - kept.len();
let kept_records: Vec<PositionRecord> = kept.into_iter().map(|e| e.record).collect();
accumulate_coverage(&mut manifest, &kept_records);
write_manifest(manifest_path, &manifest)?;
}
Ok(())
}
const STRATIFY_FORMAT_VERSION: u32 = 1;
fn load_stratify_quota_file(path: &Path) -> Result<QuotaSpec> {
let text =
fs::read_to_string(path).with_context(|| format!("cannot read quota file {path:?}"))?;
let file: QuotaSpec =
serde_json::from_str(&text).with_context(|| format!("cannot parse quota file {path:?}"))?;
if file.quotas.is_empty() {
anyhow::bail!("quota file {path:?} has an empty \"quotas\" map -- nothing to keep");
}
Ok(file)
}
fn cmd_stratify(args: StratifyArgs) -> Result<()> {
if let Some(template_path) = args.write_template.clone() {
return cmd_stratify_write_template(&args, &template_path);
}
let Some(quota_path) = args.quota.clone() else {
anyhow::bail!("stratify requires exactly one of --write-template or --quota");
};
let out = args
.out
.clone()
.ok_or_else(|| anyhow::anyhow!("--quota requires --out"))?;
cmd_stratify_apply(&args, "a_path, &out)
}
fn cmd_stratify_write_template(args: &StratifyArgs, template_path: &Path) -> Result<()> {
if args.by.is_empty() {
anyhow::bail!(
"--write-template requires --by (at least one of: phase, side, eval-bucket, wdl)"
);
}
let by_phase = args.by.iter().any(|s| s == "phase");
let by_side = args.by.iter().any(|s| s == "side");
let by_eval = args.by.iter().any(|s| s == "eval-bucket");
let by_wdl = args.by.iter().any(|s| s == "wdl");
let mut bucket_sizes: HashMap<String, usize> = HashMap::new();
let reader = BufReader::new(
File::open(&args.input).with_context(|| format!("cannot open {:?}", args.input))?,
);
for line in reader.lines() {
let line = line?;
if line.trim().is_empty() {
continue;
}
if let Ok(record) = serde_json::from_str::<PositionRecord>(&line) {
*bucket_sizes
.entry(bucket_key(&record, by_phase, by_side, by_eval, by_wdl))
.or_default() += 1;
}
}
let file = QuotaSpec {
stratify_format_version: STRATIFY_FORMAT_VERSION,
input: args.input.display().to_string(),
by: args.by.clone(),
quotas: bucket_sizes.into_iter().collect(),
};
fs::write(template_path, serde_json::to_string_pretty(&file)?)
.with_context(|| format!("cannot write {template_path:?}"))?;
eprintln!(
"done: {} bucket(s) observed → {:?} (edit \"quotas\" counts down, then run `stratify --quota`)",
file.quotas.len(),
template_path
);
Ok(())
}
fn cmd_stratify_apply(args: &StratifyArgs, quota_path: &Path, out: &Path) -> Result<()> {
let quota_file = load_stratify_quota_file(quota_path)?;
let by_phase = quota_file.by.iter().any(|s| s == "phase");
let by_side = quota_file.by.iter().any(|s| s == "side");
let by_eval = quota_file.by.iter().any(|s| s == "eval-bucket");
let by_wdl = quota_file.by.iter().any(|s| s == "wdl");
let seed = args.seed;
let reader = BufReader::new(
File::open(&args.input).with_context(|| format!("cannot open {:?}", args.input))?,
);
let mut io_error: Option<std::io::Error> = None;
let records = reader
.lines()
.enumerate()
.filter_map(|(i, line)| match line {
Ok(line) if line.trim().is_empty() => None,
Ok(line) => match serde_json::from_str::<PositionRecord>(&line) {
Ok(r) => Some(r),
Err(e) => {
tracing::warn!(line = i + 1, "parse error: {e}");
None
}
},
Err(e) => {
io_error.get_or_insert(e);
None
}
});
let result = group_aware_fill(
records,
"a_file.quotas,
seed,
|r| bucket_key(r, by_phase, by_side, by_eval, by_wdl),
|r| shogiesa_stratify::group_key(&r.source),
);
if let Some(e) = io_error {
return Err(e.into());
}
let out_file = File::create(out).with_context(|| format!("cannot create {out:?}"))?;
let mut writer = BufWriter::new(out_file);
for record in &result.kept {
serde_json::to_writer(&mut writer, record)?;
writer.write_all(b"\n")?;
}
writer.flush()?;
let over_quota = result.quota_candidates - result.kept.len();
eprintln!(
"done: {}/{} kept ({} bucket(s) in quota file) → {out:?}",
result.kept.len(),
result.total,
quota_file.quotas.len()
);
eprintln!("drop reasons:");
eprintln!(" bucket_not_in_quota {}", result.bucket_not_in_quota);
eprintln!(" over_quota {over_quota}");
if let Some(manifest_path) = &args.manifest {
let mut manifest = RunManifest::new("stratify", &args.input);
manifest.input_hash = hash_file(&args.input)?;
manifest.records_read = result.total;
manifest.records_kept = result.kept.len();
manifest.records_dropped = result.total - result.kept.len();
manifest
.drop_reasons
.insert("bucket_not_in_quota", result.bucket_not_in_quota);
manifest.drop_reasons.insert("over_quota", over_quota);
accumulate_coverage(&mut manifest, &result.kept);
manifest.max_root_share_in_any_bucket = result.max_group_share_in_any_bucket;
manifest.distinct_roots_kept = Some(result.distinct_groups_kept);
write_manifest(manifest_path, &manifest)?;
}
Ok(())
}
const GATE_OPENING_SELECTION_ALGORITHM_VERSION: &str = "gate-openings-group-aware-fill-v1";
fn piece_point_value(pt: PieceType) -> u32 {
match pt {
PieceType::Pawn => 1,
PieceType::Lance => 3,
PieceType::Knight => 3,
PieceType::Silver => 5,
PieceType::Gold => 6,
PieceType::Bishop => 8,
PieceType::Rook => 10,
PieceType::King => 0,
PieceType::ProPawn | PieceType::ProLance | PieceType::ProKnight | PieceType::ProSilver => 6,
PieceType::Horse => 10,
PieceType::Dragon => 12,
}
}
fn total_material(sfen: &str) -> Option<u32> {
const HAND_ORDER: [PieceType; 7] = [
PieceType::Rook,
PieceType::Bishop,
PieceType::Gold,
PieceType::Silver,
PieceType::Knight,
PieceType::Lance,
PieceType::Pawn,
];
let board = Board::from_sfen(sfen).ok()?;
let mut total = 0u32;
for row in &board.grid {
for (_, pt) in row.iter().flatten() {
total += piece_point_value(*pt);
}
}
for hand in &board.hand {
for (pt, &count) in HAND_ORDER.iter().zip(hand.iter()) {
total += piece_point_value(*pt) * u32::from(count);
}
}
Some(total)
}
fn gate_opening_dedup_key(sfen: &str) -> String {
sfen.split_whitespace()
.take(3)
.collect::<Vec<_>>()
.join(" ")
}
fn has_legal_move_lite(sfen: &str) -> Result<bool, shogi_usi_parser::Error> {
use shogi_usi_parser::FromUsi;
let pos = shogi_core::PartialPosition::from_usi(&format!("sfen {sfen}"))?;
Ok(!shogi_legality_lite::all_legal_moves_partial(&pos).is_empty())
}
fn is_legal_bestmove_lite(sfen: &str, mv: &str) -> Result<bool, shogi_usi_parser::Error> {
use shogi_core::ToUsi;
use shogi_usi_parser::FromUsi;
let pos = shogi_core::PartialPosition::from_usi(&format!("sfen {sfen}"))?;
Ok(shogi_legality_lite::all_legal_moves_partial(&pos)
.iter()
.any(|m| m.to_usi_owned() == mv))
}
#[cfg(test)]
mod has_legal_move_lite_tests {
use super::{has_legal_move_lite, is_legal_bestmove_lite};
#[test]
fn startpos_has_legal_moves() {
let startpos = "lnsgkgsnl/1r5b1/ppppppppp/9/9/9/PPPPPPPPP/1B5R1/LNSGKGSNL b - 1";
assert!(has_legal_move_lite(startpos).unwrap());
}
#[test]
fn checkmate_has_no_legal_moves() {
let mate = "lnsg1gsnl/5rkb1/ppppppp+Pp/9/9/9/PPPPPPP1P/1B5R1/LNSGKGSNL w P 8";
assert!(!has_legal_move_lite(mate).unwrap());
}
#[test]
fn is_legal_bestmove_lite_accepts_drop_and_promotion() {
let startpos = "lnsgkgsnl/1r5b1/ppppppppp/9/9/9/PPPPPPPPP/1B5R1/LNSGKGSNL b - 1";
assert!(is_legal_bestmove_lite(startpos, "7g7f").unwrap());
assert!(!is_legal_bestmove_lite(startpos, "5i5g").unwrap());
let drop_and_promo = "8k/9/1P7/9/9/9/9/9/8K b P 1";
assert!(is_legal_bestmove_lite(drop_and_promo, "P*5e").unwrap());
assert!(is_legal_bestmove_lite(drop_and_promo, "8c8b+").unwrap());
assert!(is_legal_bestmove_lite(drop_and_promo, "8c8b").unwrap());
assert!(!is_legal_bestmove_lite(drop_and_promo, "5e5d").unwrap());
}
}
#[cfg(test)]
mod total_material_tests {
use super::total_material;
#[test]
fn total_material_of_startpos_is_122() {
let startpos = "lnsgkgsnl/1r5b1/ppppppppp/9/9/9/PPPPPPPPP/1B5R1/LNSGKGSNL b - 1";
assert_eq!(total_material(startpos), Some(122));
}
#[test]
fn total_material_counts_hand_pieces_too() {
let sfen = "8k/9/9/9/9/9/9/9/8K b P 1";
assert_eq!(total_material(sfen), Some(1));
}
#[test]
fn total_material_of_bare_kings_is_zero() {
let sfen = "8k/9/9/9/9/9/9/9/8K b - 1";
assert_eq!(total_material(sfen), Some(0));
}
#[test]
fn total_material_rejects_unparseable_sfen() {
assert_eq!(total_material("not a real sfen"), None);
}
}
fn cmd_make_gate_openings(args: MakeGateOpeningsArgs) -> Result<()> {
let seed = args.seed;
let reader = BufReader::new(
File::open(&args.input).with_context(|| format!("cannot open {:?}", args.input))?,
);
let mut total = 0usize;
let mut invalid_sfen = 0usize;
let mut below_min_ply = 0usize;
let mut above_max_ply = 0usize;
let mut unplayable = 0usize;
let mut duplicate_sfen = 0usize;
let mut seen: HashSet<String> = HashSet::new();
let mut io_error: Option<std::io::Error> = None;
let filtered: Vec<PositionRecord> = reader
.lines()
.enumerate()
.filter_map(|(i, line)| {
let line = match line {
Ok(line) => line,
Err(e) => {
io_error.get_or_insert(e);
return None;
}
};
if line.trim().is_empty() {
return None;
}
let record: PositionRecord = match serde_json::from_str(&line) {
Ok(r) => r,
Err(e) => {
tracing::warn!(line = i + 1, "parse error: {e}");
return None;
}
};
total += 1;
if Sfen::parse(&record.sfen).is_err() {
invalid_sfen += 1;
return None;
}
if record.source.ply < args.min_ply {
below_min_ply += 1;
return None;
}
if args.max_ply.is_some_and(|max| record.source.ply > max) {
above_max_ply += 1;
return None;
}
if !args.allow_unplayable {
match has_legal_move_lite(&record.sfen) {
Ok(true) => {}
Ok(false) => {
unplayable += 1;
return None;
}
Err(e) => {
tracing::warn!(sfen = %record.sfen, "cannot evaluate legality, skipping: {e}");
unplayable += 1;
return None;
}
}
}
if !seen.insert(gate_opening_dedup_key(&record.sfen)) {
duplicate_sfen += 1;
return None;
}
Some(record)
})
.collect();
if let Some(e) = io_error {
return Err(e.into());
}
let quotas = BTreeMap::from([(String::new(), args.count)]);
let result = group_aware_fill(
filtered.into_iter(),
"as,
seed,
|_| String::new(),
|r| shogiesa_stratify::group_key(&r.source),
);
let kept = result.kept;
let out_file =
File::create(&args.out).with_context(|| format!("cannot create {:?}", args.out))?;
let mut writer = BufWriter::new(out_file);
for record in &kept {
writer.write_all(record.sfen.as_bytes())?;
writer.write_all(b"\n")?;
}
writer.flush()?;
let over_count = result.quota_candidates - kept.len();
eprintln!("done: {}/{total} kept → {:?}", kept.len(), args.out);
eprintln!("drop reasons:");
eprintln!(" invalid_sfen {invalid_sfen}");
eprintln!(" below_min_ply {below_min_ply}");
eprintln!(" above_max_ply {above_max_ply}");
eprintln!(" unplayable {unplayable}");
eprintln!(" duplicate_sfen {duplicate_sfen}");
eprintln!(" over_count {over_count}");
if let Some(manifest_path) = &args.manifest {
let mut manifest = RunManifest::new("make-gate-openings", &args.input);
manifest.input_hash = hash_file(&args.input)?;
manifest.records_read = total;
manifest.records_kept = kept.len();
manifest.records_dropped = total - kept.len();
manifest.drop_reasons.insert("invalid_sfen", invalid_sfen);
manifest.drop_reasons.insert("below_min_ply", below_min_ply);
manifest.drop_reasons.insert("above_max_ply", above_max_ply);
manifest.drop_reasons.insert("unplayable", unplayable);
manifest
.drop_reasons
.insert("duplicate_sfen", duplicate_sfen);
manifest.drop_reasons.insert("over_count", over_count);
accumulate_coverage(&mut manifest, &kept);
manifest.max_root_share_in_any_bucket = result.max_group_share_in_any_bucket;
manifest.distinct_roots_kept = Some(result.distinct_groups_kept);
manifest.selection_seed = Some(seed);
manifest.canonical_valid_count = Some(result.quota_candidates);
manifest.selection_algorithm_version = Some(GATE_OPENING_SELECTION_ALGORITHM_VERSION);
manifest.output_sha256 = Some(hash_file_sha256(&args.out)?);
let mut source_distribution: BTreeMap<String, usize> = BTreeMap::new();
let mut phase_distribution: BTreeMap<GamePhase, usize> = BTreeMap::new();
let mut material_distribution: BTreeMap<u32, usize> = BTreeMap::new();
for record in &kept {
*source_distribution
.entry(shogiesa_stratify::group_key(&record.source))
.or_default() += 1;
*phase_distribution.entry(record.tags.phase).or_default() += 1;
if let Some(material) = total_material(&record.sfen) {
*material_distribution.entry(material).or_default() += 1;
}
}
manifest.source_distribution = Some(source_distribution);
manifest.phase_distribution = Some(phase_distribution);
manifest.material_distribution = Some(material_distribution);
write_manifest(manifest_path, &manifest)?;
}
Ok(())
}
fn select_uncertain_streaming(args: &SelectArgs) -> Result<(usize, Vec<PositionRecord>)> {
let config = QualityConfig {
require_exact_score: true,
require_policy_margin: true,
require_requested_depth_reached: true,
require_engine_agreement: true,
min_policy_margin_cp: args.min_policy_margin_cp,
..Default::default()
};
let count = args.count;
let seed = args.seed;
let reader = BufReader::new(
File::open(&args.input).with_context(|| format!("cannot open {:?}", args.input))?,
);
let mut total = 0usize;
let mut heap: BinaryHeap<HeapEntry<(TotalF32, u64), PositionRecord>> =
BinaryHeap::with_capacity(count.saturating_add(1));
for (i, line) in reader.lines().enumerate() {
let line = line?;
if line.trim().is_empty() {
continue;
}
let record: PositionRecord = match serde_json::from_str(&line) {
Ok(r) => r,
Err(e) => {
tracing::warn!(line = i + 1, "parse error: {e}");
continue;
}
};
let score = evaluate_quality(&record, &config).score;
let key = (TotalF32(score), seeded_hash(seed, &record.sfen));
let index = total;
total += 1;
push_bounded(&mut heap, count, HeapEntry { key, index, record });
}
let ranked_records = heap
.into_sorted_vec()
.into_iter()
.map(|e| e.record)
.collect();
Ok((total, ranked_records))
}
fn select_coverage_streaming(args: &SelectArgs) -> Result<(usize, Vec<PositionRecord>)> {
let count = args.count;
let seed = args.seed;
let mut bucket_counts: HashMap<String, usize> = HashMap::new();
let pass1 = BufReader::new(
File::open(&args.input).with_context(|| format!("cannot open {:?}", args.input))?,
);
for line in pass1.lines() {
let line = line?;
if line.trim().is_empty() {
continue;
}
if let Ok(record) = serde_json::from_str::<PositionRecord>(&line) {
*bucket_counts
.entry(bucket_key(&record, true, true, true, false))
.or_default() += 1;
}
}
let pass2 = BufReader::new(
File::open(&args.input).with_context(|| format!("cannot open {:?}", args.input))?,
);
let mut total = 0usize;
let mut heap: BinaryHeap<HeapEntry<(usize, u64), PositionRecord>> =
BinaryHeap::with_capacity(count.saturating_add(1));
for (i, line) in pass2.lines().enumerate() {
let line = line?;
if line.trim().is_empty() {
continue;
}
let record: PositionRecord = match serde_json::from_str(&line) {
Ok(r) => r,
Err(e) => {
tracing::warn!(line = i + 1, "parse error: {e}");
continue;
}
};
let bucket_count = bucket_counts[&bucket_key(&record, true, true, true, false)];
let key = (bucket_count, seeded_hash(seed, &record.sfen));
let index = total;
total += 1;
push_bounded(&mut heap, count, HeapEntry { key, index, record });
}
let ranked_records = heap
.into_sorted_vec()
.into_iter()
.map(|e| e.record)
.collect();
Ok((total, ranked_records))
}
fn select_hard_materialized(args: &SelectArgs) -> Result<(usize, Vec<PositionRecord>)> {
let (records, _) = load_records(&args.input)?;
let total = records.len();
let count = args.count.min(total);
let seed = args.seed;
let blunder_set =
blunder_adjacent_indices(&records, args.blunder_threshold, args.blunder_window);
let hardness = |i: usize| -> (bool, bool, i32) {
let rec = &records[i];
let cp_scores: Vec<i32> = rec
.observations
.iter()
.filter_map(|o| match o.score {
Score::Cp { value } => Some(value),
Score::Mate { .. } => None,
})
.collect();
let swing = score_swing(&cp_scores).unwrap_or(0);
let disagreement = !bestmove_agreement(&rec.observations);
(blunder_set.contains(&i), disagreement, swing)
};
let mut ranked: Vec<usize> = (0..total).collect();
ranked.sort_by(|&a, &b| {
hardness(b).cmp(&hardness(a)).then_with(|| {
seeded_hash(seed, &records[a].sfen).cmp(&seeded_hash(seed, &records[b].sfen))
})
});
ranked.truncate(count);
let ranked_records = ranked.into_iter().map(|i| records[i].clone()).collect();
Ok((total, ranked_records))
}
fn hardness_for_group(
group: &[PositionRecord],
blunder_threshold: i32,
blunder_window: usize,
) -> Vec<(bool, bool, i32)> {
let evals: Vec<Option<i32>> = group.iter().map(eval_black).collect();
let mut is_blunder_adjacent = vec![false; group.len()];
for j in 1..group.len() {
if let (Some(e0), Some(e1)) = (evals[j - 1], evals[j])
&& (e1 - e0).abs() >= blunder_threshold
{
let lo = j.saturating_sub(blunder_window);
let hi = (j + blunder_window + 1).min(group.len());
for k in lo..hi {
if !group[k].observations.is_empty() {
is_blunder_adjacent[k] = true;
}
}
}
}
group
.iter()
.enumerate()
.map(|(i, rec)| {
let cp_scores: Vec<i32> = rec
.observations
.iter()
.filter_map(|o| match o.score {
Score::Cp { value } => Some(value),
Score::Mate { .. } => None,
})
.collect();
(
is_blunder_adjacent[i],
!bestmove_agreement(&rec.observations),
score_swing(&cp_scores).unwrap_or(0),
)
})
.collect()
}
type HardKey = (std::cmp::Reverse<(bool, bool, i32)>, u64);
fn flush_hard_group(
group: &mut Vec<PositionRecord>,
heap: &mut BinaryHeap<HeapEntry<HardKey, PositionRecord>>,
total: &mut usize,
count: usize,
seed: u64,
blunder_threshold: i32,
blunder_window: usize,
) {
if group.is_empty() {
return;
}
group.sort_by_key(|r| r.source.ply);
let hardness = hardness_for_group(group, blunder_threshold, blunder_window);
for (rec, h) in group.drain(..).zip(hardness) {
let key = (std::cmp::Reverse(h), seeded_hash(seed, &rec.sfen));
let index = *total;
*total += 1;
push_bounded(
heap,
count,
HeapEntry {
key,
index,
record: rec,
},
);
}
}
fn select_hard_grouped_streaming(args: &SelectArgs) -> Result<(usize, Vec<PositionRecord>)> {
let count = args.count;
let seed = args.seed;
let reader = BufReader::new(
File::open(&args.input).with_context(|| format!("cannot open {:?}", args.input))?,
);
let mut heap: BinaryHeap<HeapEntry<HardKey, PositionRecord>> =
BinaryHeap::with_capacity(count.saturating_add(1));
let mut total = 0usize;
let mut current_path: Option<String> = None;
let mut group: Vec<PositionRecord> = Vec::new();
for (i, line) in reader.lines().enumerate() {
let line = line?;
if line.trim().is_empty() {
continue;
}
let record: PositionRecord = match serde_json::from_str(&line) {
Ok(r) => r,
Err(e) => {
tracing::warn!(line = i + 1, "parse error: {e}");
continue;
}
};
if current_path.as_deref() != Some(record.source.path.as_str()) {
flush_hard_group(
&mut group,
&mut heap,
&mut total,
count,
seed,
args.blunder_threshold,
args.blunder_window,
);
current_path = Some(record.source.path.clone());
}
group.push(record);
}
flush_hard_group(
&mut group,
&mut heap,
&mut total,
count,
seed,
args.blunder_threshold,
args.blunder_window,
);
Ok((
total,
heap.into_sorted_vec()
.into_iter()
.map(|e| e.record)
.collect(),
))
}
#[derive(Debug, Serialize)]
struct LinePriorObservation {
sequence_id: String,
step: u32,
state: String,
action: String,
outcome: &'static str,
#[serde(skip_serializing_if = "Option::is_none")]
score: Option<f64>,
weight: f64,
source: String,
tags: Vec<String>,
}
#[derive(Debug, Serialize)]
struct LinePriorExportManifest {
shogiesa_version: &'static str,
git_sha: &'static str,
input_path: String,
input_hash: String,
out_path: String,
source: String,
max_ply: Option<u32>,
state_format: String,
action_format: String,
outcome_mode: String,
score_mode: String,
games_read: usize,
games_skipped: usize,
records_exported: usize,
sequence_count: usize,
outcome_distribution: BTreeMap<&'static str, usize>,
tag_distribution: BTreeMap<String, usize>,
unknown_outcome_count: usize,
}
fn cmd_lineprior_export(args: LinePriorExportArgs) -> Result<()> {
let paths = collect_game_paths(&args.input)?;
if paths.is_empty() {
anyhow::bail!("no .csa or .kif files found in {:?}", args.input);
}
let out_file =
File::create(&args.out).with_context(|| format!("cannot create {:?}", args.out))?;
let mut writer = BufWriter::new(out_file);
let mut total_games = 0usize;
let mut total_moves = 0usize;
let mut skipped_games = 0usize;
let mut sequence_ids: HashSet<String> = HashSet::new();
let mut outcome_distribution: BTreeMap<&'static str, usize> = BTreeMap::new();
let mut tag_distribution: BTreeMap<String, usize> = BTreeMap::new();
let mut input_hasher = blake3::Hasher::new();
for path in &paths {
total_games += 1;
let content = match fs::read_to_string(path) {
Ok(c) => c,
Err(e) => {
tracing::warn!(path = %path.display(), "skipped: {e}");
skipped_games += 1;
continue;
}
};
if args.manifest.is_some() {
fold_file_hash(&mut input_hasher, &content);
}
let source_path = path.to_string_lossy().into_owned();
let ext = path.extension().and_then(|e| e.to_str()).unwrap_or("");
let result = match ext {
"kif" | "ki2" => shogiesa_kif::extract_moves_from_str(&content, &source_path)
.map_err(|e| e.to_string()),
_ => shogiesa_csa::extract_moves_from_str(&content, &source_path)
.map_err(|e| e.to_string()),
};
let (raw_moves, _outcome, _reason) = match result {
Ok(v) => v,
Err(e) => {
tracing::warn!(path = %path.display(), "skipped: {e}");
skipped_games += 1;
continue;
}
};
for raw in &raw_moves {
if args.max_ply.is_some_and(|max| raw.ply > max) {
continue;
}
let sequence_id = split_root_key(&raw.source).to_string();
let outcome = raw.outcome.for_mover(raw.mover);
let phase = phase_from_ply(raw.ply).to_string();
sequence_ids.insert(sequence_id.clone());
*outcome_distribution.entry(outcome).or_insert(0) += 1;
*tag_distribution.entry(phase.clone()).or_insert(0) += 1;
let obs = LinePriorObservation {
sequence_id,
step: raw.ply,
state: raw.sfen_before.clone(),
action: raw.usi_move.clone(),
outcome,
score: None,
weight: 1.0,
source: args.source.clone(),
tags: vec!["shogi".to_string(), phase],
};
serde_json::to_writer(&mut writer, &obs)?;
writer.write_all(b"\n")?;
total_moves += 1;
}
}
writer.flush()?;
if let Some(manifest_path) = &args.manifest {
let unknown_outcome_count = outcome_distribution.get("unknown").copied().unwrap_or(0);
let manifest = LinePriorExportManifest {
shogiesa_version: env!("CARGO_PKG_VERSION"),
git_sha: env!("SHOGIESA_GIT_SHA"),
input_path: args.input.display().to_string(),
input_hash: input_hasher.finalize().to_hex().to_string(),
out_path: args.out.display().to_string(),
source: args.source.clone(),
max_ply: args.max_ply,
state_format: args.state_format.clone(),
action_format: args.action_format.clone(),
outcome_mode: args.outcome_mode.clone(),
score_mode: args.score_mode.clone(),
games_read: total_games,
games_skipped: skipped_games,
records_exported: total_moves,
sequence_count: sequence_ids.len(),
outcome_distribution,
tag_distribution,
unknown_outcome_count,
};
fs::write(manifest_path, serde_json::to_string_pretty(&manifest)?)
.with_context(|| format!("cannot write {manifest_path:?}"))?;
}
eprintln!(
"done: {total_games} games, {total_moves} moves exported, {skipped_games} games skipped → {:?}",
args.out
);
Ok(())
}
fn cmd_select(args: SelectArgs) -> Result<()> {
let (total, ranked_records) = match args.strategy.as_str() {
"uncertain" => select_uncertain_streaming(&args)?,
"coverage" => select_coverage_streaming(&args)?,
"hard" => {
if args.assume_grouped_by_source {
select_hard_grouped_streaming(&args)?
} else {
select_hard_materialized(&args)?
}
}
other => anyhow::bail!("unknown --strategy {other:?} (expected uncertain/hard/coverage)"),
};
let out_file =
File::create(&args.out).with_context(|| format!("cannot create {:?}", args.out))?;
let mut writer = BufWriter::new(out_file);
for rec in &ranked_records {
serde_json::to_writer(&mut writer, rec)?;
writer.write_all(b"\n")?;
}
writer.flush()?;
eprintln!(
"done: {}/{total} selected (strategy={}, seed={}) → {:?}",
ranked_records.len(),
args.strategy,
args.seed,
args.out
);
Ok(())
}
fn cmd_pack(args: PackArgs) -> Result<()> {
let reader = BufReader::new(
File::open(&args.input).with_context(|| format!("cannot open {:?}", args.input))?,
);
let out_file =
File::create(&args.out).with_context(|| format!("cannot create {:?}", args.out))?;
let mut writer = BufWriter::new(out_file);
pack::write_header(&mut writer)?;
let mut total = 0usize;
let mut skipped = 0usize;
let mut manifest = args
.manifest
.is_some()
.then(|| RunManifest::new("pack", &args.input));
let mut input_hasher = blake3::Hasher::new();
for (i, line) in reader.lines().enumerate() {
let line = line?;
if manifest.is_some() {
input_hasher.update(line.as_bytes());
input_hasher.update(b"\n");
}
if line.trim().is_empty() {
continue;
}
match serde_json::from_str::<PositionRecord>(&line) {
Ok(rec) => {
pack::encode_record(&rec, &mut writer)?;
total += 1;
if let Some(m) = &mut manifest {
accumulate_coverage(m, std::slice::from_ref(&rec));
}
}
Err(e) => {
tracing::warn!(line = i + 1, "JSON parse error: {e}");
skipped += 1;
}
}
}
writer.flush()?;
eprintln!("done: {total} packed, {skipped} skipped → {:?}", args.out);
if let (Some(mut m), Some(manifest_path)) = (manifest, &args.manifest) {
m.input_hash = input_hasher.finalize().to_hex().to_string();
m.records_read = total + skipped;
m.records_kept = total;
m.records_dropped = skipped;
write_manifest(manifest_path, &m)?;
}
Ok(())
}
fn cmd_unpack(args: UnpackArgs) -> Result<()> {
let mut reader = BufReader::new(
File::open(&args.input).with_context(|| format!("cannot open {:?}", args.input))?,
);
let out_file =
File::create(&args.out).with_context(|| format!("cannot create {:?}", args.out))?;
let mut writer = BufWriter::new(out_file);
pack::read_header(&mut reader)?;
let mut total = 0usize;
loop {
match pack::decode_record(&mut reader) {
Ok(rec) => {
serde_json::to_writer(&mut writer, &rec)?;
writer.write_all(b"\n")?;
total += 1;
}
Err(e) if e.kind() == std::io::ErrorKind::UnexpectedEof => break,
Err(e) => return Err(e.into()),
}
}
writer.flush()?;
eprintln!("done: {total} unpacked → {:?}", args.out);
Ok(())
}
fn parse_allowed_phases(phase: Option<&str>) -> Option<Vec<GamePhase>> {
phase.map(|s| {
s.split(',')
.filter_map(|p| match p.trim() {
"opening" => Some(GamePhase::Opening),
"middlegame" => Some(GamePhase::Middlegame),
"endgame" => Some(GamePhase::Endgame),
other => {
tracing::warn!("unknown phase {other:?}, ignoring");
None
}
})
.collect()
})
}
fn build_quality_config(
args: &FilterArgs,
allowed_phases: Option<Vec<GamePhase>>,
) -> QualityConfig {
QualityConfig {
min_observations: args.min_observations,
allowed_phases,
exclude_mate: args.exclude_mate,
exclude_in_check: args.exclude_in_check,
exclude_capture: args.exclude_capture,
exclude_timeout_salvaged: args.exclude_timeout_salvaged,
eval_min: args.eval_min,
eval_max: args.eval_max,
max_score_swing_cp: args.max_score_swing_cp,
min_policy_margin_cp: args.min_policy_margin_cp,
require_bestmove_agreement: args.require_bestmove_agreement,
require_engine_agreement: args.require_engine_agreement,
max_engine_score_swing_cp: args.max_engine_score_swing_cp,
require_exact_score: args.require_exact_score,
require_policy_margin: args.require_policy_margin,
min_depth_reached: args.min_depth_reached,
require_requested_depth_reached: args.require_requested_depth_reached,
allow_timeout_salvaged_mate: args.allow_timeout_salvaged_mate,
}
}
fn load_quality_config_preset(spec: &str) -> Result<QualityConfig> {
let (path_str, label) = spec
.rsplit_once(':')
.ok_or_else(|| anyhow::anyhow!("--preset must be FILE.json:label, got {spec:?}"))?;
let text = fs::read_to_string(path_str)
.with_context(|| format!("cannot read preset file {path_str:?}"))?;
let preset_file: TunePresetFile = serde_json::from_str(&text)
.with_context(|| format!("cannot parse preset file {path_str:?}"))?;
let candidate = preset_file.presets.get(label).ok_or_else(|| {
anyhow::anyhow!(
"preset {label:?} not found in {path_str:?}; available: {:?}",
preset_file.presets.keys().collect::<Vec<_>>()
)
})?;
Ok(candidate.config.clone())
}
fn cmd_filter(args: FilterArgs) -> Result<()> {
let config = match &args.preset {
Some(spec) => load_quality_config_preset(spec)?,
None => {
let allowed_phases = parse_allowed_phases(args.phase.as_deref());
build_quality_config(&args, allowed_phases)
}
};
let reader = BufReader::new(
File::open(&args.input).with_context(|| format!("cannot open {:?}", args.input))?,
);
let mut writer = match &args.out {
Some(out) => Some(BufWriter::new(
File::create(out).with_context(|| format!("cannot create {out:?}"))?,
)),
None => None,
};
let mut explain_writer = match &args.explain_out {
Some(path) => Some(BufWriter::new(
File::create(path).with_context(|| format!("cannot create {path:?}"))?,
)),
None => None,
};
let mut total = 0usize;
let mut passed = 0usize;
let mut skipped = 0usize;
let mut drop_reasons: BTreeMap<&'static str, usize> = BTreeMap::new();
let mut manifest = args
.manifest
.is_some()
.then(|| RunManifest::new("filter", &args.input));
let mut input_hasher = blake3::Hasher::new();
for (i, line) in reader.lines().enumerate() {
let line = line?;
if manifest.is_some() {
input_hasher.update(line.as_bytes());
input_hasher.update(b"\n");
}
if line.trim().is_empty() {
continue;
}
total += 1;
let rec: PositionRecord = match serde_json::from_str(&line) {
Ok(r) => r,
Err(e) => {
tracing::warn!(line = i + 1, "JSON parse error: {e}");
skipped += 1;
*drop_reasons.entry("parse_error").or_default() += 1;
continue;
}
};
let decision = evaluate_quality(&rec, &config);
match decision.reasons.first() {
None => {
if let Some(w) = &mut writer {
serde_json::to_writer(&mut *w, &rec)?;
w.write_all(b"\n")?;
}
passed += 1;
if let Some(m) = &mut manifest {
accumulate_coverage(m, std::slice::from_ref(&rec));
}
}
Some(reason) => {
skipped += 1;
*drop_reasons.entry(reason.as_str()).or_default() += 1;
if let Some(w) = &mut explain_writer {
serde_json::to_writer(
&mut *w,
&serde_json::json!({"record": &rec, "quality": &decision}),
)?;
w.write_all(b"\n")?;
}
}
}
}
if let Some(w) = &mut writer {
w.flush()?;
}
if let Some(w) = &mut explain_writer {
w.flush()?;
}
match &args.out {
Some(out) => eprintln!("done: {total} read, {passed} passed, {skipped} filtered → {out:?}"),
None => eprintln!("done (dry run): {total} read, {passed} passed, {skipped} filtered"),
}
if !drop_reasons.is_empty() {
eprintln!("drop reasons:");
for (reason, count) in &drop_reasons {
eprintln!(" {reason:<24} {count}");
}
}
if let (Some(mut m), Some(manifest_path)) = (manifest, &args.manifest) {
m.input_hash = input_hasher.finalize().to_hex().to_string();
m.records_read = total;
m.records_kept = passed;
m.records_dropped = skipped;
m.drop_reasons = drop_reasons;
m.filter_config = Some(config);
write_manifest(manifest_path, &m)?;
}
Ok(())
}
#[derive(Default)]
struct CoverageTally {
total: usize,
kept: usize,
dropped: usize,
drop_reasons: BTreeMap<&'static str, usize>,
}
impl CoverageTally {
fn record(&mut self, decision: &QualityDecision) {
self.total += 1;
match decision.reasons.first() {
None => self.kept += 1,
Some(reason) => {
self.dropped += 1;
*self.drop_reasons.entry(reason.as_str()).or_default() += 1;
}
}
}
fn coverage_pct(&self) -> f64 {
if self.total > 0 {
self.kept as f64 / self.total as f64 * 100.0
} else {
0.0
}
}
fn drop_reasons_csv_cell(&self) -> String {
self.drop_reasons
.iter()
.map(|(r, c)| format!("{r}={c}"))
.collect::<Vec<_>>()
.join(";")
}
}
struct SweepRow {
param: &'static str,
value: i32,
coverage: CoverageTally,
}
impl SweepRow {
fn new(param: &'static str, value: i32) -> Self {
Self {
param,
value,
coverage: CoverageTally::default(),
}
}
}
fn parse_int_list(s: &str) -> Result<Vec<i32>> {
s.split(',')
.map(|p| {
p.trim()
.parse::<i32>()
.with_context(|| format!("invalid integer {p:?} in sweep list"))
})
.collect()
}
#[derive(Default)]
struct DatasetDiagnostics {
labeled: usize,
special_bestmove: usize,
margin_buckets: BTreeMap<i32, usize>,
swing_buckets: BTreeMap<i32, usize>,
obs_score_bound_counts: BTreeMap<&'static str, usize>,
requested_depth_total: usize,
requested_depth_underreach: usize,
}
impl DatasetDiagnostics {
fn record(&mut self, rec: &PositionRecord) {
if rec.observations.is_empty() {
return;
}
self.labeled += 1;
if has_special_bestmove(&rec.observations) {
self.special_bestmove += 1;
}
let mut cp_scores = Vec::new();
for obs in &rec.observations {
if let Some(margin) = obs.policy_margin_cp {
*self
.margin_buckets
.entry((margin.div_euclid(50)) * 50)
.or_default() += 1;
}
if let Score::Cp { value } = obs.score {
cp_scores.push(value);
}
let bound_key = score_bound_str(obs.score_bound);
*self.obs_score_bound_counts.entry(bound_key).or_default() += 1;
}
if let Some(swing) = score_swing(&cp_scores) {
*self
.swing_buckets
.entry((swing.div_euclid(50)) * 50)
.or_default() += 1;
}
accumulate_requested_depth(
rec,
&mut self.requested_depth_total,
&mut self.requested_depth_underreach,
);
}
fn print(&self) {
let DatasetDiagnostics {
labeled,
special_bestmove,
margin_buckets,
swing_buckets,
obs_score_bound_counts,
requested_depth_total,
requested_depth_underreach,
} = self;
eprintln!("dataset-wide diagnostics (independent of swept thresholds):");
eprintln!(
" special bestmove: {special_bestmove:>6} ({:.1}% of labeled)",
*special_bestmove as f64 / (*labeled).max(1) as f64 * 100.0
);
if *requested_depth_total > 0 {
eprintln!(
" requested-depth underreach: {requested_depth_underreach:>6} ({:.1}% of {requested_depth_total} observations with a requested_depth)",
*requested_depth_underreach as f64 / *requested_depth_total as f64 * 100.0
);
}
if !obs_score_bound_counts.is_empty() {
eprintln!(" score bound (observations):");
for (bound, count) in obs_score_bound_counts {
eprintln!(" {bound:<10} : {count:>6}");
}
}
if !margin_buckets.is_empty() {
eprintln!(" policy_margin_cp distribution (50cp buckets):");
for (&key, &count) in margin_buckets {
eprintln!(" {key:>5}..{:<4}: {count:>6}", key + 49);
}
}
if !swing_buckets.is_empty() {
eprintln!(" score_swing_cp distribution (50cp buckets, per record):");
for (&key, &count) in swing_buckets {
eprintln!(" {key:>5}..{:<4}: {count:>6}", key + 49);
}
}
}
}
fn cmd_calibrate(args: CalibrateArgs) -> Result<()> {
if args.sweep_policy_margin.is_none() && args.sweep_score_swing.is_none() {
anyhow::bail!(
"calibrate requires at least one of --sweep-policy-margin/--sweep-score-swing"
);
}
let allowed_phases = parse_allowed_phases(args.phase.as_deref());
let base_config = QualityConfig {
min_observations: args.min_observations,
allowed_phases,
exclude_mate: args.exclude_mate,
exclude_in_check: args.exclude_in_check,
exclude_capture: args.exclude_capture,
exclude_timeout_salvaged: args.exclude_timeout_salvaged,
eval_min: args.eval_min,
eval_max: args.eval_max,
max_score_swing_cp: args.max_score_swing_cp,
min_policy_margin_cp: args.min_policy_margin_cp,
require_bestmove_agreement: args.require_bestmove_agreement,
require_engine_agreement: args.require_engine_agreement,
max_engine_score_swing_cp: args.max_engine_score_swing_cp,
require_exact_score: args.require_exact_score,
require_policy_margin: args.require_policy_margin,
min_depth_reached: args.min_depth_reached,
require_requested_depth_reached: args.require_requested_depth_reached,
allow_timeout_salvaged_mate: args.allow_timeout_salvaged_mate,
};
let mut policy_margin_rows: Vec<SweepRow> = match &args.sweep_policy_margin {
Some(s) => parse_int_list(s)?
.into_iter()
.map(|v| SweepRow::new("policy_margin", v))
.collect(),
None => Vec::new(),
};
let mut score_swing_rows: Vec<SweepRow> = match &args.sweep_score_swing {
Some(s) => parse_int_list(s)?
.into_iter()
.map(|v| SweepRow::new("score_swing", v))
.collect(),
None => Vec::new(),
};
let reader = BufReader::new(
File::open(&args.input).with_context(|| format!("cannot open {:?}", args.input))?,
);
let mut total = 0usize;
let mut diagnostics = DatasetDiagnostics::default();
for (i, line) in reader.lines().enumerate() {
let line = line?;
if line.trim().is_empty() {
continue;
}
let rec: PositionRecord = match serde_json::from_str(&line) {
Ok(r) => r,
Err(e) => {
tracing::warn!(line = i + 1, "JSON parse error: {e}");
continue;
}
};
total += 1;
for row in &mut policy_margin_rows {
let config = QualityConfig {
min_policy_margin_cp: Some(row.value),
..base_config.clone()
};
row.coverage.record(&evaluate_quality(&rec, &config));
}
for row in &mut score_swing_rows {
let config = QualityConfig {
max_score_swing_cp: Some(row.value),
..base_config.clone()
};
row.coverage.record(&evaluate_quality(&rec, &config));
}
diagnostics.record(&rec);
}
let out_file =
File::create(&args.out).with_context(|| format!("cannot create {:?}", args.out))?;
let mut writer = BufWriter::new(out_file);
writeln!(
writer,
"sweep_param,sweep_value,total,kept,dropped,coverage_pct,drop_reasons"
)?;
for row in policy_margin_rows.iter().chain(score_swing_rows.iter()) {
writeln!(
writer,
"{},{},{},{},{},{:.2},{}",
row.param,
row.value,
row.coverage.total,
row.coverage.kept,
row.coverage.dropped,
row.coverage.coverage_pct(),
row.coverage.drop_reasons_csv_cell()
)?;
}
writer.flush()?;
eprintln!(
"done: {total} read, {} labeled → {:?}",
diagnostics.labeled, args.out
);
diagnostics.print();
Ok(())
}
fn find_at_depth<'a>(
observations: &[&'a Observation],
target_depth: u32,
) -> Option<&'a Observation> {
observations
.iter()
.find(|o| o.requested_depth == Some(target_depth))
.or_else(|| observations.iter().find(|o| o.depth == target_depth))
.copied()
}
fn parse_u32_list(s: &str) -> Result<Vec<u32>> {
s.split(',')
.map(|p| {
p.trim()
.parse::<u32>()
.with_context(|| format!("invalid unsigned integer {p:?} in depth list"))
})
.collect()
}
#[derive(Default)]
struct AuditStats {
pairs: usize,
bestmove_mismatches: usize,
abs_score_error_sum: f64,
abs_score_error_count: usize,
abs_score_error_max: i32,
teacher_non_exact: usize,
student_non_exact: usize,
teacher_underreach: usize,
student_underreach: usize,
teacher_special_bestmove: usize,
student_special_bestmove: usize,
}
impl AuditStats {
fn record(
&mut self,
bestmove_match: bool,
score_error_cp: Option<i32>,
teacher: &Observation,
student: &Observation,
) {
self.pairs += 1;
if !bestmove_match {
self.bestmove_mismatches += 1;
}
if let Some(err) = score_error_cp {
let abs_err = err.abs();
self.abs_score_error_sum += abs_err as f64;
self.abs_score_error_count += 1;
self.abs_score_error_max = self.abs_score_error_max.max(abs_err);
}
if teacher.score_bound != shogiesa_core::ScoreBound::Exact {
self.teacher_non_exact += 1;
}
if student.score_bound != shogiesa_core::ScoreBound::Exact {
self.student_non_exact += 1;
}
if requested_depth_underreached(teacher) {
self.teacher_underreach += 1;
}
if requested_depth_underreached(student) {
self.student_underreach += 1;
}
if effective_bestmove_kind(teacher).is_some() {
self.teacher_special_bestmove += 1;
}
if effective_bestmove_kind(student).is_some() {
self.student_special_bestmove += 1;
}
}
fn print(&self, label: &str) {
if self.pairs == 0 {
return;
}
eprintln!(" {label}:");
eprintln!(" pairs compared : {:>6}", self.pairs);
eprintln!(
" bestmove mismatch : {:>6} ({:.1}%)",
self.bestmove_mismatches,
self.bestmove_mismatches as f64 / self.pairs as f64 * 100.0
);
if self.abs_score_error_count > 0 {
eprintln!(
" avg |score error| : {:.1}cp (max {}cp, over {} mate-free pairs)",
self.abs_score_error_sum / self.abs_score_error_count as f64,
self.abs_score_error_max,
self.abs_score_error_count
);
}
eprintln!(
" teacher non-exact : {:>6} ({:.1}%)",
self.teacher_non_exact,
self.teacher_non_exact as f64 / self.pairs as f64 * 100.0
);
eprintln!(
" student non-exact : {:>6} ({:.1}%)",
self.student_non_exact,
self.student_non_exact as f64 / self.pairs as f64 * 100.0
);
eprintln!(
" teacher underreach : {:>6} ({:.1}%)",
self.teacher_underreach,
self.teacher_underreach as f64 / self.pairs as f64 * 100.0
);
eprintln!(
" student underreach : {:>6} ({:.1}%)",
self.student_underreach,
self.student_underreach as f64 / self.pairs as f64 * 100.0
);
eprintln!(
" teacher special bm : {:>6} ({:.1}%)",
self.teacher_special_bestmove,
self.teacher_special_bestmove as f64 / self.pairs as f64 * 100.0
);
eprintln!(
" student special bm : {:>6} ({:.1}%)",
self.student_special_bestmove,
self.student_special_bestmove as f64 / self.pairs as f64 * 100.0
);
}
}
fn compute_audit_pairs<'a>(
rec: &'a PositionRecord,
teacher_depth: u32,
student_depths: &[u32],
) -> Vec<(u32, bool, Option<i32>, &'a Observation, &'a Observation)> {
let mut pairs = Vec::new();
let mut by_engine: HashMap<&str, Vec<&Observation>> = HashMap::new();
for obs in &rec.observations {
by_engine.entry(obs.engine.as_str()).or_default().push(obs);
}
for observations in by_engine.values() {
let Some(teacher) = find_at_depth(observations, teacher_depth) else {
continue;
};
for &student_depth in student_depths {
if student_depth == teacher_depth {
continue;
}
let Some(student) = find_at_depth(observations, student_depth) else {
continue;
};
let bestmove_match = bestmove_agreement(&[teacher.clone(), student.clone()]);
let score_error_cp = match (&teacher.score, &student.score) {
(Score::Cp { value: t }, Score::Cp { value: s }) => {
let t_black = cp_from_black_perspective(
*t,
teacher.score_perspective,
rec.tags.side_to_move,
);
let s_black = cp_from_black_perspective(
*s,
student.score_perspective,
rec.tags.side_to_move,
);
Some(t_black - s_black)
}
_ => None,
};
pairs.push((
student_depth,
bestmove_match,
score_error_cp,
teacher,
student,
));
}
}
pairs
}
fn cmd_audit(args: AuditArgs) -> Result<()> {
let student_depths = parse_u32_list(&args.student_depths)?;
let reader = BufReader::new(
File::open(&args.input).with_context(|| format!("cannot open {:?}", args.input))?,
);
let out_file =
File::create(&args.out).with_context(|| format!("cannot create {:?}", args.out))?;
let mut writer = BufWriter::new(out_file);
let mut total_records = 0usize;
let mut overall = AuditStats::default();
let mut per_depth: BTreeMap<u32, AuditStats> = BTreeMap::new();
for (i, line) in reader.lines().enumerate() {
let line = line?;
if line.trim().is_empty() {
continue;
}
let rec: PositionRecord = match serde_json::from_str(&line) {
Ok(r) => r,
Err(e) => {
tracing::warn!(line = i + 1, "parse error: {e}");
continue;
}
};
total_records += 1;
for (student_depth, bestmove_match, score_error_cp, teacher, student) in
compute_audit_pairs(&rec, args.teacher_depth, &student_depths)
{
overall.record(bestmove_match, score_error_cp, teacher, student);
per_depth.entry(student_depth).or_default().record(
bestmove_match,
score_error_cp,
teacher,
student,
);
serde_json::to_writer(
&mut writer,
&serde_json::json!({
"sfen": &rec.sfen,
"source": &rec.source,
"engine": &teacher.engine,
"teacher_requested_depth": teacher.requested_depth,
"teacher_depth": teacher.depth,
"teacher_score_bound": score_bound_str(teacher.score_bound),
"teacher_underreach": requested_depth_underreached(teacher),
"teacher_bestmove_kind": effective_bestmove_kind(teacher),
"student_requested_depth": student.requested_depth,
"student_depth": student.depth,
"student_score_bound": score_bound_str(student.score_bound),
"student_underreach": requested_depth_underreached(student),
"student_bestmove_kind": effective_bestmove_kind(student),
"bestmove_match": bestmove_match,
"score_error_cp": score_error_cp,
}),
)?;
writer.write_all(b"\n")?;
}
}
writer.flush()?;
eprintln!(
"done: {total_records} records read, {} pairs compared → {:?}",
overall.pairs, args.out
);
if overall.pairs == 0 {
eprintln!(
"(no (engine, student_depth) pair had both a teacher_depth={} and a listed student depth observation present)",
args.teacher_depth
);
return Ok(());
}
eprintln!("per student depth:");
for (&depth, stats) in &per_depth {
stats.print(&format!("student_depth={depth}"));
}
overall.print("overall");
Ok(())
}
struct TuneCell {
policy_margin: Option<i32>,
score_swing: Option<i32>,
coverage: CoverageTally,
audit: AuditStats,
}
impl TuneCell {
fn new(policy_margin: Option<i32>, score_swing: Option<i32>) -> Self {
Self {
policy_margin,
score_swing,
coverage: CoverageTally::default(),
audit: AuditStats::default(),
}
}
fn coverage_fraction(&self) -> f64 {
if self.coverage.total == 0 {
0.0
} else {
self.coverage.kept as f64 / self.coverage.total as f64
}
}
fn mismatch_rate(&self) -> f64 {
self.audit.bestmove_mismatches as f64 / self.audit.pairs as f64
}
fn lexicographic_key(&self) -> (i32, i32) {
(
self.policy_margin.unwrap_or(i32::MIN),
self.score_swing.unwrap_or(i32::MIN),
)
}
fn threshold_csv_cells(&self) -> (String, String) {
(
self.policy_margin
.map(|v| v.to_string())
.unwrap_or_default(),
self.score_swing.map(|v| v.to_string()).unwrap_or_default(),
)
}
fn to_csv_row(&self) -> String {
let (policy_margin, score_swing) = self.threshold_csv_cells();
let pct = |n: usize| -> String {
if self.audit.pairs == 0 {
String::new()
} else {
format!("{:.2}", n as f64 / self.audit.pairs as f64 * 100.0)
}
};
let avg_abs_score_error_cp = if self.audit.abs_score_error_count > 0 {
format!(
"{:.2}",
self.audit.abs_score_error_sum / self.audit.abs_score_error_count as f64
)
} else {
String::new()
};
let max_abs_score_error_cp = if self.audit.abs_score_error_count > 0 {
self.audit.abs_score_error_max.to_string()
} else {
String::new()
};
let total = self.coverage.total;
let kept = self.coverage.kept;
let dropped = self.coverage.dropped;
let coverage_pct = self.coverage.coverage_pct();
let drop_reasons = self.coverage.drop_reasons_csv_cell();
let audit_pairs = self.audit.pairs;
let teacher_bestmove_mismatch_pct = pct(self.audit.bestmove_mismatches);
let teacher_non_exact_pct = pct(self.audit.teacher_non_exact);
let student_non_exact_pct = pct(self.audit.student_non_exact);
let teacher_underreach_pct = pct(self.audit.teacher_underreach);
let student_underreach_pct = pct(self.audit.student_underreach);
let teacher_special_bestmove_pct = pct(self.audit.teacher_special_bestmove);
let student_special_bestmove_pct = pct(self.audit.student_special_bestmove);
format!(
"{policy_margin},{score_swing},{total},{kept},{dropped},{coverage_pct:.2},\
{drop_reasons},{audit_pairs},{teacher_bestmove_mismatch_pct},\
{avg_abs_score_error_cp},{max_abs_score_error_cp},{teacher_non_exact_pct},\
{student_non_exact_pct},{teacher_underreach_pct},{student_underreach_pct},\
{teacher_special_bestmove_pct},{student_special_bestmove_pct}"
)
}
}
fn dominates(a: &TuneCell, b: &TuneCell) -> bool {
let (a_cov, b_cov) = (a.coverage_fraction(), b.coverage_fraction());
let (a_mis, b_mis) = (a.mismatch_rate(), b.mismatch_rate());
let not_worse = a_cov >= b_cov && a_mis <= b_mis;
let strictly_better = a_cov > b_cov || a_mis < b_mis;
not_worse && strictly_better
}
fn pareto_frontier_indices(grid: &[TuneCell]) -> Vec<usize> {
let candidates: Vec<usize> = (0..grid.len())
.filter(|&i| grid[i].audit.pairs > 0)
.collect();
candidates
.iter()
.copied()
.filter(|&i| {
!candidates
.iter()
.any(|&j| j != i && dominates(&grid[j], &grid[i]))
})
.collect()
}
fn pick_broad(grid: &[TuneCell], frontier: &[usize]) -> usize {
let mut best = frontier[0];
for &i in &frontier[1..] {
let better = grid[i].coverage_fraction() > grid[best].coverage_fraction()
|| (grid[i].coverage_fraction() == grid[best].coverage_fraction()
&& grid[i].mismatch_rate() < grid[best].mismatch_rate())
|| (grid[i].coverage_fraction() == grid[best].coverage_fraction()
&& grid[i].mismatch_rate() == grid[best].mismatch_rate()
&& grid[i].lexicographic_key() < grid[best].lexicographic_key());
if better {
best = i;
}
}
best
}
fn pick_strict(grid: &[TuneCell], frontier: &[usize]) -> usize {
let mut best = frontier[0];
for &i in &frontier[1..] {
let better = grid[i].mismatch_rate() < grid[best].mismatch_rate()
|| (grid[i].mismatch_rate() == grid[best].mismatch_rate()
&& grid[i].coverage_fraction() > grid[best].coverage_fraction())
|| (grid[i].mismatch_rate() == grid[best].mismatch_rate()
&& grid[i].coverage_fraction() == grid[best].coverage_fraction()
&& grid[i].lexicographic_key() < grid[best].lexicographic_key());
if better {
best = i;
}
}
best
}
fn pick_balanced(grid: &[TuneCell], frontier: &[usize]) -> usize {
let covs = frontier.iter().map(|&i| grid[i].coverage_fraction());
let miss = frontier.iter().map(|&i| grid[i].mismatch_rate());
let cov_min = covs.clone().fold(f64::INFINITY, f64::min);
let cov_max = covs.fold(f64::NEG_INFINITY, f64::max);
let mis_min = miss.clone().fold(f64::INFINITY, f64::min);
let mis_max = miss.fold(f64::NEG_INFINITY, f64::max);
let normalize = |value: f64, lo: f64, hi: f64| {
if hi > lo {
(value - lo) / (hi - lo)
} else {
0.0
}
};
let mut best = frontier[0];
let mut best_dist = f64::INFINITY;
for &i in frontier {
let cov_norm = normalize(grid[i].coverage_fraction(), cov_min, cov_max);
let mis_norm = normalize(grid[i].mismatch_rate(), mis_min, mis_max);
let dist = ((1.0 - cov_norm).powi(2) + mis_norm.powi(2)).sqrt();
if dist < best_dist
|| (dist == best_dist && grid[i].lexicographic_key() < grid[best].lexicographic_key())
{
best = i;
best_dist = dist;
}
}
best
}
fn candidate_labels(idx: usize, broad: usize, balanced: usize, strict: usize) -> Vec<&'static str> {
[(broad, "broad"), (balanced, "balanced"), (strict, "strict")]
.into_iter()
.filter(|&(i, _)| i == idx)
.map(|(_, label)| label)
.collect()
}
fn write_tune_report(path: &Path, args: &TuneArgs, total: usize, grid: &[TuneCell]) -> Result<()> {
let mut report = String::new();
writeln!(report, "# Tuning Report")?;
writeln!(report)?;
writeln!(report, "Input: {:?}, {total} records read", args.input)?;
writeln!(
report,
"Teacher depth: {}. Student depths: {}.",
args.teacher_depth, args.student_depths
)?;
writeln!(report, "Grid: {} configurations", grid.len())?;
writeln!(report)?;
let frontier = pareto_frontier_indices(grid);
if frontier.is_empty() {
writeln!(
report,
"No grid cell had a teacher/student comparison pair -- check `--teacher-depth`/\
`--student-depths` against what's actually in the data. Pareto analysis skipped; \
see the CSV for coverage-only results."
)?;
fs::write(path, report)?;
return Ok(());
}
let broad = pick_broad(grid, &frontier);
let balanced = pick_balanced(grid, &frontier);
let strict = pick_strict(grid, &frontier);
writeln!(report, "## Why three candidates, not one")?;
writeln!(report)?;
writeln!(
report,
"Training-data needs vary by whether this round wants quantity or reliability -- \
shogiesa doesn't presume to know which, so it hands back the Pareto-optimal menu below \
instead of picking one \"correct\" threshold for you."
)?;
writeln!(report)?;
writeln!(report, "## Recommended candidates")?;
writeln!(report)?;
writeln!(
report,
"| candidate | policy_margin | score_swing | coverage | teacher_mismatch_rate | avg \\|score_error\\| cp | max \\|score_error\\| cp | audit pairs |"
)?;
writeln!(report, "|---|---|---|---|---|---|---|---|")?;
let mut seen = Vec::new();
for &idx in &[broad, balanced, strict] {
if seen.contains(&idx) {
continue;
}
seen.push(idx);
let cell = &grid[idx];
let labels = candidate_labels(idx, broad, balanced, strict).join(", ");
let (policy_margin, score_swing) = cell.threshold_csv_cells();
let avg_abs = if cell.audit.abs_score_error_count > 0 {
format!(
"{:.1}",
cell.audit.abs_score_error_sum / cell.audit.abs_score_error_count as f64
)
} else {
"n/a".to_string()
};
writeln!(
report,
"| {labels} | {policy_margin} | {score_swing} | {:.1}% | {:.1}% | {avg_abs} | {} | {} |",
cell.coverage_fraction() * 100.0,
cell.mismatch_rate() * 100.0,
cell.audit.abs_score_error_max,
cell.audit.pairs
)?;
}
if seen.len() < 3 {
writeln!(report)?;
writeln!(
report,
"(The Pareto frontier had fewer than 3 distinct points, so some candidates coincided \
on the same grid cell above.)"
)?;
}
writeln!(report)?;
writeln!(report, "## Pareto frontier")?;
writeln!(report)?;
let mut sorted_frontier = frontier.clone();
sorted_frontier.sort_by(|&a, &b| {
grid[b]
.coverage_fraction()
.total_cmp(&grid[a].coverage_fraction())
});
writeln!(
report,
"| policy_margin | score_swing | coverage | teacher_mismatch_rate | candidate |"
)?;
writeln!(report, "|---|---|---|---|---|")?;
for &idx in &sorted_frontier {
let cell = &grid[idx];
let (policy_margin, score_swing) = cell.threshold_csv_cells();
let labels = candidate_labels(idx, broad, balanced, strict).join(", ");
writeln!(
report,
"| {policy_margin} | {score_swing} | {:.1}% | {:.1}% | {labels} |",
cell.coverage_fraction() * 100.0,
cell.mismatch_rate() * 100.0
)?;
}
writeln!(report)?;
writeln!(report, "## Full grid")?;
writeln!(report)?;
if grid.len() <= 50 {
writeln!(
report,
"| policy_margin | score_swing | coverage | kept | dropped | audit pairs | teacher_mismatch_rate |"
)?;
writeln!(report, "|---|---|---|---|---|---|---|")?;
for cell in grid {
let (policy_margin, score_swing) = cell.threshold_csv_cells();
let mismatch = if cell.audit.pairs > 0 {
format!("{:.1}%", cell.mismatch_rate() * 100.0)
} else {
"n/a".to_string()
};
writeln!(
report,
"| {policy_margin} | {score_swing} | {:.1}% | {} | {} | {} | {mismatch} |",
cell.coverage_fraction() * 100.0,
cell.coverage.kept,
cell.coverage.dropped,
cell.audit.pairs
)?;
}
} else {
writeln!(
report,
"See `{:?}` for the full {}-row grid.",
args.out,
grid.len()
)?;
}
fs::write(path, report)?;
Ok(())
}
const TUNE_PRESET_FORMAT_VERSION: u32 = 1;
#[derive(Debug, Clone, Serialize, Deserialize)]
struct TunePresetCandidate {
config: QualityConfig,
coverage_fraction: f64,
#[serde(skip_serializing_if = "Option::is_none")]
mismatch_rate: Option<f64>,
audit_pairs: usize,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
struct TunePresetFile {
preset_format_version: u32,
tool: String,
input: String,
teacher_depth: u32,
student_depths: String,
presets: BTreeMap<String, TunePresetCandidate>,
}
fn write_tune_preset(
path: &Path,
args: &TuneArgs,
base_config: &QualityConfig,
grid: &[TuneCell],
) -> Result<()> {
let mut presets = BTreeMap::new();
let frontier = pareto_frontier_indices(grid);
if !frontier.is_empty() {
let broad = pick_broad(grid, &frontier);
let balanced = pick_balanced(grid, &frontier);
let strict = pick_strict(grid, &frontier);
for (label, idx) in [("broad", broad), ("balanced", balanced), ("strict", strict)] {
let cell = &grid[idx];
let config = QualityConfig {
min_policy_margin_cp: cell.policy_margin,
max_score_swing_cp: cell.score_swing,
..base_config.clone()
};
presets.insert(
label.to_string(),
TunePresetCandidate {
config,
coverage_fraction: cell.coverage_fraction(),
mismatch_rate: (cell.audit.pairs > 0).then(|| cell.mismatch_rate()),
audit_pairs: cell.audit.pairs,
},
);
}
}
let file = TunePresetFile {
preset_format_version: TUNE_PRESET_FORMAT_VERSION,
tool: "shogiesa tune".to_string(),
input: args.input.display().to_string(),
teacher_depth: args.teacher_depth,
student_depths: args.student_depths.clone(),
presets,
};
fs::write(path, serde_json::to_string_pretty(&file)?)?;
Ok(())
}
fn cmd_tune(args: TuneArgs) -> Result<()> {
if args.sweep_policy_margin.is_none() && args.sweep_score_swing.is_none() {
anyhow::bail!("tune requires at least one of --sweep-policy-margin/--sweep-score-swing");
}
let student_depths = parse_u32_list(&args.student_depths)?;
let allowed_phases = parse_allowed_phases(args.phase.as_deref());
let base_config = QualityConfig {
min_observations: args.min_observations,
allowed_phases,
exclude_mate: args.exclude_mate,
exclude_in_check: args.exclude_in_check,
exclude_capture: args.exclude_capture,
exclude_timeout_salvaged: args.exclude_timeout_salvaged,
eval_min: args.eval_min,
eval_max: args.eval_max,
max_score_swing_cp: args.max_score_swing_cp,
min_policy_margin_cp: args.min_policy_margin_cp,
require_bestmove_agreement: args.require_bestmove_agreement,
require_engine_agreement: args.require_engine_agreement,
max_engine_score_swing_cp: args.max_engine_score_swing_cp,
require_exact_score: args.require_exact_score,
require_policy_margin: args.require_policy_margin,
min_depth_reached: args.min_depth_reached,
require_requested_depth_reached: args.require_requested_depth_reached,
allow_timeout_salvaged_mate: args.allow_timeout_salvaged_mate,
};
let policy_values: Vec<Option<i32>> = match &args.sweep_policy_margin {
Some(s) => parse_int_list(s)?.into_iter().map(Some).collect(),
None => vec![args.min_policy_margin_cp],
};
let swing_values: Vec<Option<i32>> = match &args.sweep_score_swing {
Some(s) => parse_int_list(s)?.into_iter().map(Some).collect(),
None => vec![args.max_score_swing_cp],
};
let mut grid: Vec<TuneCell> = Vec::with_capacity(policy_values.len() * swing_values.len());
for &policy_margin in &policy_values {
for &score_swing_value in &swing_values {
grid.push(TuneCell::new(policy_margin, score_swing_value));
}
}
let reader = BufReader::new(
File::open(&args.input).with_context(|| format!("cannot open {:?}", args.input))?,
);
let mut total = 0usize;
let mut diagnostics = DatasetDiagnostics::default();
for (i, line) in reader.lines().enumerate() {
let line = line?;
if line.trim().is_empty() {
continue;
}
let rec: PositionRecord = match serde_json::from_str(&line) {
Ok(r) => r,
Err(e) => {
tracing::warn!(line = i + 1, "JSON parse error: {e}");
continue;
}
};
total += 1;
diagnostics.record(&rec);
let audit_pairs = compute_audit_pairs(&rec, args.teacher_depth, &student_depths);
for cell in &mut grid {
let config = QualityConfig {
min_policy_margin_cp: cell.policy_margin,
max_score_swing_cp: cell.score_swing,
..base_config.clone()
};
let decision = evaluate_quality(&rec, &config);
let kept = decision.keep;
cell.coverage.record(&decision);
if kept {
for &(_, bestmove_match, score_error_cp, teacher, student) in &audit_pairs {
cell.audit
.record(bestmove_match, score_error_cp, teacher, student);
}
}
}
}
let out_file =
File::create(&args.out).with_context(|| format!("cannot create {:?}", args.out))?;
let mut writer = BufWriter::new(out_file);
writeln!(
writer,
"policy_margin,score_swing,total,kept,dropped,coverage_pct,drop_reasons,audit_pairs,\
teacher_bestmove_mismatch_pct,avg_abs_score_error_cp,max_abs_score_error_cp,\
teacher_non_exact_pct,student_non_exact_pct,teacher_underreach_pct,student_underreach_pct,\
teacher_special_bestmove_pct,student_special_bestmove_pct"
)?;
for cell in &grid {
writeln!(writer, "{}", cell.to_csv_row())?;
}
writer.flush()?;
eprintln!(
"done: {total} read, {} grid configurations → {:?}",
grid.len(),
args.out
);
diagnostics.print();
if let Some(report_path) = &args.report {
write_tune_report(report_path, &args, total, &grid)?;
eprintln!("report → {report_path:?}");
}
if let Some(preset_path) = &args.preset_out {
write_tune_preset(preset_path, &args, &base_config, &grid)?;
eprintln!("preset → {preset_path:?}");
}
Ok(())
}
fn walk_cache_entries(cache_dir: &Path) -> Result<Vec<PathBuf>> {
let mut entries = Vec::new();
for shard in fs::read_dir(cache_dir).with_context(|| format!("cannot read {cache_dir:?}"))? {
let shard_path = shard?.path();
if !shard_path.is_dir() {
continue;
}
for entry in
fs::read_dir(&shard_path).with_context(|| format!("cannot read {shard_path:?}"))?
{
let path = entry?.path();
if path.extension().and_then(|e| e.to_str()) == Some("json") {
entries.push(path);
}
}
}
Ok(entries)
}
fn cmd_cache_stats(args: CacheStatsArgs) -> Result<()> {
let entries = walk_cache_entries(&args.cache_dir)?;
let now = std::time::SystemTime::now();
let mut total_bytes = 0u64;
let mut oldest: Option<std::time::SystemTime> = None;
let mut newest: Option<std::time::SystemTime> = None;
let mut engine_counts: BTreeMap<String, usize> = BTreeMap::new();
let mut legacy_v1_count = 0usize;
let mut schema_version_counts: BTreeMap<u32, usize> = BTreeMap::new();
let mut fingerprint_counts: BTreeMap<String, usize> = BTreeMap::new();
let mut requested_depth_counts: BTreeMap<u32, usize> = BTreeMap::new();
let mut multipv_counts: BTreeMap<u32, usize> = BTreeMap::new();
let mut search_limit_kind_counts: BTreeMap<&'static str, usize> = BTreeMap::new();
for path in &entries {
if let Ok(meta) = fs::metadata(path) {
total_bytes += meta.len();
if let Ok(modified) = meta.modified() {
oldest = Some(oldest.map_or(modified, |o| o.min(modified)));
newest = Some(newest.map_or(modified, |n| n.max(modified)));
}
}
if let Some(parsed) = fs::read_to_string(path)
.ok()
.and_then(|s| parse_cache_entry(&s))
{
*engine_counts
.entry(parsed.observation().engine.clone())
.or_default() += 1;
match &parsed {
CacheRead::V1(_) => legacy_v1_count += 1,
CacheRead::V2(entry) => {
*schema_version_counts
.entry(entry.schema_version)
.or_default() += 1;
let fingerprint_key = entry
.engine_fingerprint
.map(|fp| format!("{fp:016x}"))
.unwrap_or_else(|| "none".to_string());
*fingerprint_counts.entry(fingerprint_key).or_default() += 1;
*requested_depth_counts
.entry(entry.requested_depth)
.or_default() += 1;
*multipv_counts.entry(entry.multipv).or_default() += 1;
let kind_key = match entry.search_limit_kind {
SearchLimitKind::Depth => "depth",
SearchLimitKind::Nodes => "nodes",
};
*search_limit_kind_counts.entry(kind_key).or_default() += 1;
}
}
}
}
println!("cache entries : {}", entries.len());
println!("total size : {total_bytes} bytes");
if let Some(oldest) = oldest {
let days = now.duration_since(oldest).unwrap_or_default().as_secs() / 86400;
println!("oldest entry : {days} days old");
}
if let Some(newest) = newest {
let days = now.duration_since(newest).unwrap_or_default().as_secs() / 86400;
println!("newest entry : {days} days old");
}
if !engine_counts.is_empty() {
println!("engine distribution:");
for (engine, count) in &engine_counts {
println!(" {engine:<20} {count:>6}");
}
}
if legacy_v1_count > 0 {
println!("legacy (v1, no metadata): {legacy_v1_count} entries");
}
if !schema_version_counts.is_empty() {
println!("schema_version distribution (v2 entries only):");
for (v, count) in &schema_version_counts {
println!(" {v:<20} {count:>6}");
}
}
if !fingerprint_counts.is_empty() {
println!("engine_fingerprint distribution (v2 entries only):");
for (fp, count) in &fingerprint_counts {
println!(" {fp:<20} {count:>6}");
}
}
if !requested_depth_counts.is_empty() {
println!("requested_depth distribution (v2 entries only):");
for (d, count) in &requested_depth_counts {
println!(" {d:<20} {count:>6}");
}
}
if !multipv_counts.is_empty() {
println!("multipv distribution (v2 entries only):");
for (m, count) in &multipv_counts {
println!(" {m:<20} {count:>6}");
}
}
if !search_limit_kind_counts.is_empty() {
println!("search_limit_kind distribution (v2 entries only):");
for (k, count) in &search_limit_kind_counts {
println!(" {k:<20} {count:>6}");
}
}
Ok(())
}
fn cmd_cache_verify(args: CacheVerifyArgs) -> Result<()> {
let entries = walk_cache_entries(&args.cache_dir)?;
let mut corrupted = 0usize;
let mut legacy_v1_count = 0usize;
let mut schema_version_counts: BTreeMap<u32, usize> = BTreeMap::new();
for path in &entries {
match fs::read_to_string(path)
.ok()
.and_then(|s| parse_cache_entry(&s))
{
Some(CacheRead::V1(_)) => legacy_v1_count += 1,
Some(CacheRead::V2(entry)) => {
*schema_version_counts
.entry(entry.schema_version)
.or_default() += 1;
}
None => {
corrupted += 1;
tracing::warn!(path = %path.display(), "corrupted cache entry");
}
}
}
println!("cache entries : {}", entries.len());
println!("corrupted : {corrupted}");
if legacy_v1_count > 0 {
println!("legacy (v1, no metadata): {legacy_v1_count} entries");
}
if !schema_version_counts.is_empty() {
println!("schema_version distribution (v2 entries only):");
for (v, count) in &schema_version_counts {
println!(" {v:<20} {count:>6}");
}
}
println!(
"note: v1 (legacy) entries store no schema_version/engine_fingerprint metadata; v2 \
entries do (see the distribution above). Neither format supports a live \"does this \
match today's engine/schema\" check here -- a schema bump or engine change already \
changes future cache keys by construction (see label_cache_path), so a stale entry is \
never wrongly reused, just orphaned."
);
Ok(())
}
fn file_age(path: &Path, now: std::time::SystemTime) -> Option<std::time::Duration> {
let modified = fs::metadata(path).ok()?.modified().ok()?;
now.duration_since(modified).ok()
}
fn cmd_cache_prune(args: CachePruneArgs) -> Result<()> {
if !args.corrupted_only && !args.legacy_only && args.older_than_days.is_none() {
anyhow::bail!(
"cache prune requires --corrupted-only, --legacy-only, and/or --older-than-days"
);
}
let entries = walk_cache_entries(&args.cache_dir)?;
let now = std::time::SystemTime::now();
let max_age = args
.older_than_days
.map(|days| std::time::Duration::from_secs(days * 86400));
let mut to_delete = Vec::new();
for path in &entries {
let mut matched = false;
if args.corrupted_only || args.legacy_only {
let parsed = fs::read_to_string(path)
.ok()
.and_then(|s| parse_cache_entry(&s));
match &parsed {
None if args.corrupted_only => matched = true,
Some(CacheRead::V1(_)) if args.legacy_only => matched = true,
_ => {}
}
}
if let Some(max_age) = max_age
&& file_age(path, now).is_some_and(|age| age >= max_age)
{
matched = true;
}
if matched {
to_delete.push(path.clone());
}
}
if args.yes {
let mut deleted = 0usize;
for path in &to_delete {
match fs::remove_file(path) {
Ok(()) => deleted += 1,
Err(e) => tracing::warn!(path = %path.display(), "failed to delete: {e}"),
}
}
eprintln!(
"deleted {deleted}/{} matched entries ({} total)",
to_delete.len(),
entries.len()
);
} else {
eprintln!(
"dry run: {}/{} entries matched and would be deleted (pass --yes to actually delete)",
to_delete.len(),
entries.len()
);
}
Ok(())
}
fn collect_game_paths(input: &PathBuf) -> Result<Vec<PathBuf>> {
if input.is_file() {
return Ok(vec![input.clone()]);
}
let mut paths = Vec::new();
for entry in
fs::read_dir(input).with_context(|| format!("cannot read directory {:?}", input))?
{
let entry = entry?;
let p = entry.path();
if matches!(
p.extension().and_then(|e| e.to_str()),
Some("csa" | "kif" | "ki2")
) {
paths.push(p);
}
}
paths.sort();
Ok(paths)
}
fn load_records(path: &PathBuf) -> Result<(Vec<PositionRecord>, usize)> {
let content = fs::read_to_string(path).with_context(|| format!("cannot read {:?}", path))?;
let mut broken = 0usize;
let records: Vec<PositionRecord> = content
.lines()
.filter(|l| !l.trim().is_empty())
.enumerate()
.filter_map(
|(i, line)| match serde_json::from_str::<PositionRecord>(line) {
Ok(rec) => Some(rec),
Err(e) => {
tracing::warn!(line = i + 1, "parse error: {e}");
broken += 1;
None
}
},
)
.collect();
Ok((records, broken))
}
fn cmd_report(args: ReportArgs) -> Result<()> {
let reader = BufReader::new(
File::open(&args.input).with_context(|| format!("cannot open {:?}", args.input))?,
);
let mut n = 0usize;
let mut broken = 0usize;
let mut phases = BTreeMap::<String, usize>::new();
let mut sides = BTreeMap::<String, usize>::new();
let mut wdl_counts = BTreeMap::<&'static str, usize>::new();
let mut result_source_counts = BTreeMap::<String, usize>::new();
let mut schema_versions = BTreeMap::<u32, usize>::new();
let mut ply_sum = 0u64;
let mut ply_min = u32::MAX;
let mut ply_max = 0u32;
let mut sfen_counts: HashMap<String, usize> = HashMap::new();
let mut tag_mismatches = 0usize;
let mut invalid_sfens = 0usize;
let mut labeled = 0usize;
let mut in_check = 0usize;
let mut has_capture = 0usize;
let mut depth_disagree = 0usize;
let mut multi_engine = 0usize;
let mut engine_disagree = 0usize;
let mut special_bestmove = 0usize;
let mut depth_counts: BTreeMap<u32, usize> = BTreeMap::new();
let mut eval_buckets: BTreeMap<i32, usize> = BTreeMap::new();
let mut eval_by_phase: BTreeMap<i32, BTreeMap<String, usize>> = BTreeMap::new();
let mut eval_by_side: BTreeMap<i32, BTreeMap<String, usize>> = BTreeMap::new();
let mut cp_obs_count = 0usize;
let mut mate_obs_count = 0usize;
let mut swing_sum = 0f64;
let mut swing_count = 0usize;
let mut swing_buckets: BTreeMap<i32, usize> = BTreeMap::new();
let mut margin_sum = 0f64;
let mut margin_count = 0usize;
let mut obs_score_bound_counts: BTreeMap<&'static str, usize> = BTreeMap::new();
let mut sources = BTreeMap::<String, usize>::new();
let mut obs_with_candidates = 0usize;
let mut obs_total = 0usize;
let mut score_bound_distribution: BTreeMap<&'static str, usize> = BTreeMap::new();
let mut requested_depth_total = 0usize;
let mut requested_depth_underreach = 0usize;
for (i, line) in reader.lines().enumerate() {
let line = line?;
if line.trim().is_empty() {
continue;
}
let rec: PositionRecord = match serde_json::from_str(&line) {
Ok(r) => r,
Err(e) => {
tracing::warn!(line = i + 1, "parse error: {e}");
broken += 1;
continue;
}
};
n += 1;
*phases.entry(format!("{}", rec.tags.phase)).or_default() += 1;
*sides
.entry(format!("{}", rec.tags.side_to_move))
.or_default() += 1;
*schema_versions.entry(rec.schema_version).or_default() += 1;
*wdl_counts
.entry(shogiesa_stratify::feature_wdl(&rec))
.or_default() += 1;
*result_source_counts
.entry(
rec.game_result
.as_ref()
.map(|gr| gr.result_source.clone())
.unwrap_or_else(|| "no_game_result".to_string()),
)
.or_default() += 1;
if rec.tags.in_check {
in_check += 1;
}
if rec.tags.has_capture {
has_capture += 1;
}
let ply = rec.source.ply;
ply_sum += ply as u64;
ply_min = ply_min.min(ply);
ply_max = ply_max.max(ply);
*sfen_counts.entry(rec.sfen.clone()).or_default() += 1;
*sources.entry(rec.source.path.clone()).or_default() += 1;
match Sfen::parse(&rec.sfen) {
Ok(sfen) => {
if sfen.side_to_move() != rec.tags.side_to_move {
tag_mismatches += 1;
}
}
Err(_) => invalid_sfens += 1,
}
if rec.observations.is_empty() {
*eval_buckets.entry(i32::MIN).or_default() += 1; *eval_by_phase
.entry(i32::MIN)
.or_default()
.entry(format!("{}", rec.tags.phase))
.or_default() += 1;
*eval_by_side
.entry(i32::MIN)
.or_default()
.entry(format!("{}", rec.tags.side_to_move))
.or_default() += 1;
} else {
labeled += 1;
if !bestmove_agreement(&rec.observations) {
depth_disagree += 1;
}
if has_special_bestmove(&rec.observations) {
special_bestmove += 1;
}
if let Some(agree) = engine_bestmove_agreement(&rec.observations) {
multi_engine += 1;
if !agree {
engine_disagree += 1;
}
}
let mut cp_scores = Vec::new();
for obs in &rec.observations {
*depth_counts.entry(obs.depth).or_default() += 1;
match obs.score {
Score::Cp { value } => {
cp_obs_count += 1;
cp_scores.push(value);
}
Score::Mate { .. } => mate_obs_count += 1,
}
if let Some(margin) = obs.policy_margin_cp {
margin_sum += margin as f64;
margin_count += 1;
}
let bound_key = score_bound_str(obs.score_bound);
*obs_score_bound_counts.entry(bound_key).or_default() += 1;
}
if let Some(swing) = score_swing(&cp_scores) {
swing_sum += swing as f64;
swing_count += 1;
*swing_buckets
.entry((swing.div_euclid(50)) * 50)
.or_default() += 1;
}
if let Some(deepest) = rec.observations.iter().max_by_key(|o| o.depth) {
let key = match deepest.score {
Score::Cp { value } => {
let black_value = cp_from_black_perspective(
value,
deepest.score_perspective,
rec.tags.side_to_move,
);
(black_value.div_euclid(200)) * 200
}
Score::Mate { .. } => i32::MAX, };
*eval_buckets.entry(key).or_default() += 1;
*eval_by_phase
.entry(key)
.or_default()
.entry(format!("{}", rec.tags.phase))
.or_default() += 1;
*eval_by_side
.entry(key)
.or_default()
.entry(format!("{}", rec.tags.side_to_move))
.or_default() += 1;
}
accumulate_candidate_coverage(
&rec,
&mut obs_with_candidates,
&mut obs_total,
&mut score_bound_distribution,
);
accumulate_requested_depth(
&rec,
&mut requested_depth_total,
&mut requested_depth_underreach,
);
}
}
if n == 0 {
println!("no valid records in {:?}", args.input);
return Ok(());
}
let duplicate_sfens: usize = sfen_counts
.values()
.filter(|&&c| c > 1)
.map(|&c| c - 1)
.sum();
let duplicate_rate = duplicate_sfens as f64 / n as f64 * 100.0;
let top_source_pct = sources.values().max().copied().unwrap_or(0) as f64 / n as f64 * 100.0;
let opening_pct = phases.get("opening").copied().unwrap_or(0) as f64 / n as f64 * 100.0;
let black_count = sides.get("black").copied().unwrap_or(0);
let white_count = sides.get("white").copied().unwrap_or(0);
println!("=== shogiesa report ===");
println!("positions : {n}");
println!("broken lines : {broken}");
println!(
"ply range : {ply_min}–{ply_max} (avg {:.1})",
ply_sum as f64 / n as f64
);
println!("invalid SFENs : {invalid_sfens}");
println!("duplicate SFENs: {duplicate_sfens}");
println!("tag mismatches : {tag_mismatches} (side_to_move vs SFEN)");
println!();
println!("schema versions: {schema_versions:?}");
println!();
println!("phase distribution:");
for (phase, count) in &phases {
println!(
" {phase:<12} {count:>6} ({:.1}%)",
*count as f64 / n as f64 * 100.0
);
}
println!();
println!("side to move:");
for (side, count) in &sides {
println!(
" {side:<12} {count:>6} ({:.1}%)",
*count as f64 / n as f64 * 100.0
);
}
println!();
println!("wdl distribution (mover-relative):");
for (label, count) in &wdl_counts {
println!(
" {label:<12} {count:>6} ({:.1}%)",
*count as f64 / n as f64 * 100.0
);
}
println!();
println!("game_result source distribution:");
for (label, count) in &result_source_counts {
println!(
" {label:<24} {count:>6} ({:.1}%)",
*count as f64 / n as f64 * 100.0
);
}
println!();
println!("tag ratios:");
println!(
" in-check {in_check:>6} ({:.1}%)",
in_check as f64 / n as f64 * 100.0
);
println!(
" capture {has_capture:>6} ({:.1}%)",
has_capture as f64 / n as f64 * 100.0
);
println!();
println!("source files: {}", sources.len());
for (path, count) in sources.iter().take(10) {
println!(" {path}: {count}");
}
if sources.len() > 10 {
println!(" … and {} more", sources.len() - 10);
}
println!();
println!("source dominance:");
let top_warn = if top_source_pct > 50.0 {
"WARN: too concentrated"
} else {
"OK"
};
println!(" top source : {top_source_pct:.1}% {top_warn}");
println!();
println!("balance warnings:");
let opening_warn = if opening_pct > 50.0 {
"WARN: too high"
} else {
"OK"
};
println!(" opening ratio : {opening_pct:.1}% {opening_warn}");
let (b_pct, w_pct) = (
black_count as f64 / n as f64 * 100.0,
white_count as f64 / n as f64 * 100.0,
);
let side_warn = if b_pct > 65.0 || w_pct > 65.0 {
"WARN"
} else {
"OK"
};
println!(" side imbalance : {b_pct:.1}% / {w_pct:.1}% {side_warn}");
let dup_warn = if duplicate_rate > 5.0 {
"WARN: too high"
} else {
"OK"
};
println!(" duplicate rate : {duplicate_rate:.1}% {dup_warn}");
let unlabeled = n - labeled;
println!();
println!("observations:");
println!(
" labeled : {labeled:>6} ({:.1}%)",
labeled as f64 / n as f64 * 100.0
);
println!(
" unlabeled : {unlabeled:>6} ({:.1}%)",
unlabeled as f64 / n as f64 * 100.0
);
if labeled > 0 {
println!(
" depth disagree : {depth_disagree:>6} ({:.1}% of labeled)",
depth_disagree as f64 / labeled as f64 * 100.0
);
if multi_engine > 0 {
println!(
" engine disagree: {engine_disagree:>6} ({:.1}% of {multi_engine} multi-engine positions)",
engine_disagree as f64 / multi_engine as f64 * 100.0
);
}
println!(
" special bestmove: {special_bestmove:>5} ({:.1}% of labeled; resign/win/none)",
special_bestmove as f64 / labeled as f64 * 100.0
);
println!(" depth counts:");
for (&depth, &count) in &depth_counts {
println!(" depth {depth:>2} : {count:>6}");
}
let total_obs = cp_obs_count + mate_obs_count;
println!(
" cp/mate ratio : {cp_obs_count} cp / {mate_obs_count} mate ({:.1}% mate)",
mate_obs_count as f64 / total_obs.max(1) as f64 * 100.0
);
println!(" score bound (observations):");
for (bound, count) in &obs_score_bound_counts {
println!(" {bound:<10} : {count:>6}");
}
if swing_count > 0 {
println!(
" avg score swing: {:.1}cp (over {swing_count} records with \u{2265}2 cp observations)",
swing_sum / swing_count as f64
);
}
if margin_count > 0 {
println!(
" avg policy margin: {:.1}cp (over {margin_count} observations)",
margin_sum / margin_count as f64
);
}
if obs_with_candidates > 0 {
println!(
" multipv coverage: {obs_with_candidates:>6} ({:.1}% of {obs_total} observations)",
obs_with_candidates as f64 / obs_total as f64 * 100.0
);
println!(" score bound (multipv candidates):");
for (bound, count) in &score_bound_distribution {
println!(" {bound:<10} : {count:>6}");
}
}
if requested_depth_total > 0 {
println!(
" requested-depth underreach: {requested_depth_underreach:>6} ({:.1}% of {requested_depth_total} observations with a requested_depth)",
requested_depth_underreach as f64 / requested_depth_total as f64 * 100.0
);
}
}
if !eval_buckets.is_empty() {
let bar_max = eval_buckets.values().copied().max().unwrap_or(1);
println!();
println!("eval distribution (200cp buckets, deepest observation):");
for (&key, &count) in &eval_buckets {
let label = if key == i32::MIN {
" unlabeled ".to_string()
} else if key == i32::MAX {
" mate ".to_string()
} else {
format!(" {:+5}..{:+5}", key, key + 199)
};
let bar = "█".repeat(count * 20 / bar_max.max(1));
println!("{label}: {count:>5} {bar}");
}
}
if !swing_buckets.is_empty() {
let bar_max = swing_buckets.values().copied().max().unwrap_or(1);
println!();
println!("score swing distribution (50cp buckets, per record):");
for (&key, &count) in &swing_buckets {
let bar = "█".repeat(count * 20 / bar_max.max(1));
println!(" {key:>4}..{:<4}: {count:>5} {bar}", key + 49);
}
}
print_eval_cross_tab("eval bucket x phase", &eval_by_phase);
print_eval_cross_tab("eval bucket x side", &eval_by_side);
Ok(())
}
fn print_eval_cross_tab(title: &str, table: &BTreeMap<i32, BTreeMap<String, usize>>) {
if table.is_empty() {
return;
}
let columns: BTreeSet<String> = table.values().flat_map(|m| m.keys().cloned()).collect();
println!();
println!("{title}:");
print!(" {:<14}", "");
for col in &columns {
print!("{col:>12}");
}
println!();
for (&key, row) in table {
let label = if key == i32::MIN {
"unlabeled".to_string()
} else if key == i32::MAX {
"mate".to_string()
} else {
format!("{:+}..{:+}", key, key + 199)
};
print!(" {label:<14}");
for col in &columns {
print!("{:>12}", row.get(col).copied().unwrap_or(0));
}
println!();
}
}
const MAX_EVAL_CP_SPAN_BUCKETS: i32 = 50;
fn cmd_distribution(args: DistributionArgs) -> Result<()> {
if args.ply_bucket_size == 0 {
anyhow::bail!("--ply-bucket-size must be > 0");
}
let reader = BufReader::new(
File::open(&args.input).with_context(|| format!("cannot open {:?}", args.input))?,
);
let mut n = 0usize;
let mut broken = 0usize;
let mut cross_tally: BTreeMap<(GamePhase, SideToMove, EvalBucket), usize> = BTreeMap::new();
let mut ply_tally: BTreeMap<u32, usize> = BTreeMap::new();
let mut root_tally: HashMap<String, usize> = HashMap::new();
let mut wdl_tally: BTreeMap<&'static str, usize> = BTreeMap::new();
let mut result_source_tally: BTreeMap<String, usize> = BTreeMap::new();
let mut eval_cp_min: Option<i32> = None;
let mut eval_cp_max: Option<i32> = None;
let mut ply_min: Option<u32> = None;
let mut ply_max: Option<u32> = None;
for (i, line) in reader.lines().enumerate() {
let line = line.with_context(|| format!("cannot read {:?}", args.input))?;
if line.trim().is_empty() {
continue;
}
let rec: PositionRecord = match serde_json::from_str(&line) {
Ok(r) => r,
Err(e) => {
tracing::warn!(line = i + 1, "parse error: {e}");
broken += 1;
continue;
}
};
n += 1;
let eb = eval_bucket_of(&rec);
if let EvalBucket::Cp(v) = eb {
eval_cp_min = Some(eval_cp_min.map_or(v, |m| m.min(v)));
eval_cp_max = Some(eval_cp_max.map_or(v, |m| m.max(v)));
}
*cross_tally
.entry((rec.tags.phase, rec.tags.side_to_move, eb))
.or_default() += 1;
let ply = rec.source.ply;
*ply_tally
.entry(bucket_floor(ply, args.ply_bucket_size))
.or_default() += 1;
ply_min = Some(ply_min.map_or(ply, |m| m.min(ply)));
ply_max = Some(ply_max.map_or(ply, |m| m.max(ply)));
*root_tally
.entry(shogiesa_stratify::group_key(&rec.source))
.or_default() += 1;
*wdl_tally
.entry(shogiesa_stratify::feature_wdl(&rec))
.or_default() += 1;
*result_source_tally
.entry(
rec.game_result
.as_ref()
.map(|gr| gr.result_source.clone())
.unwrap_or_else(|| "no_game_result".to_string()),
)
.or_default() += 1;
}
if n == 0 {
println!("no valid records in {:?}", args.input);
return Ok(());
}
println!("=== shogiesa distribution ===");
println!("positions : {n}");
println!("broken lines: {broken}");
print_eval_coverage(
&cross_tally,
eval_cp_min,
eval_cp_max,
args.under_ratio,
args.over_ratio,
);
print_ply_histogram(
&ply_tally,
ply_min.unwrap(),
ply_max.unwrap(),
args.ply_bucket_size,
args.under_ratio,
args.over_ratio,
);
print_source_root_distribution(&root_tally, n);
print_wdl_distribution(&wdl_tally, n);
print_result_source_distribution(&result_source_tally, n);
Ok(())
}
fn print_coverage_row(
phase: GamePhase,
side: SideToMove,
eb: EvalBucket,
count: usize,
flag: impl std::fmt::Display,
) {
let eval_label = match eb {
EvalBucket::Cp(v) => format!("{v:+}..{:+}", v + 199),
EvalBucket::Mate => "mate".to_string(),
EvalBucket::Unlabeled => "unlabeled".to_string(),
};
let label = format!("{phase}:{side}:{eval_label}");
println!(" {label:<32}{count:>6} {flag}");
}
fn print_eval_coverage(
cross_tally: &BTreeMap<(GamePhase, SideToMove, EvalBucket), usize>,
eval_cp_min: Option<i32>,
eval_cp_max: Option<i32>,
under_ratio: f64,
over_ratio: f64,
) {
println!();
println!("phase x side x eval-bucket coverage:");
let phases = [
GamePhase::Opening,
GamePhase::Middlegame,
GamePhase::Endgame,
];
let sides = [SideToMove::Black, SideToMove::White];
let cp_mean = mean_of(
&cross_tally
.iter()
.filter(|&(&(_, _, eb), _)| matches!(eb, EvalBucket::Cp(_)))
.map(|(_, &c)| c)
.collect::<Vec<_>>(),
);
let span_ok = match (eval_cp_min, eval_cp_max) {
(Some(lo), Some(hi)) => (hi - lo) / 200 < MAX_EVAL_CP_SPAN_BUCKETS,
_ => true, };
let mut cp_cells_total = 0usize;
let mut cp_cells_missing = 0usize;
let mut sentinel_cells_missing = 0usize;
for &phase in &phases {
for &side in &sides {
for eb in [EvalBucket::Unlabeled, EvalBucket::Mate] {
let count = cross_tally.get(&(phase, side, eb)).copied().unwrap_or(0);
if count == 0 {
sentinel_cells_missing += 1;
print_coverage_row(phase, side, eb, count, BucketStatus::Missing);
} else {
print_coverage_row(phase, side, eb, count, "-");
}
}
if let (Some(lo), Some(hi)) = (eval_cp_min, eval_cp_max)
&& span_ok
{
let mut v = lo;
while v <= hi {
cp_cells_total += 1;
let eb = EvalBucket::Cp(v);
let count = cross_tally.get(&(phase, side, eb)).copied().unwrap_or(0);
let status = classify_bucket(count, cp_mean, under_ratio, over_ratio);
if status == BucketStatus::Missing {
cp_cells_missing += 1;
}
print_coverage_row(phase, side, eb, count, status);
v += 200;
}
}
}
}
if !span_ok {
println!(
" cp-bucket span too wide ({}..{}) -- exceeds {MAX_EVAL_CP_SPAN_BUCKETS} enumerated \
buckets; showing observed cp buckets only, missing-bucket detection skipped for this run",
eval_cp_min.unwrap(),
eval_cp_max.unwrap()
);
for (&(phase, side, eb), &count) in cross_tally
.iter()
.filter(|&(&(_, _, eb), _)| matches!(eb, EvalBucket::Cp(_)))
{
let flag = classify_bucket(count, cp_mean, under_ratio, over_ratio);
print_coverage_row(phase, side, eb, count, flag);
}
} else {
println!(" (cp-grid: {cp_cells_total} cells, {cp_cells_missing} missing)");
}
println!(
" (sentinel cells (mate/unlabeled x phase/side): 12 cells, {sentinel_cells_missing} empty)"
);
}
fn print_ply_histogram(
ply_tally: &BTreeMap<u32, usize>,
ply_min: u32,
ply_max: u32,
bucket_size: u32,
under_ratio: f64,
over_ratio: f64,
) {
println!();
println!("ply distribution (bucket size {bucket_size}):");
let mean = mean_of(&ply_tally.values().copied().collect::<Vec<_>>());
let lo = bucket_floor(ply_min, bucket_size);
let hi = bucket_floor(ply_max, bucket_size);
let mut total = 0usize;
let mut missing = 0usize;
let mut b = lo;
while b <= hi {
total += 1;
let count = ply_tally.get(&b).copied().unwrap_or(0);
let flag = classify_bucket(count, mean, under_ratio, over_ratio);
if flag == BucketStatus::Missing {
missing += 1;
}
println!(" {b:>4}..{:<4}{count:>6} {flag}", b + bucket_size - 1);
b += bucket_size;
}
println!(" ({total} buckets enumerated, {missing} missing)");
}
fn print_source_root_distribution(root_tally: &HashMap<String, usize>, n: usize) {
println!();
println!("source-root distribution:");
let distinct_roots = root_tally.len();
let top_root_count = root_tally.values().copied().max().unwrap_or(0);
let top_root_pct = top_root_count as f64 / n as f64 * 100.0;
let warn = if top_root_pct > 50.0 {
"WARN: too concentrated"
} else {
"OK"
};
println!(" distinct roots : {distinct_roots}");
println!(" top root : {top_root_pct:.1}% {warn}");
}
fn print_wdl_distribution(wdl_tally: &BTreeMap<&'static str, usize>, n: usize) {
println!();
println!("wdl distribution (mover-relative):");
for (label, count) in wdl_tally {
let pct = *count as f64 / n as f64 * 100.0;
println!(" {label:<10}{count:>8} {pct:.1}%");
}
}
fn print_result_source_distribution(tally: &BTreeMap<String, usize>, n: usize) {
println!();
println!("game_result source distribution:");
for (label, count) in tally {
let pct = *count as f64 / n as f64 * 100.0;
println!(" {label:<24}{count:>8} {pct:.1}%");
}
}
fn cmd_validate(args: ValidateArgs) -> Result<()> {
let file = File::open(&args.input).with_context(|| format!("cannot read {:?}", args.input))?;
let reader = BufReader::new(file);
let mut total_lines = 0usize;
let mut valid_json = 0usize;
let mut valid_records = 0usize;
let mut tag_mismatches = 0usize;
let mut invalid_sfens = 0usize;
let mut schema_versions = BTreeMap::<u32, usize>::new();
let mut seen_sfens: HashSet<String> = HashSet::new();
let mut duplicate_sfens = 0usize;
for line in reader.lines() {
let line = line.with_context(|| format!("cannot read {:?}", args.input))?;
if line.trim().is_empty() {
continue;
}
total_lines += 1;
let Ok(val) = serde_json::from_str::<serde_json::Value>(&line) else {
continue;
};
valid_json += 1;
let Ok(rec) = serde_json::from_value::<PositionRecord>(val) else {
continue;
};
valid_records += 1;
*schema_versions.entry(rec.schema_version).or_default() += 1;
if !seen_sfens.insert(rec.sfen.clone()) {
duplicate_sfens += 1;
}
match Sfen::parse(&rec.sfen) {
Ok(sfen) => {
if sfen.side_to_move() != rec.tags.side_to_move {
tag_mismatches += 1;
}
}
Err(_) => invalid_sfens += 1,
}
}
let broken = total_lines - valid_json;
let has_problems = tag_mismatches > 0 || broken > 0 || invalid_sfens > 0;
println!("=== shogiesa validate ===");
println!("total lines : {total_lines}");
println!("valid JSON : {valid_json}");
println!("valid records : {valid_records}");
println!("broken lines : {broken}");
println!("invalid SFENs : {invalid_sfens}");
println!("duplicate SFENs: {duplicate_sfens}");
println!("tag mismatches : {tag_mismatches} (side_to_move vs SFEN)");
println!("schema versions: {schema_versions:?}");
if has_problems {
println!();
if broken > 0 {
println!("WARN: {broken} broken lines");
}
if invalid_sfens > 0 {
println!("WARN: {invalid_sfens} invalid SFENs");
}
if tag_mismatches > 0 {
println!("WARN: {tag_mismatches} side_to_move tag mismatches");
}
if args.strict {
std::process::exit(1);
}
} else {
println!();
println!("OK");
}
Ok(())
}
#[cfg(test)]
mod label_pipeline_tests {
use super::*;
use shogiesa_core::{PositionTags, SourceInfo};
fn rec(tag: &str) -> PositionRecord {
PositionRecord::new(
"startpos".to_string(),
SourceInfo {
kind: "test".to_string(),
path: tag.to_string(),
ply: 1,
root_id: None,
variation_id: None,
branch_from_ply: None,
},
PositionTags {
phase: GamePhase::Opening,
side_to_move: SideToMove::Black,
in_check: false,
has_capture: false,
},
)
}
fn tag(record: &PositionRecord) -> &str {
&record.source.path
}
#[test]
fn reorder_push_holds_back_out_of_order_arrivals() {
let mut pending = BTreeMap::new();
let mut next_id = 0u64;
let ready = reorder_push(&mut pending, &mut next_id, 1, rec("b"));
assert!(ready.is_empty());
assert_eq!(pending.len(), 1);
let ready = reorder_push(&mut pending, &mut next_id, 0, rec("a"));
assert_eq!(ready.iter().map(tag).collect::<Vec<_>>(), vec!["a", "b"]);
assert!(pending.is_empty());
assert_eq!(next_id, 2);
}
#[test]
fn reorder_push_restores_order_from_arbitrary_arrival_order() {
let mut pending = BTreeMap::new();
let mut next_id = 0u64;
let arrivals = [(3, "d"), (1, "b"), (0, "a"), (2, "c")];
let mut output = Vec::new();
for (id, t) in arrivals {
output.extend(reorder_push(&mut pending, &mut next_id, id, rec(t)));
}
assert_eq!(
output.iter().map(tag).collect::<Vec<_>>(),
vec!["a", "b", "c", "d"]
);
}
#[test]
fn reorder_push_buffer_size_equals_arrivals_ahead_of_next_id() {
let mut pending = BTreeMap::new();
let mut next_id = 0u64;
let mut max_buffered = 0usize;
for id in (1..=20u64).rev() {
let ready = reorder_push(&mut pending, &mut next_id, id, rec("x"));
assert!(ready.is_empty(), "job 0 hasn't arrived yet");
max_buffered = max_buffered.max(pending.len());
}
assert_eq!(max_buffered, 20);
let ready = reorder_push(&mut pending, &mut next_id, 0, rec("a"));
assert_eq!(ready.len(), 21, "job 0 unblocks every buffered successor");
assert!(pending.is_empty());
}
}
#[cfg(test)]
mod resume_index_tests {
use super::*;
fn labeled_line(sfen: &str, path: &str, ply: u32, bestmove: &str) -> String {
let rec = serde_json::json!({
"schema_version": SCHEMA_VERSION,
"sfen": sfen,
"source": {"kind": "test", "path": path, "ply": ply},
"tags": {"phase": "opening", "side_to_move": "black", "in_check": false, "has_capture": false},
"observations": [{
"engine": "test", "depth": 4, "score": {"kind": "cp", "value": 10},
"bestmove": bestmove,
}],
});
rec.to_string()
}
#[test]
fn index_maps_keys_to_the_offset_of_their_line_and_a_later_duplicate_wins() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("resume.jsonl");
let lines = [
labeled_line("sfen-a", "g.csa", 1, "7g7f"),
labeled_line("sfen-b", "g.csa", 2, "3c3d"),
labeled_line("sfen-a", "g.csa", 1, "2g2f"),
];
fs::write(&path, lines.join("\n") + "\n").unwrap();
let index = build_resume_index(&path).unwrap();
assert_eq!(index.len(), 2, "two distinct keys, despite three lines");
let mut file = File::open(&path).unwrap();
let offset_a = index[&("sfen-a".to_string(), "g.csa".to_string(), 1)];
let obs_a = read_resume_observations_at(&mut file, offset_a).unwrap();
assert_eq!(
obs_a[0].bestmove, "2g2f",
"must resolve to the later (last-wins) occurrence of a duplicated key"
);
let offset_b = index[&("sfen-b".to_string(), "g.csa".to_string(), 2)];
let obs_b = read_resume_observations_at(&mut file, offset_b).unwrap();
assert_eq!(obs_b[0].bestmove, "3c3d");
}
}
#[cfg(test)]
mod extract_zobrist_dedup_tests {
use super::*;
use shogiesa_core::{PositionTags, SourceInfo};
fn rec_with_sfen(sfen: &str) -> PositionRecord {
PositionRecord::new(
sfen.to_string(),
SourceInfo {
kind: "test".to_string(),
path: "game".to_string(),
ply: 1,
root_id: None,
variation_id: None,
branch_from_ply: None,
},
PositionTags {
phase: GamePhase::Opening,
side_to_move: SideToMove::Black,
in_check: false,
has_capture: false,
},
)
}
#[test]
fn two_distinct_unparseable_sfens_are_both_kept_and_counted_as_skipped() {
let mut seen_hashes = HashSet::new();
let mut skipped = 0usize;
let a = rec_with_sfen("not a valid sfen");
let b = rec_with_sfen("also not valid");
assert!(zobrist_from_sfen(&a.sfen).is_none());
assert!(zobrist_from_sfen(&b.sfen).is_none());
assert!(!zobrist_dedup_keep(&a, &mut seen_hashes, &mut skipped));
assert!(!zobrist_dedup_keep(&b, &mut seen_hashes, &mut skipped));
assert_eq!(skipped, 2, "both unparseable records counted as skipped");
assert!(seen_hashes.is_empty(), "no sentinel hash was ever inserted");
}
#[test]
fn valid_duplicate_sfen_is_deduped_normally() {
let mut seen_hashes = HashSet::new();
let mut skipped = 0usize;
let sfen = "lnsgkgsnl/1r5b1/ppppppppp/9/9/9/PPPPPPPPP/1B5R1/LNSGKGSNL b - 1";
assert!(zobrist_dedup_keep(
&rec_with_sfen(sfen),
&mut seen_hashes,
&mut skipped
));
assert!(!zobrist_dedup_keep(
&rec_with_sfen(sfen),
&mut seen_hashes,
&mut skipped
));
assert_eq!(skipped, 0, "a real duplicate isn't counted as skipped");
}
}
#[cfg(test)]
mod fingerprint_tests {
use super::*;
#[test]
fn assign_split_bucket_is_a_stable_golden_value() {
assert_eq!(
assign_split_bucket(42, "some-fixed-key", 0.1, 0.1),
SplitBucket::Train
);
}
#[test]
fn engine_options_hash_is_a_stable_golden_value() {
assert_eq!(
engine_options_hash(&[("MultiPV".to_string(), "4".to_string())]),
1923589341701319780
);
}
#[test]
fn label_cache_key_is_a_64_char_hex_digest() {
let cache = LabelCache {
dir: PathBuf::from("/tmp/cache"),
engine_options_hash: engine_options_hash(&[]),
multipv: 1,
engine_fingerprint: None,
engine_fingerprint_mode: EngineFingerprintMode::None,
hits: Arc::new(AtomicUsize::new(0)),
misses: Arc::new(AtomicUsize::new(0)),
};
let path = label_cache_path(
&cache,
"startpos",
"engine",
Some("1.0"),
SearchLimit::Depth(8),
);
let key = path.file_stem().unwrap().to_str().unwrap();
assert_eq!(key.len(), 64);
assert!(key.chars().all(|c| c.is_ascii_hexdigit()));
}
}
#[cfg(test)]
mod label_cache_correctness_tests {
use super::*;
fn cache_with_fingerprint(fp: Option<u64>) -> LabelCache {
LabelCache {
dir: PathBuf::from("/tmp/cache"),
engine_options_hash: engine_options_hash(&[]),
multipv: 1,
engine_fingerprint: fp,
engine_fingerprint_mode: if fp.is_some() {
EngineFingerprintMode::Content
} else {
EngineFingerprintMode::None
},
hits: Arc::new(AtomicUsize::new(0)),
misses: Arc::new(AtomicUsize::new(0)),
}
}
#[test]
fn atomic_cache_write_never_leaves_a_torn_file_under_concurrent_writers() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("shared_key.json");
let handles: Vec<_> = (0..8)
.map(|i| {
let path = path.clone();
std::thread::spawn(move || {
write_cache_entry_atomically(&path, &format!(r#"{{"n":{i}}}"#))
})
})
.collect();
for h in handles {
h.join().unwrap().unwrap();
}
let content = fs::read_to_string(&path).unwrap();
let parsed: serde_json::Value = serde_json::from_str(&content)
.expect("file must always be complete, valid JSON, never a torn partial write");
assert!(parsed["n"].is_number());
let leftover_temp_files = fs::read_dir(dir.path())
.unwrap()
.filter_map(|e| e.ok())
.filter(|e| e.file_name().to_string_lossy().contains(".tmp."))
.count();
assert_eq!(leftover_temp_files, 0, "no temp files should remain behind");
}
#[test]
fn engine_fingerprint_content_mode_differs_for_different_binaries() {
let dir = tempfile::tempdir().unwrap();
let path_a = dir.path().join("engine_a");
let path_b = dir.path().join("engine_b");
fs::write(&path_a, b"binary A content").unwrap();
fs::write(&path_b, b"binary B content").unwrap();
let fp_a = compute_engine_fingerprint(EngineFingerprintMode::Content, &path_a);
let fp_b = compute_engine_fingerprint(EngineFingerprintMode::Content, &path_b);
assert!(fp_a.is_some());
assert_ne!(
fp_a, fp_b,
"two different binaries must not silently share a cache identity"
);
}
#[test]
fn engine_fingerprint_none_mode_produces_no_fingerprint() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("engine");
fs::write(&path, b"anything").unwrap();
assert_eq!(
compute_engine_fingerprint(EngineFingerprintMode::None, &path),
None
);
}
#[test]
fn engine_fingerprint_falls_back_to_none_when_path_is_unreadable() {
let unreadable = PathBuf::from("this-path-almost-certainly-does-not-exist-anywhere");
assert_eq!(
compute_engine_fingerprint(EngineFingerprintMode::Content, &unreadable),
None
);
assert_eq!(
compute_engine_fingerprint(EngineFingerprintMode::Metadata, &unreadable),
None
);
}
#[test]
fn label_cache_path_differs_when_engine_fingerprint_differs() {
let path_a = label_cache_path(
&cache_with_fingerprint(Some(111)),
"startpos",
"engine",
Some("1.0"),
SearchLimit::Depth(8),
);
let path_b = label_cache_path(
&cache_with_fingerprint(Some(222)),
"startpos",
"engine",
Some("1.0"),
SearchLimit::Depth(8),
);
assert_ne!(
path_a, path_b,
"a rebuilt engine binary must not collide with a stale cache entry"
);
}
}
#[cfg(test)]
mod calibrate_helper_tests {
use super::*;
use shogiesa_core::QualityReason;
fn decision(reasons: Vec<QualityReason>) -> QualityDecision {
QualityDecision {
keep: reasons.is_empty(),
score: 1.0,
reasons,
score_swing_cp: None,
bestmove_agreement: true,
cp_count: 0,
mate_count: 0,
}
}
#[test]
fn coverage_tally_records_kept_and_first_reason_only() {
let mut tally = CoverageTally::default();
tally.record(&decision(vec![]));
tally.record(&decision(vec![
QualityReason::PolicyMargin,
QualityReason::ScoreSwing,
]));
assert_eq!(tally.total, 2);
assert_eq!(tally.kept, 1);
assert_eq!(tally.dropped, 1);
assert_eq!(tally.drop_reasons_csv_cell(), "policy_margin=1");
assert_eq!(tally.coverage_pct(), 50.0);
}
#[test]
fn dataset_diagnostics_skips_unlabeled_records() {
let rec: PositionRecord = serde_json::from_str(
r#"{"schema_version":8,"sfen":"x","source":{"kind":"csa","path":"g.csa","ply":1},
"tags":{"phase":"opening","side_to_move":"black","in_check":false,"has_capture":false},
"observations":[]}"#,
)
.unwrap();
let mut diagnostics = DatasetDiagnostics::default();
diagnostics.record(&rec);
assert_eq!(
diagnostics.labeled, 0,
"empty observations must not count as labeled"
);
}
#[test]
fn dataset_diagnostics_buckets_a_labeled_record() {
let rec: PositionRecord = serde_json::from_str(
r#"{"schema_version":8,"sfen":"x","source":{"kind":"csa","path":"g.csa","ply":1},
"tags":{"phase":"opening","side_to_move":"black","in_check":false,"has_capture":false},
"observations":[{"engine":"e","engine_version":null,"depth":4,"score":{"kind":"cp","value":50},
"bestmove":"7g7f","nodes":null,"time_ms":null,"pv":null,"policy_margin_cp":80}]}"#,
)
.unwrap();
let mut diagnostics = DatasetDiagnostics::default();
diagnostics.record(&rec);
assert_eq!(diagnostics.labeled, 1);
assert_eq!(diagnostics.margin_buckets.get(&50), Some(&1)); assert_eq!(diagnostics.obs_score_bound_counts.get("exact"), Some(&1));
}
}
#[cfg(test)]
mod tune_pareto_tests {
use super::*;
fn cell_with(id: i32, kept_pct: usize, mismatch_pct: usize) -> TuneCell {
let mut cell = TuneCell::new(Some(id), None);
cell.coverage.total = 100;
cell.coverage.kept = kept_pct;
cell.audit.pairs = 100;
cell.audit.bestmove_mismatches = mismatch_pct;
cell
}
#[test]
fn pick_balanced_range_normalizes_before_computing_distance() {
let cells = vec![
cell_with(0, 95, 8),
cell_with(1, 60, 4),
cell_with(2, 20, 2),
];
let frontier = pareto_frontier_indices(&cells);
assert_eq!(
frontier,
vec![0, 1, 2],
"all three are mutually non-dominated: coverage and mismatch both decrease together"
);
assert_eq!(
pick_balanced(&cells, &frontier),
1,
"the middle trade-off point, not the highest-coverage one"
);
}
#[test]
fn pareto_frontier_excludes_a_strictly_dominated_point() {
let cells = vec![
cell_with(0, 70, 0), cell_with(1, 30, 0),
];
assert_eq!(pareto_frontier_indices(&cells), vec![0]);
}
#[test]
fn pick_broad_and_strict_pick_opposite_ends() {
let cells = vec![cell_with(0, 90, 10), cell_with(1, 40, 2)];
let frontier = pareto_frontier_indices(&cells);
assert_eq!(pick_broad(&cells, &frontier), 0);
assert_eq!(pick_strict(&cells, &frontier), 1);
}
}
#[cfg(test)]
mod merge_observations_tests {
use super::*;
fn obs(engine: &str, depth: u32, requested_depth: Option<u32>, bestmove: &str) -> Observation {
Observation {
engine: engine.to_string(),
engine_version: None,
depth,
requested_depth,
requested_nodes: None,
search_limit_kind: SearchLimitKind::Depth,
score: Score::Cp { value: 0 },
score_perspective: ScorePerspective::default(),
score_bound: shogiesa_core::ScoreBound::default(),
bestmove: bestmove.to_string(),
bestmove_kind: None,
nodes: None,
time_ms: None,
seldepth: None,
nps: None,
hashfull: None,
pv: None,
policy_margin_cp: None,
candidates: Vec::new(),
engine_options_hash: None,
weight_sha256: None,
was_timeout_salvaged: false,
}
}
fn nodes_obs(engine: &str, depth: u32, requested_nodes: u64, bestmove: &str) -> Observation {
Observation {
requested_depth: None,
requested_nodes: Some(requested_nodes),
search_limit_kind: SearchLimitKind::Nodes,
..obs(engine, depth, None, bestmove)
}
}
#[test]
fn no_collision_appends_both() {
let mut base = vec![obs("e1", 4, Some(4), "7g7f")];
let incoming = vec![obs("e2", 4, Some(4), "3c3d")];
let collisions =
merge_observations_into(&mut base, incoming, MergeObservationPolicy::KeepBoth);
assert_eq!(collisions, 0);
assert_eq!(base.len(), 2);
}
#[test]
fn keep_both_on_collision_keeps_both() {
let mut base = vec![obs("e1", 4, Some(4), "7g7f")];
let incoming = vec![obs("e1", 4, Some(4), "3c3d")]; let collisions =
merge_observations_into(&mut base, incoming, MergeObservationPolicy::KeepBoth);
assert_eq!(collisions, 0); assert_eq!(base.len(), 2);
}
#[test]
fn prefer_primary_on_collision_drops_secondary() {
let mut base = vec![obs("e1", 4, Some(4), "7g7f")];
let incoming = vec![obs("e1", 4, Some(4), "3c3d")];
let collisions =
merge_observations_into(&mut base, incoming, MergeObservationPolicy::PreferPrimary);
assert_eq!(collisions, 1);
assert_eq!(base.len(), 1);
assert_eq!(base[0].bestmove, "7g7f"); }
#[test]
fn prefer_secondary_on_collision_replaces_primary() {
let mut base = vec![obs("e1", 4, Some(4), "7g7f")];
let incoming = vec![obs("e1", 4, Some(4), "3c3d")];
let collisions =
merge_observations_into(&mut base, incoming, MergeObservationPolicy::PreferSecondary);
assert_eq!(collisions, 1);
assert_eq!(base.len(), 1);
assert_eq!(base[0].bestmove, "3c3d"); }
#[test]
fn multiple_simultaneous_collisions_all_resolved() {
let mut base = vec![obs("e1", 4, Some(4), "7g7f"), obs("e2", 6, Some(6), "3c3d")];
let incoming = vec![
obs("e1", 4, Some(4), "aaaa"), obs("e2", 6, Some(6), "bbbb"), obs("e3", 8, Some(8), "cccc"), ];
let collisions =
merge_observations_into(&mut base, incoming, MergeObservationPolicy::PreferSecondary);
assert_eq!(collisions, 2);
assert_eq!(base.len(), 3);
assert!(base.iter().any(|o| o.bestmove == "aaaa"));
assert!(base.iter().any(|o| o.bestmove == "bbbb"));
assert!(base.iter().any(|o| o.bestmove == "cccc"));
}
#[test]
fn engine_version_mismatch_is_not_a_collision() {
let mut base = vec![obs("e1", 4, Some(4), "7g7f")];
let mut versioned = obs("e1", 4, Some(4), "3c3d");
versioned.engine_version = Some("2.0".to_string());
let collisions = merge_observations_into(
&mut base,
vec![versioned],
MergeObservationPolicy::PreferPrimary,
);
assert_eq!(collisions, 0);
assert_eq!(base.len(), 2);
}
#[test]
fn shallow_and_deep_same_engine_never_collide_under_any_policy() {
for policy in [
MergeObservationPolicy::KeepBoth,
MergeObservationPolicy::PreferPrimary,
MergeObservationPolicy::PreferSecondary,
] {
let mut base = vec![obs("e1", 4, Some(4), "7g7f")];
let deep = vec![obs("e1", 12, Some(12), "3c3d")];
let collisions = merge_observations_into(&mut base, deep, policy);
assert_eq!(collisions, 0);
assert_eq!(base.len(), 2, "both depths must survive under every policy");
}
}
#[test]
fn nodes_mode_observations_with_different_requested_nodes_never_collide() {
let mut base = vec![nodes_obs("e1", 8, 50_000, "7g7f")];
let incoming = vec![nodes_obs("e1", 8, 200_000, "3c3d")]; let collisions =
merge_observations_into(&mut base, incoming, MergeObservationPolicy::PreferPrimary);
assert_eq!(collisions, 0, "different requested_nodes must not collide");
assert_eq!(base.len(), 2, "both node-limited observations must survive");
}
}