diff --git a/docs/licenses/ISSUE_569_SIMILARITY_AUDIT.md b/docs/licenses/ISSUE_569_SIMILARITY_AUDIT.md new file mode 100644 index 00000000..5a0b4ad7 --- /dev/null +++ b/docs/licenses/ISSUE_569_SIMILARITY_AUDIT.md @@ -0,0 +1,105 @@ +# Issue #569 — full-tree similarity audit against the Without Number books + +**What this is.** The rights matrix ([`ISSUE_569_WITHOUT_NUMBER_RIGHTS.md`](ISSUE_569_WITHOUT_NUMBER_RIGHTS.md)) +closed every item of #569 except one, in its own words: *"a complete repository-wide similarity audit +remains outstanding"*. This is that audit, run on **2026-10-25**, with its method, corpus fingerprints +and measured result. + +It changes no file outside this record, deletes nothing, and claims no licence. + +--- + +## 1. Why not a keyword search + +The rights matrix already refuses that shortcut: a name/keyword ban *"would falsely reject legitimate +attribution and miss paraphrased copying"*. Both halves are real. A search for "Without Number" hits +this very document; a search for distinctive nouns hits our own setting vocabulary. And neither finds +a lifted sentence that changed the character names. + +So the audit measures **n-gram containment**: does any file in this tree contain a run of **N +consecutive words** that also appears in a licence-blocked source? `N = 10` — long enough that a shared +run is copying rather than coincidence, short enough to catch a lifted fragment inside a longer +sentence. Normalisation is lowercase word tokens with punctuation dropped, so reformatting, +re-wrapping and case changes do not hide anything. + +## 2. Two corpora, because a match is not a verdict + +Four books, two licence statuses, and the distinction decides whether a hit is a problem: + +| corpus | documents | status | what a hit means | +| --- | --- | --- | --- | +| **blocked** | Stars Without Number Revised (free ed., 261 pp); Ashes Without Number (free ed., 261 pp) | no SRD located, no licence statement in either file | a hit here is the thing the audit exists to catch | +| **attributable** | Cities Without Number SRD; Worlds Without Number SRD (both CC0) | CC0 — reusable with attribution | a hit that *also* occurs here is **attributable**, not a violation | + +The comparison corpus is a hash set. **No third-party text is stored, printed or committed** — the +report carries our path, the line, the run length and the classification, never the matched words. + +Corpus fingerprints, so a later run can prove it used the same books: + +| document | sha256 | +| --- | --- | +| `StarsWithoutNumberRevised-FreeEdition-122917.pdf` | `22ef6826a5b350d66888eef386b3c7f8bf02f091c18183a241e52c3cf2f44a3f` | +| `AshesWithoutNumber_FreeVersion_071025.pdf` | `033766a846f3106b72a8f72876f4dc98d319102cfc97fc39e6b79ffa9c58d3a6` | +| CWN SRD (`CWN-SRD-V5` markdown, 65 468 unique 10-grams) | repository snapshot, 2026-10-25 | +| WWN SRD (`WWNSRDV1` markdown, 80 759 unique 10-grams) | repository snapshot, 2026-10-25 | + +## 3. Result + +``` +$ python3 tools/audit_without_number_expression.py \ + --source StarsWithoutNumberRevised-FreeEdition-122917.pdf \ + --source AshesWithoutNumber_FreeVersion_071025.pdf \ + --cc0-source CWN-SRD-V5 --cc0-source WWNSRDV1 +``` + +| quantity | value | +| --- | ---: | +| text files scanned (`.md .py .json .yaml .yml .gd .txt .html`, excluding `.git`/caches) | **946** | +| unique blocked-corpus 10-word runs | 373 766 | +| unique attributable-corpus 10-word runs | 138 919 | +| **first-party hits** | **0** | +| first-party blocked-only hits | **0** | +| quarantined hits | 32, all in one file | +| quarantined blocked-only hits | **0** | + +**The first-party tree is clean**: no file outside the preserved corpora shares a ten-word run with +either blocked book. + +### The one finding, stated precisely + +`docs/archive/RULEBOOK.md` — the archived pre-#120 rulebook — contains **32** ten-word runs that also +occur in the blocked corpus. **All 32 also occur in the CC0 SRD corpus**, so they are attributable SRD +wording (the family's shared skill-level table and skill descriptions), not a copy of protected +expression. The file is additionally inside the `archive/` quarantine, which the release policy already +excludes from packaging. + +Disposition: **report, do not delete**. The file is preserved history, the overlap is licence-permitted +with attribution, and the live rulebook (`RULEBOOK.md`, `rulebook/`) has **zero** hits. If a future +release ever includes the archive, those 32 runs need the CWN/WWN attribution notice — which +`docs/sources/ATTRIBUTION.md` is the place for. + +## 4. Enforcement, and the honest limit + +- **The corpus is not shippable.** The books are copyrighted products on the operator's disk; embedding + their text (or a shingle fingerprint derived from it) in the repository would be the very thing this + audit prevents. **This makes a similarity gate impossible in CI**, and the audit says so instead of + pretending otherwise. +- **The tool is therefore an operator tool**, documented and tested (`tests/test_audit_without_number_expression.py`, + 9 code tests) so the run is reproducible by anyone holding the books. +- **The CI-side gate remains `tools/release_packaging_guard.py`** — manifest, resolved author licence, + quarantine classes, and the `LICENSE`/`NOTICE` absence rule. It does not and cannot detect n-gram + overlap; #569's item 4 is satisfied by the release gate plus this operator audit, and the division of + labour is stated here rather than implied. +- **Re-run trigger:** a new edition of any book (fingerprints above change), a new SRD, or any future + import of third-party prose. The audit is cheap: 22 seconds for 946 files against 373 k shingles. + +## 5. What was not done, and why + +- **No paraphrase detection.** Containment catches verbatim runs. A genuine paraphrase detector is an + LLM-assisted similarity review; it is out of scope for an offline deterministic tool, and the rights + matrix's own rule (human review before admitting text) remains the control that matters. +- **No archive remediation.** Deleting preserved history is forbidden by the repo's own policy + (*"report rather than silently delete"*); the overlap is recorded instead. +- **No licence selection.** D10 stays the author's; nothing here creates `LICENSE` or `NOTICE`. + +— Ai (愛) diff --git a/scripts/noopunk_text.py b/scripts/noopunk_text.py index 81fd148c..739f8481 100755 --- a/scripts/noopunk_text.py +++ b/scripts/noopunk_text.py @@ -9,6 +9,7 @@ from __future__ import annotations import argparse +import random from pathlib import Path import sys @@ -86,12 +87,16 @@ def _build_engine(args: argparse.Namespace) -> tuple[GameEngine, object | None, controllers[STARGATE_ANALYST_ID] = llm controllers[MARS_ARCHIVE_ID] = llm + # A seeded run is replayable (#570); an unseeded one uses the global generator, like a tabletop. + rng = random.Random(args.seed) if args.seed is not None else None + if args.load: engine = load_game( args.load, connection=connection, controllers=controllers, gm_controller=llm if args.llm_gm else None, + rng=rng, ) else: engine = GameEngine( @@ -99,6 +104,7 @@ def _build_engine(args: argparse.Namespace) -> tuple[GameEngine, object | None, connection=connection, controllers=controllers, gm_controller=llm if args.llm_gm else None, + rng=rng, ) return engine, llm, human_id @@ -114,6 +120,10 @@ def main() -> int: parser.add_argument("--db", default="", help="SQLite event-log path (default: in-memory)") parser.add_argument("--load", default="", help="Load JSON save") parser.add_argument("--save", default="noopunk-save.json", help="Default JSON save path") + parser.add_argument( + "--seed", type=int, default=None, + help="Seed the layer-A dice so a run is replayable; omit for tabletop-style randomness", + ) parser.add_argument("--ollama", action="store_true", help="Enable optional Ollama/Concordia actors") parser.add_argument("--llm-gm", action="store_true", help="Use the Ollama controller for brief GM narration") args = parser.parse_args() @@ -170,6 +180,7 @@ def main() -> int: connection=engine.connection, controllers=engine.controllers, gm_controller=engine.gm_controller, + rng=engine.rng, ) print(f"Loaded: {path}") print(engine.describe_room(human_id)) diff --git a/src/text_game/engine.py b/src/text_game/engine.py index bf559499..095682bb 100644 --- a/src/text_game/engine.py +++ b/src/text_game/engine.py @@ -63,6 +63,9 @@ class GameEngine: connection: sqlite3.Connection | None = None controllers: dict[str, Any] = field(default_factory=dict) gm_controller: Any | None = None + #: Dice source for the layer-A kernel. `None` means the global `random`, which is what a + #: tabletop session does; `random.Random(seed)` makes a run replayable (#570). + rng: Any | None = None events: list[Event] = field(default_factory=list) def __post_init__(self) -> None: @@ -363,6 +366,9 @@ def _resolve_test_verb(self, actor: ActorState, args: tuple[str, ...], action: A "skill": skill, "modifier": modifier, "roll": roll, + # A pinned `roll` still wins inside the kernel; otherwise the engine's own dice source + # decides, so a seeded run and a tabletop run use the same path. + "rng": self.rng, } if defender is not None: kwargs["defender_sheet"] = defender[1] diff --git a/src/text_game/persistence.py b/src/text_game/persistence.py index 77802a69..ec182276 100644 --- a/src/text_game/persistence.py +++ b/src/text_game/persistence.py @@ -33,6 +33,7 @@ def load_game( connection: sqlite3.Connection | None = None, controllers: dict | None = None, gm_controller=None, + rng=None, ) -> GameEngine: payload = json.loads(Path(path).read_text(encoding="utf-8")) events = [Event.from_dict(item) for item in payload.get("events", [])] @@ -43,6 +44,7 @@ def load_game( connection=connection, controllers=dict(controllers or {}), gm_controller=gm_controller, + rng=rng, events=events, ) if connection is not None and events: diff --git a/strategy_game/tests/test_deterministic_playthroughs.py b/strategy_game/tests/test_deterministic_playthroughs.py index 4d312ce7..fc3ae575 100644 --- a/strategy_game/tests/test_deterministic_playthroughs.py +++ b/strategy_game/tests/test_deterministic_playthroughs.py @@ -146,6 +146,27 @@ def __init__( save_name="unused.json", saves=False, ), + # --- layer A, the legacy tabletop runtime (#570 lists it explicitly) ------------------- + # `scripts/noopunk_text.py --seed` is what makes these replayable: the layer-A kernel + # otherwise consumes the global generator, so two runs would differ on the roll line. + # The save files are not compared: `simulation_id` is minted per process, so the JSON is + # deliberately unique even when the play is identical (the play is what is asserted). + _Run( + name="legacy layer A: prefall (seeded)", + argv=[str(ROOT / "scripts" / "noopunk_text.py"), "--scenario", "prefall", "--seed", "42"], + stdin="look\ntest skill Perceive\nquit\n", + marker="Perceive test:", + save_name="layer-a-prefall.json", + saves=False, + ), + _Run( + name="legacy layer A: issue74 (seeded)", + argv=[str(ROOT / "scripts" / "noopunk_text.py"), "--scenario", "issue74", "--seed", "7"], + stdin="look\ntest skill Perceive\nquit\n", + marker="Perceive test:", + save_name="layer-a-issue74.json", + saves=False, + ), ) diff --git a/strategy_game/tests/test_issue572_extension_contracts.py b/strategy_game/tests/test_issue572_extension_contracts.py new file mode 100644 index 00000000..4b0e7c13 --- /dev/null +++ b/strategy_game/tests/test_issue572_extension_contracts.py @@ -0,0 +1,199 @@ +"""Executable contracts for the #572 extension map — the six worked examples, chassis-neutral. + +`docs/design/ISSUE_572_NOOPUNK_EXTENSIONS.md` ends with a gate: publish specific acceptance tests for +"a physical intrusion, an air-gapped target, a natural PSI attempt, a psychotronic device, a +BCI-mediated message, and an astral transition". Five of those six do not depend on a die at all — +they are about **reach, permission and locus**, which is exactly what the document says must be decided +*before* any resolution. So they can be pinned now; the numeric half stays gated on #566. + +Code-only, per `AGENTS.md` and #271. Every assertion is on existing executable behaviour +(`strategy_game/cyber.py`, `strategy_game/rpg/psi.py`, `src/rules/cross_domain_state.py`, +`src/rules/equipment_catalog_index.py`) — this file adds no rule and no number. + +The one place the document promises more than the code delivers is marked in `TestPsychotronicDevice`: +the map says a device-mediated session "may be used by someone who lacks natural PSI", while `psi.py` +gates on the natural capability only. The test pins the *current* state so the gap is visible rather +than assumed. + +Run: + python -m unittest discover -s strategy_game/tests -p "test_issue572_extension_contracts.py" +""" + +from __future__ import annotations + +import sys +import unittest +from pathlib import Path +from random import Random + +ROOT = Path(__file__).resolve().parents[2] +if str(ROOT) not in sys.path: + sys.path.insert(0, str(ROOT)) + +from src.rules import cross_domain_state as cds # noqa: E402 +from src.rules import equipment_catalog_index as catalog # noqa: E402 +from strategy_game import cyber # noqa: E402 +from strategy_game.characters import Tag # noqa: E402 +from strategy_game.rpg.fate import Character, Outcome # noqa: E402 +from strategy_game.rpg.psi import PsiMode, resolve_psi_action # noqa: E402 +from strategy_game.rpg.skills import SkillLayerError # noqa: E402 + + +def _actor(tags: tuple[Tag, ...] = ()) -> Character: + return Character( + id="actor", + name="An actor", + aspects=[], + skills={"INT": 3, "SAV": 2, "WIL": 3, "COG": 2}, + tags=tags, + ) + + +class AccessGateTests(unittest.TestCase): + """Contract 1 — physical intrusion, and contract 2 — the air-gapped target.""" + + @classmethod + def setUpClass(cls) -> None: + cls.model = cyber.load_connectivity() + + def test_physical_intrusion_needs_a_body_on_site(self) -> None: + """A local path is a route only when someone or something is actually there.""" + without = self.model.resolve_access("isolated", "acceleration", "field_netrunner") + with_body = self.model.resolve_access( + "isolated", "acceleration", "field_netrunner", local_presence=True + ) + self.assertFalse(without.possible) + self.assertTrue(with_body.possible) + self.assertIn("locally delivered", with_body.reason) + + def test_an_air_gapped_target_cannot_be_entered_remotely_at_any_price(self) -> None: + """No dice, no authority and no luck cross an air gap: impossibility, not difficulty. + + The *reason* changes with the era — during the NHI Crisis there is no public network left to + argue about — but the verdict never does, and that is the contract. + """ + for era in ("acceleration", "nhi_crisis", "after_crisis"): + with self.subTest(era=era): + verdict = self.model.resolve_access( + "isolated", era, "remote_public_service", + local_presence=True, authorised=True, + ) + self.assertFalse(verdict.possible) + self.assertTrue(verdict.reason) + self.assertRegex(verdict.reason, r"isolated|field operation") + + def test_a_social_tie_is_not_a_network_route(self) -> None: + """The five refused ids are the whole of that rule, and they cannot be smuggled in.""" + for refused in ("social_tie", "personal_contact", "faction_relationship", + "influence_edge", "network_edge"): + with self.subTest(path=refused): + self.assertIn(refused, self.model.refused_path_ids) + self.assertNotIn(refused, self.model.access_paths) + + def test_a_delivered_device_creates_one_specific_path_not_a_general_one(self) -> None: + courier = self.model.resolve_access("isolated", "acceleration", "courier_compromise") + remote = self.model.resolve_access("isolated", "acceleration", "remote_public_service") + self.assertTrue(courier.possible) + self.assertFalse(remote.possible) + + +class Contract3NaturalPsiTests(unittest.TestCase): + """PSI is a capability and a mode, not a network route and not a virus.""" + + def test_the_capability_gate_fires_before_any_dice(self) -> None: + with self.assertRaises(SkillLayerError) as caught: + resolve_psi_action(_actor(), "target", PsiMode.REMOTE_PERCEPTION, 2, rng=Random(1)) + self.assertIn("REMOTE_VIEWER", str(caught.exception)) + + def test_a_capability_holder_resolves_on_the_one_engine(self) -> None: + actor = _actor((Tag("competence", "REMOTE_VIEWER"),)) + result = resolve_psi_action(actor, "target", PsiMode.REMOTE_PERCEPTION, 2, rng=Random(1)) + self.assertIsInstance(result.outcome, Outcome) + + def test_natural_psi_does_not_consult_connectivity(self) -> None: + """The same actor sees the same target whether the site is reachable or air-gapped. + + PSI is not a remote-login: `cyber` says the target is unreachable, and `psi` still resolves. + That asymmetry is the whole reason the family's hacking model cannot be imported for PSI. + """ + model = cyber.load_connectivity() + self.assertFalse(model.resolve_access("isolated", "acceleration", "remote_public_service").possible) + actor = _actor((Tag("competence", "REMOTE_VIEWER"),)) + result = resolve_psi_action(actor, "an air-gapped terminal", PsiMode.REMOTE_PERCEPTION, 2, + rng=Random(7)) + self.assertIsInstance(result.outcome, Outcome) + + +class TestPsychotronicDevice(unittest.TestCase): + """Contract 4 — a psychotronic device is a technology channel, not an innate capability.""" + + def test_the_catalog_carries_psychotronic_devices(self) -> None: + entries = catalog.search_catalog("psychotronic") + self.assertTrue(entries, "the equipment catalog must name psychotronic devices") + + def test_the_engine_still_gates_on_natural_capability_only(self) -> None: + """Recorded gap, not a bug silently papered over. + + `ISSUE_572_NOOPUNK_EXTENSIONS.md` says a technology-driven session "may be used by someone + who lacks natural PSI". `psi.py` has no device parameter, so today the engine refuses that + actor. Pinned here so the claim cannot be read as implemented. + """ + with self.assertRaises(SkillLayerError): + resolve_psi_action(_actor(), "target", PsiMode.REMOTE_PERCEPTION, 2, rng=Random(1)) + + +class Contract5BciMessageTests(unittest.TestCase): + """A BCI-mediated message travels a modelled route, and the quarantine route is a crime.""" + + @classmethod + def setUpClass(cls) -> None: + cls.model = cyber.load_connectivity() + + def test_an_unreachable_target_cannot_be_messaged(self) -> None: + self.assertFalse( + self.model.resolve_access("isolated", "nhi_crisis", "remote_public_service").possible + ) + + def test_an_authorised_link_can_carry_one(self) -> None: + self.assertTrue( + self.model.resolve_access( + "public_gateway", "acceleration", "authorized_link", authorised=True + ).possible + ) + + def test_the_quarantine_route_is_illegal_not_merely_hard(self) -> None: + verdict = self.model.resolve_access( + "isolated", "acceleration", "bridge_old_internet", local_presence=True + ) + self.assertTrue(verdict.possible) + self.assertTrue(verdict.is_containment_breach) + self.assertTrue(verdict.illegality) + + +class Contract6AstralTransitionTests(unittest.TestCase): + """One locus, one realm, and the body left behind.""" + + def test_embodied_and_projected_are_mutually_reachable(self) -> None: + self.assertTrue(cds.legal_transition(cds.EMBODIED, cds.ASTRAL_PROJECTED)) + self.assertTrue(cds.legal_transition(cds.ASTRAL_PROJECTED, cds.EMBODIED)) + + def test_two_projected_states_are_not_directly_connected(self) -> None: + """A dive cannot become a projection: the body is the only way between them.""" + self.assertFalse(cds.legal_transition(cds.CYBER_IMMERSED, cds.ASTRAL_PROJECTED)) + self.assertFalse(cds.legal_transition(cds.ASTRAL_PROJECTED, cds.CYBER_IMMERSED)) + + def test_the_body_is_not_controllable_while_projected(self) -> None: + self.assertTrue(cds.controls_body(cds.EMBODIED)) + self.assertFalse(cds.controls_body(cds.ASTRAL_PROJECTED)) + self.assertFalse(cds.controls_body(cds.CYBER_IMMERSED)) + + def test_exactly_one_realm_and_one_locus(self) -> None: + self.assertEqual(cds.realms_for(cds.ASTRAL_PROJECTED), frozenset({cds.NOOSPACE})) + self.assertFalse(cds.can_act_in(cds.ASTRAL_PROJECTED, cds.PHYSICAL)) + self.assertEqual(cds.single_locus(cds.ASTRAL_PROJECTED), cds.ASTRAL_PROJECTED) + with self.assertRaises(cds.CrossDomainError): + cds.single_locus((cds.EMBODIED, cds.ASTRAL_PROJECTED)) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_audit_without_number_expression.py b/tests/test_audit_without_number_expression.py new file mode 100644 index 00000000..fa34889c --- /dev/null +++ b/tests/test_audit_without_number_expression.py @@ -0,0 +1,135 @@ +"""Executable contracts for the issue #569 similarity audit. + +Code-only, per `AGENTS.md` and issue #271. The audit is the item the rights matrix left open +(*"a complete repository-wide similarity audit remains outstanding"*), so its arithmetic is pinned +here on **synthetic** corpora — the real books are not in the repository and must not be. + +What is asserted: + +* normalisation is deterministic and punctuation/case insensitive, so a copied sentence is found even + after reformatting; +* a shared run of `N` words is a hit and a paraphrase is not, which is the difference between a + similarity audit and the name/keyword ban the rights matrix correctly refuses; +* a hit that also exists in the CC0 corpus is classified **attributable**, not blocked-only — the + distinction that stops the audit from either excusing copying or crying wolf; +* the report never carries the matched text, only the location and the count. + +Run: + python -m unittest discover -s tests -p "test_audit_without_number_expression.py" +""" + +from __future__ import annotations + +import sys +import tempfile +import unittest +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[1] +if str(ROOT / "tools") not in sys.path: + sys.path.insert(0, str(ROOT / "tools")) +if str(ROOT) not in sys.path: + sys.path.insert(0, str(ROOT)) + +import audit_without_number_expression as audit # noqa: E402 + +SOURCE = ( + "A character has six attributes ranging from three to eighteen reflecting a range " + "from the minimum viable capacity for a playable character to the maximum normal." +) +PARAPHRASE = ( + "Player characters carry six stats spanning three to eighteen, from the weakest viable " + "build up to the strongest ordinary human." +) + + +def tree_with(files: dict[str, str]) -> Path: + root = Path(tempfile.mkdtemp(prefix="wn-audit-")) + for rel, text in files.items(): + path = root / rel + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(text, encoding="utf-8") + return root + + +class NormalisationTests(unittest.TestCase): + def test_words_strips_case_and_punctuation(self) -> None: + self.assertEqual(audit.words("Roll 2d6, add SKILL!"), ["roll", "2d6", "add", "skill"]) + + def test_shingles_are_a_set_of_the_requested_length(self) -> None: + tokens = "one two three four five six".split() + self.assertEqual(len(audit.shingles(tokens, 3)), 4) + self.assertEqual(audit.shingles(tokens[:2], 3), set()) + + +class DetectionTests(unittest.TestCase): + def test_a_copied_run_is_found_and_a_paraphrase_is_not(self) -> None: + corpus = audit.Corpus() + corpus.add("blocked-book", SOURCE) + copied = tree_with({"docs/copied.md": "Intro.\n\n" + SOURCE + "\n"}) + paraphrased = tree_with({"docs/own.md": PARAPHRASE}) + + self.assertTrue(audit.scan_file(copied / "docs/copied.md", corpus, copied)) + self.assertEqual(audit.scan_file(paraphrased / "docs/own.md", corpus, paraphrased), []) + + def test_the_finding_points_at_the_right_line(self) -> None: + corpus = audit.Corpus() + corpus.add("blocked-book", SOURCE) + root = tree_with({"a.md": "line one\nline two\n" + SOURCE + "\n"}) + findings = audit.scan_file(root / "a.md", corpus, root) + self.assertTrue(findings) + self.assertEqual(findings[0].line, 3) + self.assertEqual(findings[0].matched_words, audit.SHINGLE_WORDS) + + def test_quarantined_paths_are_classified(self) -> None: + corpus = audit.Corpus() + corpus.add("blocked-book", SOURCE) + root = tree_with({"docs/archive/old.md": SOURCE}) + report = audit.audit(corpus, None, root) + self.assertEqual(report["first_party"]["hits"], 0) + self.assertGreater(report["quarantined"]["hits"], 0) + self.assertEqual(report["quarantined"]["files"][0]["path"], "docs/archive/old.md") + self.assertTrue(report["clean"]) + + +class AttributionTests(unittest.TestCase): + def test_a_hit_that_is_also_cc0_is_attributable_not_blocked(self) -> None: + blocked = audit.Corpus() + blocked.add("blocked-book", SOURCE) + cc0 = audit.Corpus() + cc0.add("cc0-srd", SOURCE) + root = tree_with({"src/borrowed.md": SOURCE}) + + report = audit.audit(blocked, cc0, root) + self.assertEqual(report["first_party"]["hits"], report["first_party"]["attributable"]) + self.assertEqual(report["first_party"]["blocked_only"], 0) + self.assertTrue(report["clean"]) + + def test_a_blocked_only_hit_is_not_clean(self) -> None: + blocked = audit.Corpus() + blocked.add("blocked-book", SOURCE) + root = tree_with({"src/copied.md": SOURCE}) + report = audit.audit(blocked, audit.Corpus(), root) + self.assertEqual(report["first_party"]["blocked_only"], report["first_party"]["hits"]) + self.assertFalse(report["clean"]) + + +class ReportShapeTests(unittest.TestCase): + def test_the_report_never_carries_the_matched_text(self) -> None: + blocked = audit.Corpus() + blocked.add("blocked-book", SOURCE) + root = tree_with({"src/copied.md": SOURCE}) + report = audit.audit(blocked, None, root) + rendered = audit.format_report(report) + for word in ("attributes", "eighteen", "viable"): + self.assertNotIn(word, rendered.lower()) + + def test_the_audit_is_deterministic(self) -> None: + blocked = audit.Corpus() + blocked.add("blocked-book", SOURCE) + root = tree_with({"src/a.md": SOURCE, "src/b.md": PARAPHRASE}) + self.assertEqual(audit.audit(blocked, None, root), audit.audit(blocked, None, root)) + + +if __name__ == "__main__": + unittest.main() diff --git a/tools/audit_without_number_expression.py b/tools/audit_without_number_expression.py new file mode 100644 index 00000000..58510218 --- /dev/null +++ b/tools/audit_without_number_expression.py @@ -0,0 +1,288 @@ +"""Issue #569 — measure actual textual overlap between this tree and the Without Number books. + +The rights matrix (`docs/licenses/ISSUE_569_WITHOUT_NUMBER_RIGHTS.md`) leaves one item open in its +own words: *"a complete repository-wide similarity audit remains outstanding"*. It also states the +reason a keyword search cannot close it — a name ban "would falsely reject legitimate attribution and +miss paraphrased copying". This tool is the honest version: an **n-gram containment audit**. + +It answers one question: *does any file in this tree contain a run of N consecutive words that also +appears in a licence-blocked source document?* Ten words is long enough that a match is copying rather +than coincidence, and short enough to catch a lifted sentence fragment. + +Design decisions that matter: + +* **Two corpora, because a match alone is not a verdict.** The *blocked* corpus is the licence-blocked + books — Stars Without Number Revised and Ashes Without Number. The *attributable* corpus is the CC0 + SRD text — Cities Without Number and Worlds Without Number. A hit that also appears in the CC0 SRD + is **usable with attribution**; a hit that appears *only* in a blocked edition is the thing this + audit exists to catch. Reporting one number instead of two would either excuse copying or cry wolf. +* **No third-party text is ever printed, stored or committed.** The report names *our* file, the line, + and how many matched words there were. Printing the matched run would reproduce exactly the + expression the audit exists to keep out. +* **The corpus is not shippable**, so this is an operator tool and not a CI gate: the books live on the + operator's disk and must not enter the repository. `tools/release_packaging_guard.py` remains the + CI-side gate (manifest, licence resolution, quarantine policy). +* **Quarantined paths are separated, not ignored.** `legacy/` and `archive/` are preserved corpora that + the release policy already excludes; they are reported in their own section so a real violation in + first-party paths is never hidden by them. + +Usage:: + + python3 tools/audit_without_number_expression.py \ + --source /path/StarsWithoutNumberRevised-FreeEdition-122917.pdf \ + --source /path/AshesWithoutNumber_FreeVersion_071025.pdf \ + --cc0-source /path/CWN-SRD-V5 \ + --cc0-source /path/WWNSRDV1 + + python3 tools/audit_without_number_expression.py --json --source ... # machine-readable +""" + +from __future__ import annotations + +import argparse +import hashlib +import json +import re +import sys +from collections import Counter +from dataclasses import dataclass, field +from pathlib import Path +from typing import Iterable, Iterator, Mapping, Sequence + +REPO_ROOT = Path(__file__).resolve().parent.parent + +#: Consecutive-word run length. Ten is the threshold: long enough that a shared run is copying. +SHINGLE_WORDS = 10 + +#: Files whose text is compared. Deliberately no binaries, no PDFs, no generated archives. +SCANNED_SUFFIXES = (".md", ".py", ".json", ".yaml", ".yml", ".gd", ".txt", ".html") + +#: Never walked. `.git` is metadata; `__pycache__` is bytecode. +SKIP_DIRS = {".git", "__pycache__", ".pytest_cache", ".mypy_cache", "node_modules", ".ruff_cache"} + +#: Preserved corpora the release policy already excludes. Reported separately. +QUARANTINE_PREFIXES = ("legacy/", "archive/", "docs/archive/", "rulebook/source_snapshots/") + +_WORD = re.compile(r"[a-z0-9']+") + + +def words(text: str) -> list[str]: + """Normalise to lowercase word tokens with punctuation and case removed.""" + return _WORD.findall(text.lower()) + + +def _hash(phrase: str) -> int: + return int.from_bytes(hashlib.blake2b(phrase.encode("utf-8"), digest_size=8).digest(), "big") + + +def shingles(token_stream: Sequence[str], size: int = SHINGLE_WORDS) -> set[int]: + """Hashes of every `size`-word run. A set, because repetition in the source is irrelevant.""" + if len(token_stream) < size: + return set() + return {_hash(" ".join(token_stream[i:i + size])) for i in range(len(token_stream) - size + 1)} + + +@dataclass +class Corpus: + """The licence-blocked books, reduced to shingle hashes (no text retained).""" + + documents: dict[str, int] = field(default_factory=dict) # doc id -> number of shingles + hashes: set[int] = field(default_factory=set) # union + + def add(self, doc_id: str, text: str) -> None: + found = shingles(words(text)) + self.documents[doc_id] = len(found) + self.hashes |= found + + +@dataclass(frozen=True) +class Finding: + path: str + line: int + matched_words: int + quarantined: bool + also_in_cc0: bool = False + + +def corpus_from_paths(specs: Iterable[str]) -> Corpus: + """Build a corpus from PDFs or directories of `.md`/`.txt` (the SRD repos are markdown).""" + corpus = Corpus() + for spec in specs: + path = Path(spec) + if path.is_dir(): + text = "\n".join( + p.read_text(encoding="utf-8") + for p in sorted(path.rglob("*")) + if p.is_file() and p.suffix.lower() in (".md", ".txt") + ) + corpus.add(path.name, text) + elif path.suffix.lower() == ".pdf": + corpus.add(path.name, pdf_text(path)) + else: + corpus.add(path.name, path.read_text(encoding="utf-8")) + return corpus + + +def pdf_text(path: Path) -> str: + """Extract text from a PDF. Imported lazily so the tool runs without PyMuPDF on PATH. + + `pymupdf` is tried before `fitz` because the legacy `fitz` shim prints a deprecation warning to + stdout, which would corrupt `--json` output. + """ + try: + import pymupdf as fitz # type: ignore[import-not-found] + except ImportError: + try: + import fitz # type: ignore[import-not-found,no-redef] + except ImportError: # pragma: no cover - operator environment detail + raise SystemExit( + "PyMuPDF is required for PDF sources (pip install pymupdf), or pass " + "--source-text with pre-extracted .txt files instead." + ) from None + with fitz.open(str(path)) as doc: + return "\n".join(page.get_text() for page in doc) + + +def iter_text_files(root: Path = REPO_ROOT) -> Iterator[Path]: + for path in sorted(root.rglob("*")): + if not path.is_file() or path.suffix.lower() not in SCANNED_SUFFIXES: + continue + if any(part in SKIP_DIRS for part in path.parts): + continue + yield path + + +def is_quarantined(rel: str) -> bool: + return any(rel.startswith(prefix) for prefix in QUARANTINE_PREFIXES) + + +def scan_file(path: Path, corpus: Corpus, root: Path = REPO_ROOT, + size: int = SHINGLE_WORDS, attributable: Corpus | None = None) -> list[Finding]: + """Find every run of `size` words in `path` that also occurs in the corpus. + + Line numbers are recovered by tracking token positions, so a finding points at the first line of + the matched run without ever printing the matched text. + """ + try: + text = path.read_text(encoding="utf-8") + except (UnicodeDecodeError, OSError): + return [] + tokens = words(text) + if len(tokens) < size or not corpus.hashes: + return [] + + # Token index -> line number, so a hit can be located without quoting it. + line_of: list[int] = [] + line = 1 + for raw_line in text.splitlines(keepends=True): + line_of.extend([line] * len(words(raw_line))) + line += raw_line.count("\n") + while len(line_of) < len(tokens): + line_of.append(line) + + rel = path.relative_to(root).as_posix() + quarantined = is_quarantined(rel) + findings: list[Finding] = [] + for i in range(len(tokens) - size + 1): + digest = _hash(" ".join(tokens[i:i + size])) + if digest in corpus.hashes: + findings.append(Finding( + rel, line_of[i], size, quarantined, + also_in_cc0=bool(attributable and digest in attributable.hashes), + )) + return findings + + +def audit(corpus: Corpus, attributable: Corpus | None = None, root: Path = REPO_ROOT, + size: int = SHINGLE_WORDS) -> dict: + """Scan `root` against a *blocked* corpus, classifying hits that also occur in a CC0 corpus.""" + attributable = attributable or Corpus() + findings: list[Finding] = [] + scanned = 0 + for path in iter_text_files(root): + scanned += 1 + findings.extend(scan_file(path, corpus, root, size, attributable)) + + def summarise(rows: Iterable[Finding]) -> dict: + rows = list(rows) + blocked_only = [f for f in rows if not f.also_in_cc0] + return { + "hits": len(rows), + "blocked_only": len(blocked_only), + "attributable": len(rows) - len(blocked_only), + "files": [{"path": p, "hits": n, + "blocked_only": sum(1 for f in blocked_only if f.path == p)} + for p, n in sorted(Counter(f.path for f in rows).items())], + } + + first_party = summarise(f for f in findings if not f.quarantined) + quarantined = summarise(f for f in findings if f.quarantined) + return { + "format": "noopunk.issue569.similarity_audit", + "version": 1, + "issue": 569, + "shingle_words": size, + "corpus": {"documents": corpus.documents, "unique_shingles": len(corpus.hashes)}, + "attributable_corpus": {"documents": attributable.documents, + "unique_shingles": len(attributable.hashes)}, + "scanned_files": scanned, + "first_party": first_party, + "quarantined": quarantined, + "clean": first_party["blocked_only"] == 0, + } + + +def format_report(report: Mapping, *, limit: int = 25) -> str: + corpus = report["corpus"] + attributable = report["attributable_corpus"] + lines = [ + "Without Number expression audit — issue #569 (n-gram containment; no source text retained)", + f"corpus: {', '.join(f'{k} ({v} shingles)' for k, v in corpus['documents'].items())}" + f" — {corpus['unique_shingles']} unique {report['shingle_words']}-word runs", + f"attributable (CC0 SRD): {', '.join(f'{k} ({v})' for k, v in attributable['documents'].items())}" + f" — {attributable['unique_shingles']} unique runs", + f"scanned: {report['scanned_files']} text files", + "", + f"First-party hits: {report['first_party']['hits']} " + f"({report['first_party']['blocked_only']} blocked-only, " + f"{report['first_party']['attributable']} attributable)", + ] + for row in report["first_party"]["files"][:limit]: + lines.append(f" {row['path']}: {row['hits']} ({row['blocked_only']} blocked-only)") + if not report["first_party"]["files"]: + lines.append(" none — no first-party file shares a 10-word run with the corpus") + lines.append("") + lines.append(f"Quarantined hits: {report['quarantined']['hits']} " + f"({report['quarantined']['blocked_only']} blocked-only) " + f"in {len(report['quarantined']['files'])} preserved-corpus files") + for row in report["quarantined"]["files"][:limit]: + lines.append(f" {row['path']}: {row['hits']} ({row['blocked_only']} blocked-only)") + return "\n".join(lines) + + +def main(argv: Iterable[str] | None = None) -> int: + parser = argparse.ArgumentParser(description=__doc__, + formatter_class=argparse.RawDescriptionHelpFormatter) + parser.add_argument("--source", action="append", default=[], metavar="PDF", + help="licence-blocked book (can be repeated)") + parser.add_argument("--source-text", action="append", default=[], metavar="TXT", + help="pre-extracted text of a book (can be repeated)") + parser.add_argument("--cc0-source", action="append", default=[], metavar="PATH", + help="CC0 SRD (PDF, .md/.txt file, or a directory of markdown); " + "a hit that also appears here is attributable rather than a violation") + parser.add_argument("--root", default=str(REPO_ROOT), help="tree to scan") + parser.add_argument("--words", type=int, default=SHINGLE_WORDS, help="run length") + parser.add_argument("--json", action="store_true", help="emit JSON") + args = parser.parse_args(list(argv) if argv is not None else None) + + if not args.source and not args.source_text: + parser.error("at least one --source or --source-text is required") + blocked = corpus_from_paths([*args.source, *args.source_text]) + cc0 = corpus_from_paths(args.cc0_source) + report = audit(blocked, cc0, Path(args.root), args.words) + print(json.dumps(report, indent=2) if args.json else format_report(report)) + return 0 if report["clean"] else 1 + + +if __name__ == "__main__": + raise SystemExit(main())