Something went wrong. Try again.
bayes for days
Something went wrong. Try again.
Python
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479"""A corpus has to outlive the worktree it was recorded in, and say what it is.
Both halves were paid for on the same afternoon: one `git worktree remove--force` destroyed 360 matches and 21937 decisions, and a fit against a corpusrecorded with an older feature set produced a weight vector nobody could tellfrom a good one. The tests below are those two failures, made cheap."""
import contextlibimport ioimport jsonimport osimport shutilimport tempfileimport unittestfrom pathlib import Pathfrom unittest import mock
from sds.corpus import ( DATA, MANIFEST, CorpusError, Manifest, Review, Stale, basis_fingerprint, check, check_basis, check_data, corpora_root, listing, resolve, save, stamp,)from sds.epoch import EPOCHfrom sds.train import read_corpusfrom tests.synthetic import LOCAL_FEATURE, write_run
PLANTED = {"expected_damage": 1.0, "overkill": -0.5, "p_kill": 0.25}
class CorpusTest(unittest.TestCase): """A store in a temporary directory, so nothing here touches the real one."""
def setUp(self): self.tmp = tempfile.TemporaryDirectory() self.root = Path(self.tmp.name) self.store = self.root / "corpora" self.previous = os.environ.get("SDS_CORPORA") os.environ["SDS_CORPORA"] = str(self.store) self.addCleanup(self.tmp.cleanup) self.addCleanup(self._restore)
def _restore(self): if self.previous is None: os.environ.pop("SDS_CORPORA", None) else: os.environ["SDS_CORPORA"] = self.previous
def run_dir(self, name="run", **kwargs): directory = write_run(self.root / name, PLANTED, matches=3, decisions=6, **kwargs) (directory / "baseline.json").write_text( json.dumps( { "bot": "sds-bot --weights weights/fitted.json --explore 0.1", "scenario": "suite:mirror:abcdef123456", } ) ) return directory
class TestSavedCorpusOutlivesItsRun(CorpusTest): """The failure this module exists because of.
The 360-match corpus was a directory of symlinks into a worktree. Removing the worktree removed the corpus, and `runs/` being gitignored meant there was nothing left that even named it. """
def test_data_survives_the_source_directory_being_deleted(self): source = self.run_dir() manifest = save(source, "overnight") shutil.rmtree(source) self.assertFalse(source.exists())
self.assertTrue(manifest.data.is_dir()) corpus = read_corpus(manifest.data) self.assertEqual(len(corpus.matches), manifest.matches) self.assertEqual(len(corpus.decisions), manifest.decisions)
def test_the_per_match_log_and_report_are_kept(self): """The only written record of why MegaMek refused something.
These were dropped on the grounds that the `.log` was the bulk of a run directory. On a real 120-match run it is 4.0M against 7.3G of decision logs, and the victory condition, the seat map and every refusal reason are in it and nowhere else. A corpus exists so the run can be deleted. """ source = self.run_dir() (source / "mirror-lance-seed1-abcdef12.log").write_text( " victory : a side loses at 70% of its BV destroyed\n" ) (source / "mirror-lance-seed1-abcdef12.report.html").write_text("<html></html>") manifest = save(source, "overnight") shutil.rmtree(source)
kept = {path.name for path in manifest.data.iterdir()} self.assertIn("mirror-lance-seed1-abcdef12.log", kept) self.assertIn("mirror-lance-seed1-abcdef12.report.html", kept) self.assertIn( "victory", (manifest.data / "mirror-lance-seed1-abcdef12.log").read_text(), )
def test_nothing_in_the_store_is_a_symlink(self): manifest = save(self.run_dir(), "overnight") for path in manifest.directory.rglob("*"): self.assertFalse(path.is_symlink(), f"{path} is a symlink")
def test_the_store_is_outside_the_run(self): source = self.run_dir() manifest = save(source, "overnight") self.assertNotIn(source.resolve(), manifest.data.resolve().parents) self.assertEqual(manifest.directory.parent, corpora_root())
def test_the_manifest_sits_beside_the_ignored_data(self): # The layout the .gitignore rule depends on: `corpora/*/data/` ignored, # the manifest one level up and committed. manifest = save(self.run_dir(), "overnight") self.assertTrue((manifest.directory / MANIFEST).is_file()) self.assertEqual(manifest.data.name, DATA)
def test_saving_over_a_corpus_needs_asking(self): self.run_dir() save(self.root / "run", "overnight") with self.assertRaises(CorpusError) as caught: save(self.root / "run", "overnight") self.assertIn("--replace", str(caught.exception)) save(self.root / "run", "overnight", replace=True)
class TestManifestRecordsWhatProducedIt(CorpusTest): def test_it_carries_the_provenance_a_fit_needs(self): manifest = save(self.run_dir(), "overnight", notes="the overnight one") self.assertEqual(manifest.epoch, EPOCH) self.assertEqual(manifest.scenario, "suite:mirror:abcdef123456") self.assertEqual(manifest.weights, "weights/fitted.json") self.assertTrue(manifest.explore) self.assertAlmostEqual(manifest.explore_rate, 0.1) self.assertEqual(manifest.matches, 3) self.assertEqual(manifest.notes, "the overnight one") self.assertEqual(manifest.source, "run")
def test_the_feature_list_is_sorted_and_complete(self): manifest = save(self.run_dir(), "overnight") expected = sorted([*PLANTED, LOCAL_FEATURE]) self.assertEqual(manifest.features, expected) self.assertEqual(manifest.learnable, sorted(PLANTED)) self.assertEqual(manifest.local, [LOCAL_FEATURE])
def test_it_round_trips_through_the_file(self): saved = save(self.run_dir(), "overnight") loaded = Manifest.load("overnight") self.assertEqual(loaded, saved)
def test_a_manifest_from_a_newer_sds_is_named_rather_than_crashed_on(self): manifest = save(self.run_dir(), "overnight") path = manifest.directory / MANIFEST document = json.loads(path.read_text()) document["provenance_v2"] = "something this version has never heard of" path.write_text(json.dumps(document)) with self.assertRaises(CorpusError) as caught: Manifest.load("overnight") self.assertIn("provenance_v2", str(caught.exception))
class TestStaleCorpusIsRefused(CorpusTest): """The second failure: a fit against a corpus that is not about this bot.
Same rule as `baseline.Incomparable`. The numbers are the same shape and mean different things, so a mismatch is an error rather than a footnote in a report nobody reads to the end. """
def test_a_different_epoch_names_what_changed(self): manifest = save(self.run_dir(), "overnight") manifest.epoch = EPOCH - 1 with self.assertRaises(Stale) as caught: check(manifest, reference=[]) message = str(caught.exception) self.assertIn(f"epoch {EPOCH - 1}", message) self.assertIn(f"epoch {EPOCH}", message) # The epoch's own note, not just its number: which change made the two # runs different games. from sds.epoch import describe
self.assertIn(describe(EPOCH), message)
def test_the_same_epoch_is_fine(self): manifest = save(self.run_dir(), "overnight") manifest.explore = False self.assertEqual(check(manifest, reference=[]), Review())
def test_requested_features_the_corpus_lacks_are_named(self): manifest = save(self.run_dir(), "overnight") asked = ["p_kill", "expected_damage", "level_tmm", "cover_quality", "los_out"] with self.assertRaises(Stale) as caught: check(manifest, requested=asked, reference=[]) message = str(caught.exception) self.assertIn("3 of the 5", message) for absent in ("level_tmm", "cover_quality", "los_out"): self.assertIn(absent, message) # And not the two it does have. self.assertNotIn("p_kill,", message)
def test_a_feature_the_baseline_m_uses_and_the_corpus_lacks_is_warned_about(self): # The expensive one: 21 features recorded, 26 in the checkout, and a fit # that silently produces a vector about the 21. A warning rather than a # refusal only because every corpus recorded today is missing the # positional features, so refusing would fire on every fit and teach # everybody to pass --stale-ok without reading it. manifest = save(self.run_dir(), "overnight") reference = sorted([*PLANTED, "exposure", "concentration"]) review = check(manifest, reference=reference) self.assertEqual(len(review.warnings), 1) message = review.warnings[0] self.assertIn("2 of them are absent", message) self.assertIn("exposure", message) self.assertIn("concentration", message) self.assertIn(str(len(manifest.features)), message)
def test_a_feature_the_corpus_has_and_the_baseline_does_not_is_a_note(self): manifest = save(self.run_dir(), "overnight") review = check(manifest, reference=["p_kill"]) self.assertTrue(any("expected_damage" in note for note in review.notes))
def test_exploration_is_said_out_loud(self): manifest = save(self.run_dir(), "overnight") review = check(manifest, reference=[]) self.assertTrue(any("off-policy" in note for note in review.notes))
def test_data_the_manifest_does_not_describe_is_refused(self): # The manifest is what gets committed and read years later, so it has to # be the truth about the directory beside it. manifest = save(self.run_dir(), "overnight") check_data(manifest, manifest.features) with self.assertRaises(Stale) as caught: check_data(manifest, [*manifest.features, "smokestack_index"]) self.assertIn("smokestack_index", str(caught.exception))
class TestResolvingATarget(CorpusTest): def test_a_name_resolves_to_the_data_and_its_manifest(self): saved = save(self.run_dir(), "overnight") target, manifest = resolve("overnight") self.assertEqual(target, saved.data) self.assertEqual(manifest, saved)
def test_a_corpus_directory_resolves_the_same_way(self): saved = save(self.run_dir(), "overnight") target, manifest = resolve(saved.directory) self.assertEqual(target, saved.data) self.assertIsNotNone(manifest)
def test_a_plain_run_directory_has_no_manifest(self): source = self.run_dir() target, manifest = resolve(source) self.assertEqual(target, source) self.assertIsNone(manifest)
class TestListing(CorpusTest): def test_it_lists_by_name(self): self.run_dir() save(self.root / "run", "beta") save(self.root / "run", "alpha") self.assertEqual([m.name for m in listing()], ["alpha", "beta"])
def test_an_empty_store_lists_nothing(self): self.assertEqual(listing(), [])
if __name__ == "__main__": unittest.main()
class TestBasisFingerprint(unittest.TestCase): """A name list cannot see a feature that keeps its name and changes meaning.
`p_kill` asked `P(total damage >= centre torso health)` and was overconfident by between 3.5x and 1144x depending on the volley's shape. Its replacement asks `P(a lethal location is destroyed)`. Same name, same normalisation, a different quantity, and every weight fitted against the old one is void. """
def entry(self, name="p_kill", norm="bounded", description="The chance of a kill."): return {"name": name, "normalisation": norm, "description": description}
def test_the_same_basis_hashes_the_same(self): a = [self.entry(), self.entry("overkill")] self.assertEqual(basis_fingerprint(a), basis_fingerprint(list(reversed(a))))
def test_a_changed_description_changes_the_basis(self): before = basis_fingerprint([self.entry(description="P(total >= CT health).")]) after = basis_fingerprint([self.entry(description="P(a lethal location dies).")]) self.assertNotEqual(before, after)
def test_a_changed_normalisation_changes_the_basis(self): self.assertNotEqual( basis_fingerprint([self.entry(norm="bounded")]), basis_fingerprint([self.entry(norm="rank")]), )
def test_an_added_feature_changes_the_basis(self): self.assertNotEqual( basis_fingerprint([self.entry()]), basis_fingerprint([self.entry(), self.entry("p_breach")]), )
def test_a_mismatched_basis_is_refused_and_says_why(self): manifest = Manifest(name="old", basis="aaaaaaaaaaaa") with self.assertRaises(Stale) as caught: check_basis(manifest, "bbbbbbbbbbbb") message = str(caught.exception) self.assertIn("aaaaaaaaaaaa", message) self.assertIn("bbbbbbbbbbbb", message)
def test_a_corpus_with_no_basis_is_not_refused(self): # Recorded before the fingerprint existed. Not worth refusing over. check_basis(Manifest(name="old", basis=""), "bbbbbbbbbbbb")
# What the bot would answer. The stamp reads its basis out of the built# binary, and whether a worktree has one built is not what these tests are# about - a fresh worktree has no `target/release/sds-bot` and the whole class# would fail there for a reason that has nothing to do with stamping.CATALOGUE = [ {"name": "expected_damage", "normalisation": "fraction", "description": "One."}, {"name": "cover_quality", "normalisation": "fraction", "description": "Two."}, {"name": "enemy_nearest", "normalisation": "local", "description": "Three."},]
class AStampedRunIsCheckable(unittest.TestCase): """A run directory says what recorded it, so `fit` cannot skip the check.
The hole this closes: `fit` runs `check_basis` only when a manifest is found, and a bench run never wrote one. It printed a note instead, and a note is a thing you read past at four in the morning. A corpus recorded against a retired basis was fitted that way. """
def setUp(self): self.root = Path(tempfile.mkdtemp()) self.run = self.root / "20260824T000000Z-bench" self.run.mkdir(parents=True) patch = mock.patch("sds.catalogue.from_bot", return_value={"features": CATALOGUE}) patch.start() self.addCleanup(patch.stop)
def tearDown(self): shutil.rmtree(self.root, ignore_errors=True)
def test_the_stamp_makes_the_directory_it_writes_into(self): """The run directory does not exist yet when the stamp is written.
`_run_dir` names it and the first match creates it, so a stamp that assumed the directory was there wrote nothing and printed a note - the whole fix, silently absent, found by running it rather than by this suite, which had been handing `stamp` a directory that already existed. """ fresh = self.root / "20260825T000000Z-bench" self.assertFalse(fresh.exists()) stamp(fresh) self.assertTrue((fresh / "manifest.json").is_file())
def test_a_stamped_run_resolves_to_a_manifest(self): stamp(self.run) _, manifest = resolve(self.run) self.assertIsNotNone(manifest, "a stamped run must bring its manifest to `fit`")
def test_the_stamp_carries_what_the_guards_read(self): manifest = stamp(self.run, bot="sds-bot --explore 0.3", explore_rate=0.3) # The two the guards actually use, plus the provenance a reader wants. self.assertTrue(manifest.basis, "no basis means check_basis cannot fire") self.assertTrue(manifest.features, "no features means check_data cannot fire") self.assertEqual(manifest.epoch, EPOCH) self.assertTrue(manifest.explore) self.assertEqual(manifest.explore_rate, 0.3)
def test_a_stamped_run_refuses_a_fit_across_a_basis_change(self): stamp(self.run) _, manifest = resolve(self.run) with self.assertRaises(Stale): check_basis(manifest, "0000deadbeef")
def test_a_basis_the_stamp_could_not_read_is_said_out_loud(self): """An empty basis is the hole reopening, so it must not be quiet.
`check_basis` returns early on an empty basis, so a stamp that failed to read the binary leaves a manifest that looks checked and is not. """ with ( mock.patch("sds.catalogue.from_bot", side_effect=RuntimeError("not built")), contextlib.redirect_stderr(io.StringIO()) as caught, ): manifest = stamp(self.run) self.assertEqual(manifest.basis, "") self.assertIn("will not be checked", caught.getvalue())
def test_counts_are_zero_until_the_corpus_is_saved(self): # Written before the first match, so there is nothing to count yet. # `check` and `check_basis` read neither, and `corpus save` fills them. manifest = stamp(self.run) self.assertEqual((manifest.matches, manifest.decisions), (0, 0))
class TestRunnerImageIsPinned(unittest.TestCase): """A run records the image it executed in, and the name is not `:latest`.
The default used to be a bare `lance-blue/sds-runner`, which docker reads as `:latest`: one mutable name on a daemon several agents share. A rebuild by any of them replaced the image a running benchmark was using, between one match and the next, and nothing failed - the matches carried on in a different environment. On this machine `:latest` and a fresh build of the Dockerfile were already two different image IDs. """
def test_the_tag_comes_from_the_dockerfile(self): import hashlib
from sds.match import REPO, default_image
digest = hashlib.sha256((REPO / "bridge" / "Dockerfile").read_bytes()).hexdigest() self.assertEqual(default_image(), f"lance-blue/sds-runner:df{digest[:12]}")
def test_build_and_run_derive_the_same_name(self): # `scripts/build.sh` computes it in bash and `match.py` in python. If # they ever disagree the harness runs an image nothing built. import subprocess
from sds.match import REPO, default_image
out = subprocess.run( ["sha256sum", str(REPO / "bridge" / "Dockerfile")], capture_output=True, text=True, check=True, ).stdout self.assertEqual(default_image(), f"lance-blue/sds-runner:df{out[:12]}")
def test_an_explicit_override_still_wins(self): import os from unittest import mock
from sds.match import default_image
with mock.patch.dict(os.environ, {"SDS_IMAGE": "someone/else:1"}): self.assertEqual(default_image(), "someone/else:1")
def test_the_stamp_records_it(self): directory = Path(tempfile.mkdtemp()) / "20260101T000000Z-bench" from sds import corpus
manifest = corpus.stamp(directory, bot="/work/bot", weights="w.json", image="pinned:1") self.assertEqual(manifest.image, "pinned:1") self.assertEqual(json.loads((directory / "manifest.json").read_text())["image"], "pinned:1")
def test_a_manifest_written_before_this_still_reads(self): # Every stored corpus predates the field. directory = Path(tempfile.mkdtemp()) directory.mkdir(parents=True, exist_ok=True) (directory / "manifest.json").write_text(json.dumps({"name": "old", "commit": "abc"})) from sds import corpus
self.assertEqual(corpus.Manifest.read(directory / "manifest.json").image, "")