diff --git a/README.md b/README.md index 62f32c9..a698c64 100644 --- a/README.md +++ b/README.md @@ -211,7 +211,9 @@ NO_COLOR=1 TERM=xterm-256color LC_ALL=C.UTF-8 \ uv run pytest tests/presentation/test_snapshots.py -m snapshot --snapshot-update ``` -Regenerate the demo GIF after updating snapshots: +Regenerate the demo GIF after UI changes. The script drives the real app with +a simulated typist (seeded history, mid-lesson practice, stats, settings), so +it does not depend on the snapshot baselines: ```bash uv pip install pillow cairosvg # one-off; system cairo on macOS/Linux diff --git a/docs/CHANGELOG.md b/docs/CHANGELOG.md index 8c6749f..bd7f80d 100644 --- a/docs/CHANGELOG.md +++ b/docs/CHANGELOG.md @@ -5,6 +5,35 @@ rationale lives in commit history/diffs — these are pointers, not narratives. Milestone-level feature work (what shipped in M1–M4, the keybr algorithm design) stays in `PLAN.md` §5/§6. +## 2.2.0 + +- **One focus ladder** — `select_lesson_focus` (`domain/focus.py`) replaces + `select_focus`, `select_focus_transition`, `remedial_focus` and the + newest-key unmeasured-pair fallback. Order: keys not cleared, unlock-gate + cohort pairs, weak pairs by (1 − confidence) × language frequency + (`TransitionTable.pair_weight`), then review. Each rule keeps the last + lesson's focus until it clears or stalls at 3× its attempt floor + (`domain.confidence.stall_attempts_cap`, moved from `domain.unlock`). + Removed: the lesson WPM gate (`remedial_focus`) and the review penalty in + focus scoring. Sessions now save `focus_pair` when the focus was a pair + (written only when set, still schema 5; old rows read back unchanged). +- **Attempt floors count real presses** — `combine_sessions` now sums + `attempt_count` over the window without recency weights; speed and accuracy + keep theirs. Weighted attempts settled at about 3.24 × presses per lesson, + so a key or cohort pair with 2–3 presses per lesson never reached + `min_confidence_attempts` (10) or the transition stall cap (12) and blocked + the next letter. The aggregates cache moves to `schema_version` 2; older + cache files read as a miss and rebuild from the session rows (session rows + are unchanged). +- **Accuracy is scored against 95%, not 100%** — skill and confidence use + `min(speed, min(1, accuracy / ACCURACY_TARGET))` with + `ACCURACY_TARGET = 0.95` (`domain/confidence.py`), so one typo no longer + needs about 199 clean presses before a key clears. Recency-weighted errors + stay fractional instead of being rounded, so one typo in the previous + lesson counts as 0.7. The transition gate now calls the same + `clears_threshold` rule as key unlock and focus. HUD and Stats still show + raw accuracy. + ## 2.1.0 - **Session store: per-session stats instead of keystroke logs** — a finished diff --git a/docs/PLAN.md b/docs/PLAN.md index fe9a0c7..eef0d4a 100644 --- a/docs/PLAN.md +++ b/docs/PLAN.md @@ -68,7 +68,7 @@ keystrike/ │ │ ├── protocols.py # Clock, IdGenerator, SessionRepository, SettingsRepository, LayoutRepository, LanguageProvider, CodeSnippetProvider, AggregatesCache, FreeformTextProvider, StatsRebuilder, LearningRateEstimator │ │ ├── null_adapters.py # NullSessionRepository, NullStatsRebuilder, NullLearningRateEstimator [DONE] │ │ ├── aggregate.py # pure: Keystroke → KeyStats rollup [DONE] -│ │ ├── confidence.py # pure: key_confidence, confidence_of, select_focus, compute_unlocked [DONE] +│ │ ├── confidence.py # pure: key_confidence, confidence_of, select_lesson_focus, compute_unlocked [DONE] │ │ ├── generator.py # pure: AdaptiveGenerator (Markov + focus letter) [DONE] │ │ ├── markov.py # TransitionTable value object + sampler [DONE] │ │ ├── regression.py # pure: polynomial fit, estimate_sessions_to_goal [DONE] @@ -233,7 +233,7 @@ all fixed before M3 began (95 tests passing at that point) — see shorter contexts, filtered to the unlocked alphabet), `domain/generator.py` (`AdaptiveGenerator`: word length 2–4 by default, `p_stop = min(1, 1.3**length/max_len)`, focus letter guaranteed via injection if the Markov walk didn't produce it), -`domain/confidence.py` (`confidence_of`, `compute_unlocked`, `select_focus` — +`domain/confidence.py` (`confidence_of`, `compute_unlocked`, `select_lesson_focus` — faithful to §6 below), `application/build_lesson.py` (`BuildLesson` ties layout + stats + settings + language provider into a `Lesson(text, state)`), `infrastructure/languages/` (`BundledLanguageProvider`, gzip+JSON), and @@ -288,7 +288,7 @@ characters (`():=_[]"`) that aren't part of any `Layout.learn_order` — a strict filter would reject nearly every real snippet. Instead `select_snippet` weight-picks by how often the snippet contains the focus char (`snippet.count(focus_char) + 1`), so unlock/focus state still comes -from the exact same `compute_unlocked`/`select_focus` logic as English mode +from the exact same `compute_unlocked`/`select_lesson_focus` logic as English mode (factored out to `_lesson_progress()`, shared by `BuildLesson` and the new `BuildCodeLesson` in `application/build_lesson.py`) — it just doesn't gate which literal characters can appear in the practice text. diff --git a/docs/assets/demo.gif b/docs/assets/demo.gif index 2400e3d..5f16a09 100644 Binary files a/docs/assets/demo.gif and b/docs/assets/demo.gif differ diff --git a/docs/research/typing-pedagogy.md b/docs/research/typing-pedagogy.md index 701acef..2f46a91 100644 --- a/docs/research/typing-pedagogy.md +++ b/docs/research/typing-pedagogy.md @@ -48,6 +48,6 @@ Sources: ## Gap found in current implementation -~~`confidence_of`/`key_confidence` (`confidence.py:18-36`) compute mastery purely from `mean_time_ns` vs. target speed. `KeyStats.error_count` is tracked and persisted (`aggregate.py:55`, `models.py:34`) but never read by `confidence_of`, `compute_unlocked`, or `select_focus` — so a key typed fast but frequently wrong currently reads as "confident" and can unlock the next key prematurely, contradicting Keybr's dual speed+accuracy threshold and the accuracy-first principle (point 2 above).~~ **Fixed**: `confidence_of` uses `min(speed, accuracy)` where speed is `target_ms / actual_ms` and accuracy is `samples / (samples + error_count)`. Fast-but-sloppy or slow-but-accurate keys no longer compensate each other (unlike a product). Windowed aggregates also recency-weight mean time, accuracy, and attempt counts so stale practice doesn't mask recent sloppiness. `compute_unlocked`/`select_focus`/`GetHeatmap` all inherit this via `confidence_of`. +~~`confidence_of`/`key_confidence` (`confidence.py:18-36`) compute mastery purely from `mean_time_ns` vs. target speed. `KeyStats.error_count` is tracked and persisted (`aggregate.py:55`, `models.py:34`) but never read by `confidence_of`, `compute_unlocked`, or `select_focus` — so a key typed fast but frequently wrong currently reads as "confident" and can unlock the next key prematurely, contradicting Keybr's dual speed+accuracy threshold and the accuracy-first principle (point 2 above).~~ **Fixed**: `confidence_of` uses `min(speed, accuracy score)` where speed is `target_ms / actual_ms`, accuracy is `samples / (samples + error_count)`, and the accuracy score is accuracy ÷ `ACCURACY_TARGET` (0.95, capped at 1.0) — accuracy above about 95% is the goal, not zero errors (see the higher-ed study above). Fast-but-sloppy or slow-but-accurate keys no longer compensate each other (unlike a product). Windowed aggregates also recency-weight mean time and accuracy (errors stay fractional) so stale practice doesn't mask recent sloppiness; attempt counts stay unweighted, because they are the evidence for the attempt floors and the transition stall cap. `compute_unlocked`/`select_lesson_focus`/`GetHeatmap` all inherit this via `confidence_of`. ~~`AdaptiveGenerator.generate_lesson` (`generator.py:30-37`) sampled every word from the plain language-frequency Markov table and only guaranteed the focus key appeared *once* across the whole lesson (via `_inject_focus`) — everything else was generic full-alphabet text, unlike Keybr's "adaptive per-letter frequency weighting based on personal error/latency history" (line 20 above) and short of the deliberate-practice principle (point 4).~~ **Fixed**: `confidence.py` now has `practice_weight(confidence, max_bias=3.0)` — weak keys get up to 4x sampling weight, mastered keys (confidence ≥ 1.0) get baseline weight 1.0. `TransitionTable.sample` (`markov.py`) takes an optional `char_weights` map that multiplies each candidate's language-frequency weight before `rng.choices`; `AdaptiveGenerator.generate_word`/`generate_lesson` thread it through. `BuildLesson` (`build_lesson.py`) computes `char_weights` from each unlocked key's live confidence, so generated text is now generally biased toward weak keys, not just guaranteed to touch the single focus key once. diff --git a/docs/wiki/Confidence-Tuning.md b/docs/wiki/Confidence-Tuning.md index 1d08444..7dc44d5 100644 --- a/docs/wiki/Confidence-Tuning.md +++ b/docs/wiki/Confidence-Tuning.md @@ -22,7 +22,7 @@ keys are ignored, and missing nested tables silently fall back to defaults). | Setting | `settings.toml` key | Default | What it does | | --- | --- | --- | --- | | Confidence session window | `confidence_session_window` | `10` | How many recent sessions are replayed into rolling per-key stats used for skill, unlocks, focus, and the heatmap. | -| Min key attempts | `[unlock].min_confidence_attempts` | `10` | Minimum presses on a key before its confidence reaches full weight. Below this, confidence ramps linearly (fewer attempts → lower score). | +| Min key attempts | `[unlock].min_confidence_attempts` | `10` | Minimum presses on a key in the session window before its confidence reaches full weight. Below this, confidence ramps linearly (fewer attempts → lower score). | | Min bigram attempts | `[unlock].min_transition_confidence_attempts` | `4` | Same ramp for letter-pair (transition) confidence. Default is lower because bigrams are practiced less often than single keys. | | Gating bigram limit | `[unlock].gating_bigram_limit` | `4` | Directed newest-letter bigrams that must calibrate before the next letter opens. Values are clamped to `2`–`4`. | | Next-letter unlock threshold | `[unlock].next_letter_unlock_threshold` | `1.0` | Skill threshold every currently-unlocked key must clear before the next letter opens. | @@ -43,11 +43,16 @@ multipliers should be **≥ 1.0**. `lesson_word_count` should be **≥ 1**. `word_min_fraction` should be in **(0.0, 1.0]**. `max_word_repeats` should be **≥ 1**. Generated word bounds should be **≥ 1** with min ≤ max. -Skill and confidence both use **min(speed, accuracy)**, not their product: a key -must be both fast enough and accurate enough to read as mastered. Speed is -`target_ms / actual_ms`; accuracy is correct attempts ÷ total attempts. Skill -is that ratio without attempt ramp; confidence scales skill by how many presses -you have in the window (see min attempt floors above). +Skill and confidence both use **min(speed, accuracy score)**, not their +product: a key must be both fast enough and accurate enough to read as +mastered. Speed is `target_ms / actual_ms`; accuracy is correct attempts ÷ +total attempts. The accuracy score is accuracy ÷ 95% (`ACCURACY_TARGET` in +`domain/confidence.py`), capped at 1.0: the goal is accuracy above about 95%, +not zero errors (see `docs/research/typing-pedagogy.md`). So a key at target +speed with 60 correct presses and one recent typo (98.4%) clears; 90% does +not. Skill is that ratio without attempt ramp; confidence scales skill by how +many presses you have in the window (see min attempt floors above). The HUD +and Stats show raw accuracy. Example (defaults shown): @@ -80,8 +85,12 @@ max_len = 4 **Session window** — Confidence is computed from aggregates over your last *N* sessions (per layout), not your entire history. Within that window, **recent sessions count more** than older ones (exponential decay, default 0.7 per step -back), so a bad or good last session moves confidence faster than the flat -average of all window sessions. A longer window smooths noise but reacts slowly; +back) for speed and accuracy, so a bad or good last session moves confidence +faster than the flat average of all window sessions. Errors keep their weight +as a fraction: one typo two lessons back counts as 0.7 of an error, not a +full one. Attempt counts are **not** weighted: the attempt floors and the +stall cap below count every real press in the window, so a key or pair that +gets two or three presses per lesson still reaches its floor. A longer window smooths noise but reacts slowly; a shorter window tracks recent form but can under-sample rare keys and block unlocks until those keys appear often enough in recent drills. @@ -94,8 +103,8 @@ This pruning is irreversible: if you raise `confidence_session_window` later, the wider window fills only with sessions recorded after the change. **Min key attempts** — Prevents a lucky fast streak from reading as mastery. -Until you've pressed a key at least this many times (within the windowed -stats), its confidence is scaled down. Higher values mean slower unlocks and +Until you've pressed a key at least this many times (real presses within the +session window), its confidence is scaled down. Higher values mean slower unlocks and more conservative focus selection; `1` effectively disables the ramp. **Min bigram attempts** — Same idea for prev→next letter pairs. Transition @@ -121,41 +130,20 @@ directions with up to its two most-recent practiced peers, bounded by transition data, so incidental measurements cannot expand or replace it. Every member must reach confidence 1.0 with `[unlock].min_transition_confidence_attempts` attempts. -A `transition_stall_attempts_cap` (`domain.unlock.default_transition_stall_attempts_cap`, -3× the transition calibration floor by default) releases a specific pair -that's been drilled past the cap without clearing threshold, so one stubborn -bigram can't block progression forever. - -**Focus selection** is letter-first: while any unlocked key is below the skill -threshold or key-attempt floor, the lesson emphasizes the weakest unlocked -**key**. While transition progression is blocked, deficient members of the -same gating cohort drive transition focus and share guaranteed lesson coverage. -An older pair can preempt only after enough samples show genuine raw -speed/accuracy regression; sparse old calibration does not delay progression. -Outside a blocked progression gate, normal measured transition review remains. - -**Focus is sticky.** Once a key or bigram becomes the focus, it keeps that -focus across lesson builds until it individually clears both the skill -threshold and its attempt floor (the same two-part gate as unlocks, via -`domain.confidence.clears_threshold`) — an unrelated stale-but-mastered key's -review urgency, or the transition gate activating, cannot steal focus away -mid-calibration. Only once *every* unlocked key/bigram has cleared does -review-urgency-based staleness compete for focus across the whole set. - -## Lesson WPM gate - -A finished session's own words-per-minute is compared against its own target -speed (converted CPM→WPM the same way `SettingsScreen` does). If that -session's WPM fell short, the *next* lesson's focus is confined to the -weakest key or bigram from that session's own `lesson_alphabet` — the letters -that actually appeared in the text — instead of ordinary weakest-across-window -selection or the newest-key transition gate picking elsewhere. This remains in -effect lesson-over-lesson until a lesson's WPM meets target again, at which -point normal focus selection (including stickiness, above) resumes. Sessions -with no recorded target (`target_speed_cpm == 0`, e.g. very old data) never -trigger this gate. There is no separate HUD label — the gate only narrows -*which* key/bigram is eligible for focus, so the usual `wk`/`cal`/`rev` reason -still applies to whichever one is chosen. +Every member must pass the same `domain.confidence.clears_threshold` rule as +key unlock and focus. +A `transition_stall_attempts_cap` (`domain.confidence.stall_attempts_cap`, +3× the transition calibration floor by default, so 12 real attempts in the +window) releases a specific pair that's been drilled past the cap without +clearing threshold, so one stubborn bigram can't block progression forever. +At two cohort-coverage slots per lesson, a stuck pair reaches the cap on its +sixth lesson. + +**Focus selection** follows one ordered list: keys that have not cleared, +then the unlock-gate cohort pairs, then weak pairs ranked by weakness × +language frequency, then review of the most overdue key or pair. The last +lesson's focus is kept while it still needs work, until it stalls at the same +3× attempt cap. See [Focus states](Focus-States) for the full rules. ### Transition stats diff --git a/docs/wiki/Focus-States.md b/docs/wiki/Focus-States.md index d9aa96a..d58c99f 100644 --- a/docs/wiki/Focus-States.md +++ b/docs/wiki/Focus-States.md @@ -38,40 +38,37 @@ Reading left to right: 2. **Reason** — `wk`, `cal`, or `rev`. Calibrating adds press progress (`9/10`). 3. **Speed** — target timing ÷ actual timing for the focus key or pair. 4. **Accuracy** — correct attempts ÷ total attempts (percent). -5. **Confidence** — min(speed, accuracy), scaled by attempt count during calibration. Goal is 1.0 for mastery. +5. **Confidence** — min(speed, accuracy score), scaled by attempt count during calibration. The accuracy score is accuracy ÷ 95%, capped at 1.0, so 95% accuracy already counts in full. Goal is 1.0 for mastery. -## Key vs transition focus +## How focus is chosen -**Key focus** applies while any unlocked key is below the performance skill -threshold or the configured key-attempt floor (see `blocks_transition_focus` -in the codebase). Never-typed and sparsely sampled keys therefore take priority. -The lesson emphasizes the weakest unlocked key by ramped confidence, adjusted -for review urgency. +Focus follows one ordered list of rules. The first rule that has candidates +picks the focus (`select_lesson_focus` in `domain/focus.py`): -**Transition focus** applies after every unlocked key meets both requirements. -When the next letter is gated, the same stable 2–4 newest-letter cohort controls -unlocking and lesson coverage. Sparse older pairs cannot preempt it; an older -pair can preempt only after enough samples show raw performance regression. -Same-key repeats (double letters) never count as transitions. +1. **Keys first.** If any unlocked key has not cleared (skill below 1.0 or + fewer presses than `[unlock].min_confidence_attempts`), the weakest such + key is the focus. Never-typed keys count as not cleared. +2. **The unlock gate.** If the next letter waits on the newest key's pair + cohort, the weakest pair of that cohort that is not ready is the focus. +3. **Weak pairs.** Otherwise, the pair that is not cleared and has the highest + (1 − confidence) × language frequency is the focus. Pairs that do not occur + in the language are skipped. Same-key repeats (double letters) never count. +4. **Review.** When everything has cleared, the most overdue key or pair + (by review urgency) is the focus. +5. **Fallback.** Otherwise, the weakest key. ## Focus stays put until it clears -Once a key or pair becomes the focus, it keeps that focus lesson-over-lesson -until it individually clears both confidence (skill ≥ 1.0) and its attempt -floor — the same bar unlocks use. A different key/pair going stale, or a -transition gate opening up, does not steal focus away mid-calibration. Once -everything unlocked has cleared, focus is free to move to whichever key/pair -is due for review (`rev`) again. - -## Lesson WPM gate - -If a finished lesson's overall words-per-minute came in under its own target, -the next lesson's focus is pinned to the weakest key or pair from *that -lesson's own text* until a lesson's WPM meets target again — a low-speed -lesson pulls focus back to its own weak point rather than drifting to -whatever else is weakest across the full practice history. This doesn't add a -new HUD label; the chosen key/pair still shows the ordinary `wk`, `cal`, or -`rev` reason. +In rules 1–3, the last lesson's focus is kept while it is still a candidate +of that rule. So a key or pair keeps the focus lesson-over-lesson until it +clears both skill and its attempt floor — the same bar unlocks use. + +A focus that does not clear after 3× its attempt floor in the session window +(30 presses for a key, 12 for a pair, by default) is **stalled**. It goes back +into the pool, so the weakest candidate is picked again from all of them. + +The last focus comes from the last saved session: `focus_key`, plus +`focus_pair` when the focus was a pair. ## Heatmap underline diff --git a/pyproject.toml b/pyproject.toml index a04dd9a..a1a6ee9 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "keystrike" -version = "2.1.0" +version = "2.2.0" description = "Offline terminal typing tutor with keybr-style adaptive drills and per-layout stats" readme = "README.md" requires-python = ">=3.12" diff --git a/scripts/generate_demo_gif.py b/scripts/generate_demo_gif.py index e7774d1..473b500 100644 --- a/scripts/generate_demo_gif.py +++ b/scripts/generate_demo_gif.py @@ -1,56 +1,164 @@ -"""Build docs/assets/demo.gif from presentation snapshot SVGs. +"""Build docs/assets/demo.gif from a seeded, simulated Keystrike run. -Run after `uv sync --group snapshot` and -`pytest tests/presentation/test_snapshots.py --snapshot-update`. +Drives the real app (with the test fakes for clock/storage) through a +deterministic typist who is fast on the home row, slow and error-prone on a +few keys, and improves over ~16 sessions. Then screenshots Home, a lesson in +progress, Stats (trends + heatmap), one letter's detail, and Settings. Requires pillow + cairosvg (one-off: `uv pip install pillow cairosvg`). -On macOS with Homebrew cairo: `DYLD_LIBRARY_PATH=/opt/homebrew/lib`. +On macOS `brew install cairo`; the script finds the Homebrew library itself. """ from __future__ import annotations +import asyncio import io import os +import random import sys from pathlib import Path - -import cairosvg -from PIL import Image +from typing import cast _ROOT = Path(__file__).resolve().parent.parent -_SNAPSHOTS = _ROOT / "tests/presentation/__snapshots__/test_snapshots" -_OUT = _ROOT / "docs/assets/demo.gif" +sys.path.insert(0, str(_ROOT)) # `tests.*` fakes and the fully-wired test app -_FRAME_MS = 10_000 # ~10 s per frame in the README demo GIF - -_FRAMES = ( - "test_home_screen_snapshot.svg", - "test_practice_screen_snapshot.svg", - "test_stats_screen_snapshot.svg", - "test_settings_screen_snapshot.svg", - "test_home_screen_snapshot.svg", +from tests.fakes import FakeClock # noqa: E402 +from tests.presentation.test_practice_screen import ( # noqa: E402 + _build_app, # pyright: ignore[reportPrivateUsage] ) +from textual.pilot import Pilot # noqa: E402 +from keystrike.presentation.screens.practice import PracticeScreen # noqa: E402 -def _svg_to_image(path: Path) -> Image.Image: - png = cairosvg.svg2png(url=str(path), output_width=720) - return Image.open(io.BytesIO(png)).convert("P", palette=Image.ADAPTIVE) +_OUT = _ROOT / "docs/assets/demo.gif" +_FRAME_MS = 5_000 +_TERMINAL = (80, 24) +_SEED_SESSIONS = 16 +_MID_LESSON_FRACTION = 0.45 # how much of the lesson is typed in the practice frame + +# Textual key names for the non-letter characters a lesson can contain. +_KEY_NAMES = {" ": "space", ",": "comma", ".": "full_stop", ";": "semicolon", "/": "slash"} + +_HOME_ROW = set("asdfghjkl") +_WEAK_KEYS = {"r": 2.0, "u": 1.8, "w": 1.6} # slower and error-prone throughout +# Per-keystroke seconds before jitter/improvement. Tuned against the default +# 100 WPM goal (120 ms/char) so the final heatmap mixes green, yellow and red. +_BASE_S = {"home": 0.16, "other": 0.26} +_SPEEDUP_PER_SESSION = 0.97 +_SLIP_P = {"weak": 0.25, "other": 0.04} +_SLIP_DECAY_PER_SESSION = 0.95 + + +def _setup_cairo_path() -> None: + if sys.platform != "darwin" or "DYLD_LIBRARY_PATH" in os.environ: + return + # `brew install cairo` does not always link libcairo into /opt/homebrew/lib. + for brew_lib in (Path("/opt/homebrew/opt/cairo/lib"), Path("/opt/homebrew/lib")): + if (brew_lib / "libcairo.2.dylib").is_file(): + os.environ["DYLD_LIBRARY_PATH"] = str(brew_lib) + return + + +class _Typist: + """Deterministic keystroke timing/error model that improves per session.""" + + def __init__(self, rng: random.Random) -> None: + self._rng = rng + self.session = 0 + + def delay_ns(self, char: str) -> int: + base = _BASE_S["home"] if char in _HOME_ROW or char == " " else _BASE_S["other"] + base *= _WEAK_KEYS.get(char, 1.0) + base *= _SPEEDUP_PER_SESSION**self.session # gets faster over time + jitter = self._rng.uniform(0.8, 1.25) + return int(base * jitter * 1e9) + + def slips(self, char: str) -> bool: + p = _SLIP_P["weak"] if char in _WEAK_KEYS else _SLIP_P["other"] + return self._rng.random() < p * _SLIP_DECAY_PER_SESSION**self.session + + def wrong_key(self, char: str) -> str: + pool = [c for c in "asdfghjklqwertyuiopzxcvbnm" if c != char] + return self._rng.choice(pool) + + +def _key(char: str) -> str: + return _KEY_NAMES.get(char, char) + + +def _lesson_text(screen: object) -> str: + assert isinstance(screen, PracticeScreen) + return screen._session.target_text # pyright: ignore[reportPrivateUsage] + + +async def _type(pilot: Pilot[None], clock: FakeClock, typist: _Typist, text: str) -> None: + for char in text: + if typist.slips(char) and char != " ": + clock.advance(typist.delay_ns(char)) + await pilot.press(_key(typist.wrong_key(char))) + clock.advance(typist.delay_ns(char)) + await pilot.press(_key(char)) + + +async def _seed_and_capture() -> list[str]: + app, clock, repo, _settings = _build_app() + rng = random.Random(7) + typist = _Typist(rng) + svgs: list[str] = [] + + async with app.run_test(size=_TERMINAL) as pilot: + await pilot.press("enter") # Home → Practice + await pilot.pause() + for i in range(_SEED_SESSIONS): + typist.session = i + clock.wall += 6 * 60 # sessions a few minutes apart, all "today" + await _type(pilot, clock, typist, _lesson_text(app.screen)) + await pilot.pause() + assert len(repo.headers) == i + 1, "lesson did not finish" + + # Frame: a lesson in progress with a slip already made. + typist.session = _SEED_SESSIONS + text = _lesson_text(app.screen) + cut = int(len(text) * _MID_LESSON_FRACTION) + await _type(pilot, clock, typist, text[:cut]) + await pilot.pause() + practice_svg = app.export_screenshot() + + await pilot.press("escape") # abort the in-progress lesson → Home + await pilot.pause() + svgs.append(app.export_screenshot()) # Home + svgs.append(practice_svg) + + await pilot.press("s") + await pilot.pause() + svgs.append(app.export_screenshot()) # Stats overview + await pilot.press("r") + await pilot.pause() + svgs.append(app.export_screenshot()) # Letter detail for a weak key + await pilot.press("escape") + await pilot.press("escape") + await pilot.pause() + + await pilot.press("o") + await pilot.pause() + svgs.append(app.export_screenshot()) # Settings + await pilot.press("escape") + await pilot.pause() + + svgs.append(svgs[0]) # loop back to Home + return svgs def main() -> None: - if sys.platform == "darwin" and "DYLD_LIBRARY_PATH" not in os.environ: - # `brew install cairo` does not always link libcairo into /opt/homebrew/lib. - for brew_lib in (Path("/opt/homebrew/opt/cairo/lib"), Path("/opt/homebrew/lib")): - if (brew_lib / "libcairo.2.dylib").is_file(): - os.environ["DYLD_LIBRARY_PATH"] = str(brew_lib) - break + _setup_cairo_path() + import cairosvg # noqa: PLC0415 (needs DYLD_LIBRARY_PATH set first) + from PIL import Image # noqa: PLC0415 + svgs = asyncio.run(_seed_and_capture()) frames: list[Image.Image] = [] - for name in _FRAMES: - svg = _SNAPSHOTS / name - if not svg.is_file(): - raise SystemExit(f"missing snapshot: {svg}") - frames.append(_svg_to_image(svg)) + for svg in svgs: + png = cast("bytes", cairosvg.svg2png(bytestring=svg.encode("utf-8"), output_width=720)) + frames.append(Image.open(io.BytesIO(png)).convert("P", palette=Image.Palette.ADAPTIVE)) _OUT.parent.mkdir(parents=True, exist_ok=True) frames[0].save( @@ -61,7 +169,7 @@ def main() -> None: loop=0, optimize=True, ) - print(f"wrote {_OUT} ({_OUT.stat().st_size // 1024} KiB)") + print(f"wrote {_OUT} ({_OUT.stat().st_size // 1024} KiB, {len(frames)} frames)") if __name__ == "__main__": diff --git a/src/keystrike/application/build_lesson.py b/src/keystrike/application/build_lesson.py index 17b83f4..e905800 100644 --- a/src/keystrike/application/build_lesson.py +++ b/src/keystrike/application/build_lesson.py @@ -9,14 +9,12 @@ from random import Random from keystrike.application.alphabet_sync import sync_alphabet_size -from keystrike.application.session_queries import ( - latest_session_header, - session_wpm_below_target, -) +from keystrike.application.session_queries import latest_session_header from keystrike.domain.confidence import ( CONFIDENCE_GOOD, accuracy_of, attempts_of, + clears_threshold, confidence_of, is_same_key_transition, key_confidence, @@ -24,6 +22,7 @@ round_confidence, skill_from_stats, skill_of, + stall_attempts_cap, target_ms_per_char, transition_accuracy_of, transition_confidence, @@ -32,13 +31,9 @@ from keystrike.domain.enums import FocusKind from keystrike.domain.focus import ( FocusReason, - blocks_transition_focus, coverage_deficit_factor, - focus_key_from_transition, practice_weight, - remedial_focus, - select_focus, - select_focus_transition, + select_lesson_focus, ) from keystrike.domain.generator import ( AdaptiveGenerator, @@ -48,6 +43,7 @@ weak_focus_word_quota, ) from keystrike.domain.learn_order import keyboard_order +from keystrike.domain.markov import TransitionTable from keystrike.domain.models import ( Bigram, KeyStats, @@ -70,7 +66,6 @@ ) from keystrike.domain.unlock import ( compute_unlocked, - default_transition_stall_attempts_cap, gating_bigram_is_ready, newest_key_transition_gate_progress, ) @@ -249,6 +244,7 @@ class Lesson: state: LessonState focus_reason: FocusReason | None skill_heatmap: dict[int, float] + focus_pair: Bigram | None = None focus_confidence: float | None = None focus_speed: float | None = None focus_accuracy: float | None = None @@ -276,9 +272,7 @@ class _GatingState: def _gating_state(ctx: _LessonContext) -> _GatingState: order = keyboard_order(ctx.layout) - stall_cap = default_transition_stall_attempts_cap( - ctx.settings.unlock.min_transition_confidence_attempts - ) + stall_cap = stall_attempts_cap(ctx.settings.unlock.min_transition_confidence_attempts) unlocked = compute_unlocked( order, ctx.settings.alphabet_size, @@ -288,12 +282,14 @@ def _gating_state(ctx: _LessonContext) -> _GatingState: transitions=ctx.transitions, transition_stall_attempts_cap=stall_cap, ) - keys_need_focus = blocks_transition_focus( - unlocked, - ctx.stats, - ctx.target, - threshold=CONFIDENCE_GOOD, - min_attempts=ctx.settings.unlock.min_confidence_attempts, + keys_need_focus = not all( + clears_threshold( + ctx.stats.get(cp), + ctx.target, + threshold=CONFIDENCE_GOOD, + min_attempts=ctx.settings.unlock.min_confidence_attempts, + ) + for cp in unlocked ) cohort = newest_key_gating_cohort( unlocked, @@ -337,45 +333,19 @@ def _resolve_lesson_focus( ctx: _LessonContext, gating: _GatingState, ) -> tuple[int, Bigram | None]: - remedial = None - if ctx.remedial_alphabet is not None: - remedial = remedial_focus( - ctx.remedial_alphabet, - gating.unlocked, - ctx.stats, - ctx.transitions, - ctx.target, - now=ctx.now, - threshold=CONFIDENCE_GOOD, - min_attempts=ctx.settings.unlock.min_confidence_attempts, - min_transition_attempts=ctx.settings.unlock.min_transition_confidence_attempts, - ) - focus_bigram: Bigram | None = None - if remedial is not None: - focus, focus_bigram = remedial - else: - if not gating.keys_need_focus: - focus_bigram = select_focus_transition( - gating.unlocked, - ctx.transitions, - ctx.target, - ctx.now, - key_stats=ctx.stats, - gating_candidates=gating.gating_bigrams if gating.transition_blocked else None, - min_attempts=ctx.settings.unlock.min_transition_confidence_attempts, - ) - - if focus_bigram is not None: - focus = focus_key_from_transition(*focus_bigram) - else: - focus = select_focus( - gating.unlocked, - ctx.stats, - ctx.target, - ctx.now, - min_attempts=ctx.settings.unlock.min_confidence_attempts, - ) - return focus, focus_bigram + return select_lesson_focus( + gating.unlocked, + ctx.stats, + ctx.transitions, + ctx.target, + ctx.now, + pair_frequency=ctx.table.pair_weight, + gate=gating.gating_bigrams, + last_key=ctx.last_key, + last_pair=ctx.last_pair, + min_attempts=ctx.settings.unlock.min_confidence_attempts, + min_transition_attempts=ctx.settings.unlock.min_transition_confidence_attempts, + ) def _lesson_progress( @@ -501,9 +471,10 @@ class _LessonContext: transitions: Mapping[Bigram, TransitionStats] now: float target: float - # The just-finished session's own lesson alphabet, when that session's own - # WPM fell short of its own target -- None otherwise (normal selection). - remedial_alphabet: tuple[int, ...] | None = None + table: TransitionTable + # The last finished session's focus, kept while it still needs work. + last_key: int | None = None + last_pair: Bigram | None = None @dataclass(slots=True) @@ -547,6 +518,7 @@ def __call__(self, layout_name: str) -> Lesson: state=progress.state, focus_reason=explanation.reason, skill_heatmap=progress.skill_heatmap, + focus_pair=progress.focus_bigram, focus_confidence=focus_confidence if explanation.reason else None, focus_speed=explanation.speed, focus_accuracy=explanation.accuracy, @@ -559,6 +531,7 @@ def _load_context(self, layout_name: str) -> _LessonContext: settings = self.settings_repo.load() layout = self.layout_repo.get(layout_name) aggregates = self.aggregates_cache.get(layout_name) + last = latest_session_header(self.session_repo, layout_name) stats: Mapping[int, KeyStats] = aggregates.keys if aggregates else {} transitions: Mapping[Bigram, TransitionStats] = aggregates.transitions if aggregates else {} return _LessonContext( @@ -568,19 +541,11 @@ def _load_context(self, layout_name: str) -> _LessonContext: transitions=transitions, now=self.clock.wall_epoch(), target=target_ms_per_char(settings.target_speed_cpm), - remedial_alphabet=self._remedial_alphabet(layout_name), + table=self.language_provider.transitions(settings.lang), + last_key=last.focus_key if last is not None else None, + last_pair=last.focus_pair if last is not None else None, ) - def _remedial_alphabet(self, layout_name: str) -> tuple[int, ...] | None: - """The last finished session's own lesson alphabet, when that - session's own WPM missed its own target -- otherwise None.""" - last = latest_session_header(self.session_repo, layout_name) - if last is None: - return None - if not session_wpm_below_target(last): - return None - return last.lesson_alphabet - def _generate_text( self, ctx: _LessonContext, @@ -593,8 +558,7 @@ def _generate_text( transition_weights: dict[Bigram, float], gating_bigrams: tuple[Bigram, ...], ) -> str: - table = self.language_provider.transitions(ctx.settings.lang) - generator = AdaptiveGenerator(table=table, rng=self.rng) + generator = AdaptiveGenerator(table=ctx.table, rng=self.rng) weighting = LessonWeighting( char_weights=char_weights, transition_weights=transition_weights, diff --git a/src/keystrike/application/prepare_practice.py b/src/keystrike/application/prepare_practice.py index 76d23d1..945edf6 100644 --- a/src/keystrike/application/prepare_practice.py +++ b/src/keystrike/application/prepare_practice.py @@ -7,7 +7,7 @@ from keystrike.application.build_lesson import BuildLesson from keystrike.domain.enums import Mode from keystrike.domain.focus import FocusReason -from keystrike.domain.models import Layout +from keystrike.domain.models import Bigram, Layout from keystrike.domain.null_adapters import NULL_AGGREGATES_ENSURER from keystrike.domain.protocols import ( AggregatesEnsurer, @@ -23,6 +23,7 @@ class SessionPrep: layout: str mode: Mode focus_key: int | None + focus_pair: Bigram | None focus_reason: FocusReason | None focus_confidence: float | None focus_speed: float | None @@ -51,6 +52,7 @@ def __call__(self) -> SessionPrep | None: layout=settings.layout, mode=Mode.ADAPTIVE, focus_key=lesson.focus_key, + focus_pair=lesson.focus_pair, focus_reason=lesson.focus_reason, focus_confidence=lesson.focus_confidence, focus_speed=lesson.focus_speed, diff --git a/src/keystrike/application/session_queries.py b/src/keystrike/application/session_queries.py index bdab425..3bee747 100644 --- a/src/keystrike/application/session_queries.py +++ b/src/keystrike/application/session_queries.py @@ -2,7 +2,7 @@ from __future__ import annotations -from keystrike.domain.generator import typical_chars_per_word, wpm_from_cpm +from keystrike.domain.generator import typical_chars_per_word from keystrike.domain.models import SessionResult from keystrike.domain.protocols import SessionRepository @@ -44,30 +44,6 @@ def previous_session_header( def latest_session_header(repo: SessionRepository, layout: str) -> SessionResult | None: """Most recently finished session for ``layout``, if any — the header - `BuildLesson` reads to decide whether the next lesson needs the - lesson-WPM remedial-focus gate (`session_wpm_below_target`).""" + `BuildLesson` reads to keep the last lesson's focus.""" headers = list(repo.iter_headers(layout)) return max(headers, key=lambda h: h.started_at, default=None) - - -def session_wpm_below_target( - result: SessionResult, - *, - generated_min_len: int | None = None, - generated_max_len: int | None = None, -) -> bool: - """Whether a finished session's own WPM fell short of its own target - speed. Drives `BuildLesson`'s remedial lesson-alphabet focus gate; - legacy sessions with no recorded target (``target_speed_cpm == 0``) - never trigger it. Word-length bounds come from the session header - (snapshotted at finish); optional overrides exist for tests.""" - if result.target_speed_cpm <= 0 or result.words_completed <= 0: - return False - min_len = generated_min_len if generated_min_len is not None else result.generated_min_len - max_len = generated_max_len if generated_max_len is not None else result.generated_max_len - target_wpm = wpm_from_cpm( - result.target_speed_cpm, - generated_min_len=min_len, - generated_max_len=max_len, - ) - return compute_wpm(result) < target_wpm diff --git a/src/keystrike/application/session_use_cases.py b/src/keystrike/application/session_use_cases.py index bdb32a8..0aff852 100644 --- a/src/keystrike/application/session_use_cases.py +++ b/src/keystrike/application/session_use_cases.py @@ -13,12 +13,13 @@ ) from keystrike.domain.confidence import ( confidence_of, + stall_attempts_cap, target_ms_per_char, ) from keystrike.domain.enums import Mode, SessionState from keystrike.domain.generator import effective_generated_word_bounds from keystrike.domain.learn_order import keyboard_order -from keystrike.domain.models import Keystroke, SessionResult, SessionStats +from keystrike.domain.models import Bigram, Keystroke, SessionResult, SessionStats from keystrike.domain.null_adapters import ( NULL_LAYOUT_REPOSITORY, NULL_SETTINGS_REPOSITORY, @@ -39,7 +40,7 @@ note_keystroke_for_timer, skip_leading_whitespace, ) -from keystrike.domain.unlock import compute_unlocked, default_transition_stall_attempts_cap +from keystrike.domain.unlock import compute_unlocked @dataclass(slots=True) @@ -55,6 +56,7 @@ def __call__( mode: Mode = Mode.ADAPTIVE, lang: str = "en", focus_key: int | None = None, + focus_pair: Bigram | None = None, ) -> Session: return Session( id=self.id_gen.new_id(), @@ -65,6 +67,7 @@ def __call__( started_at_wall=self.clock.wall_epoch(), started_at_ns=self.clock.now_ns(), focus_key=focus_key, + focus_pair=focus_pair, ) @@ -132,7 +135,7 @@ class _DraftTiming: duration_ns: int -def _snapshot_unlock_state( +def _snapshot_finish_state( session: Session, duration_ns: int, stats: SessionStats, @@ -161,16 +164,17 @@ def _snapshot_unlock_state( target, tuning=settings.unlock, transitions=combined.transitions, - transition_stall_attempts_cap=default_transition_stall_attempts_cap( + transition_stall_attempts_cap=stall_attempts_cap( settings.unlock.min_transition_confidence_attempts ), ) - return unlocked, { + key_confidence = { cp: confidence_of( cp, combined.keys, target, min_attempts=settings.unlock.min_confidence_attempts ) for cp in unlocked } + return unlocked, key_confidence @dataclass(slots=True) @@ -191,7 +195,7 @@ def __call__(self, session: Session) -> SessionResult: settings = self.settings_repo.load() target_speed_cpm = settings.target_speed_cpm stats = tally_session(session.keystrokes) - unlocked_keys, key_confidence = _snapshot_unlock_state( + unlocked_keys, key_confidence = _snapshot_finish_state( session, duration_ns, stats, @@ -217,6 +221,7 @@ def __call__(self, session: Session) -> SessionResult: mode=session.mode, lesson_alphabet=tuple(sorted({ord(c) for c in session.target_text})), focus_key=session.focus_key, + focus_pair=session.focus_pair, total_keystrokes=session.total_count, correct_keystrokes=session.correct_count, words_completed=count_words_completed(session.target_text, session.position), diff --git a/src/keystrike/application/stats_use_cases.py b/src/keystrike/application/stats_use_cases.py index 7b84fbb..764e9e1 100644 --- a/src/keystrike/application/stats_use_cases.py +++ b/src/keystrike/application/stats_use_cases.py @@ -161,7 +161,7 @@ def _aggregate_speed_accuracy( target_ms: float, ) -> tuple[float, float]: total_samples = 0 - total_errors = 0 + total_errors = 0.0 weighted_time = 0.0 for key_stats in stats.values(): total_samples += key_stats.samples diff --git a/src/keystrike/domain/aggregate.py b/src/keystrike/domain/aggregate.py index 572e6e4..8988073 100644 --- a/src/keystrike/domain/aggregate.py +++ b/src/keystrike/domain/aggregate.py @@ -7,7 +7,8 @@ The reduction happens in two stages. `tally_session` folds the keystroke stream into exact integer `KeyTally` counters per key and per bigram — that `SessionStats` is what gets persisted with the session header. `combine_sessions` -then merges the stored tallies of a window of sessions with recency weights. +then merges the stored tallies of a window of sessions: recency weights on +speed and accuracy, plain counts for attempts. """ from __future__ import annotations @@ -228,7 +229,7 @@ class MergedFields: samples: int mean_time_ns: float - error_count: int + error_count: float attempt_count: int last_seen: float @@ -237,8 +238,9 @@ def _weighted_merge_fields( entries: Sequence[tuple[HasConfidenceFields, float]], ) -> MergedFields: weighted_samples = sum(weight * stats.samples for stats, weight in entries) + # Errors stay fractional: rounding would turn one older typo (weight 0.7) + # into a full error and fail the accuracy target on its own. weighted_errors = sum(weight * stats.error_count for stats, weight in entries) - weighted_attempts = sum(weight * stats.attempt_count for stats, weight in entries) if weighted_samples > 0: mean = ( sum(stats.mean_time_ns * weight * stats.samples for stats, weight in entries) @@ -250,8 +252,10 @@ def _weighted_merge_fields( return MergedFields( samples=_rounded_weighted_count(weighted_samples), mean_time_ns=mean, - error_count=round(weighted_errors), - attempt_count=_rounded_weighted_count(weighted_attempts), + error_count=weighted_errors, + # Unweighted: the attempt floors and the transition stall cap count + # real presses in the window, so decay cannot hold them below the bar. + attempt_count=sum(stats.attempt_count for stats, _ in entries), last_seen=last_seen, ) @@ -260,7 +264,7 @@ def _combine_key_maps_weighted( maps: Sequence[dict[int, KeyStats]], weights: Sequence[float], ) -> dict[int, KeyStats]: - """Merge key stats with recency weights on speed, accuracy, and attempts.""" + """Merge key stats with recency weights on speed and accuracy.""" by_cp: dict[int, list[tuple[KeyStats, float]]] = {} for m, weight in zip(maps, weights, strict=True): for cp, stats in m.items(): @@ -312,8 +316,10 @@ def combine_sessions( """Merge per-session stats into one layout aggregate. Sessions must be in chronological order. Recent sessions weigh more on - mean time, accuracy, and attempt counts so the sample ramp tracks recent - practice, not stale volume alone. + mean time and accuracy so stale practice cannot mask recent form. Attempt + counts are plain sums over the window: they are evidence (the attempt + floors and the stall cap), and decay would keep a key or pair that gets a + few presses per lesson below the floor forever. """ if not sessions: return LayoutAggregates(keys={}, transitions={}) @@ -334,12 +340,12 @@ def infer_key_stat_samples(samples: int, mean_time_ns: float) -> int: return samples -def infer_key_stat_attempt_count(samples: int, error_count: int, attempt_count: int) -> int: +def infer_key_stat_attempt_count(samples: int, error_count: float, attempt_count: int) -> int: """Legacy caches predate a stored `attempt_count`; treat samples + errors as the inferred total whenever the stored value is non-positive.""" inferred = samples + error_count if attempt_count <= 0 and inferred > 0: - return inferred + return max(1, round(inferred)) return attempt_count diff --git a/src/keystrike/domain/confidence.py b/src/keystrike/domain/confidence.py index d59ad00..fa01174 100644 --- a/src/keystrike/domain/confidence.py +++ b/src/keystrike/domain/confidence.py @@ -29,6 +29,11 @@ CONFIDENCE_DECIMALS = 2 # Mastery threshold: skill/confidence at or above this counts as "good". CONFIDENCE_GOOD = 1.0 +# Accuracy that counts as fully accurate for skill/confidence. The methodology +# (docs/research/typing-pedagogy.md: "accuracy first", structured training with +# accuracy above 95%) asks for a low error rate, not zero errors — so accuracy +# is scored against this target instead of against 100%. +ACCURACY_TARGET = 0.95 def round_confidence(value: float) -> float: @@ -60,7 +65,7 @@ class HasConfidenceFields(Protocol): @property def samples(self) -> int: ... @property - def error_count(self) -> int: ... + def error_count(self) -> float: ... @property def attempt_count(self) -> int: ... @property @@ -90,9 +95,15 @@ def accuracy_of(key_stats: KeyStats) -> float: return _accuracy(key_stats) +def _accuracy_score(stats: HasConfidenceFields) -> float: + """Accuracy against `ACCURACY_TARGET`, capped at 1.0: the accuracy term of + skill/confidence. Raw accuracy (`accuracy_of`) stays the display value.""" + return min(1.0, _accuracy(stats) / ACCURACY_TARGET) + + def _effective_attempt_count( samples: int, - error_count: int, + error_count: float, attempt_count: int, *, mean_time_ns: float = 0.0, @@ -106,7 +117,7 @@ def _effective_attempt_count( return attempt_count inferred = samples + error_count if inferred > 0: - return inferred + return max(1, round(inferred)) return _inferred_attempt(mean_time_ns) @@ -153,21 +164,22 @@ def confidence_from_stats( min_attempts: int, ) -> float: """Shared body for `confidence_of`/`transition_confidence_of`: min(speed, - accuracy) so fast-but-sloppy or slow-but-accurate cannot read as mastered, - scaled down until `min_attempts` so a lucky first session can't read as - mastered (see docs/research/typing-pedagogy.md). 0.0 when never practiced. + accuracy score) so fast-but-sloppy or slow-but-accurate cannot read as + mastered, scaled down until `min_attempts` so a lucky first session can't + read as mastered (see docs/research/typing-pedagogy.md). 0.0 when never + practiced. """ if stats is None: return 0.0 - raw = min(key_confidence(target, stats.mean_time_ns), _accuracy(stats)) + raw = min(key_confidence(target, stats.mean_time_ns), _accuracy_score(stats)) return round_confidence(raw * confidence_sample_factor(_attempts(stats), minimum=min_attempts)) def skill_from_stats(stats: HasConfidenceFields | None, target: float) -> float: - """Performance skill without attempt ramp — for display only.""" + """Performance skill without attempt ramp: min(speed, accuracy score).""" if stats is None: return 0.0 - raw = min(key_confidence(target, stats.mean_time_ns), _accuracy(stats)) + raw = min(key_confidence(target, stats.mean_time_ns), _accuracy_score(stats)) return round_confidence(raw) @@ -189,6 +201,18 @@ def skill_of(codepoint: int, stats: Mapping[int, KeyStats], target: float) -> fl return skill_from_stats(stats.get(codepoint), target) +# ponytail: fixed multiplier; upgrade to a Settings field if a stuck key or +# pair turns out to need per-user tuning. +STALL_ATTEMPTS_MULTIPLIER = 3 + + +def stall_attempts_cap(min_attempts: int) -> int: + """Real (unweighted) window attempts after which a key or pair that still + has not cleared counts as stalled: the transition gate releases it + (`domain.unlock`) and focus stops holding on to it (`domain.focus`).""" + return min_attempts * STALL_ATTEMPTS_MULTIPLIER + + def clears_threshold( stats: HasConfidenceFields | None, target: float, diff --git a/src/keystrike/domain/focus.py b/src/keystrike/domain/focus.py index f5a1698..0b9fcaf 100644 --- a/src/keystrike/domain/focus.py +++ b/src/keystrike/domain/focus.py @@ -4,10 +4,11 @@ from __future__ import annotations -from collections.abc import Mapping, Sequence +from collections.abc import Callable, Mapping, Sequence from dataclasses import dataclass from .confidence import ( + CONFIDENCE_GOOD, MIN_CONFIDENCE_ATTEMPTS, MIN_TRANSITION_CONFIDENCE_ATTEMPTS, HasConfidenceFields, @@ -15,7 +16,7 @@ clears_threshold, confidence_from_stats, review_urgency, - skill_from_stats, + stall_attempts_cap, ) from .enums import FocusKind from .models import ( @@ -25,7 +26,7 @@ KeyStats, TransitionStats, ) -from .newest_key import newest_practiced_key_pairs, unlocked_cross_key_pairs +from .newest_key import unlocked_cross_key_pairs # Re-exported for domain.generator, which imports these boost defaults from # this module rather than `domain.models` directly. @@ -58,246 +59,101 @@ def __post_init__(self) -> None: raise ValueError(f"{self.kind} must not have a pair") -def _focus_score_from_entry( - stats: HasConfidenceFields | None, - target: float, - now: float, - *, - review_penalty: float, - min_attempts: int, -) -> float: - """Shared body for `_focus_score`/`_transition_focus_score`: confidence, - discounted further the longer it's been since last practiced, so a stale - mastered key/pair can still outrank a merely-weak recent one.""" - urgency = review_urgency(stats.last_seen if stats else 0.0, now) - confidence = confidence_from_stats(stats, target, min_attempts=min_attempts) - return confidence * (1.0 - review_penalty * urgency) +def _cleared(stats: HasConfidenceFields | None, target: float, min_attempts: int) -> bool: + return clears_threshold(stats, target, threshold=CONFIDENCE_GOOD, min_attempts=min_attempts) -def _focus_score( - codepoint: int, - stats: Mapping[int, KeyStats], - target: float, - now: float, - *, - review_penalty: float, - min_attempts: int = MIN_CONFIDENCE_ATTEMPTS, -) -> float: - return _focus_score_from_entry( - stats.get(codepoint), - target, - now, - review_penalty=review_penalty, - min_attempts=min_attempts, - ) +def _stalled(stats: HasConfidenceFields | None, min_attempts: int) -> bool: + """Enough presses that holding focus longer is not helping (plateau): + the same cap the transition gate uses to stop waiting on a pair.""" + return stats is not None and attempts_of(stats) >= stall_attempts_cap(min_attempts) -def blocks_transition_focus( - unlocked: Sequence[int], - stats: Mapping[int, KeyStats], - target: float, - *, - threshold: float = 1.0, - min_attempts: int = MIN_CONFIDENCE_ATTEMPTS, -) -> bool: - """True when any unlocked key lacks required performance or evidence.""" - return any( - not clears_threshold(stats.get(cp), target, threshold=threshold, min_attempts=min_attempts) - for cp in unlocked - ) +def _pick[T]( + candidates: Sequence[T], + last: T | None, + stats_of: Callable[[T], HasConfidenceFields | None], + min_attempts: int, + rank: Callable[[T], float], +) -> T: + """The last lesson's focus while it is still a candidate and not stalled, + else the lowest-ranked candidate.""" + if last is not None and last in candidates and not _stalled(stats_of(last), min_attempts): + return last + return min(candidates, key=rank) -def select_focus( +def select_lesson_focus( unlocked: Sequence[int], - stats: Mapping[int, KeyStats], + key_stats: Mapping[int, KeyStats], + transitions: Mapping[Bigram, TransitionStats], target: float, now: float, *, - review_penalty: float = 0.5, - threshold: float = 1.0, + pair_frequency: Callable[[Bigram], float], + gate: Sequence[Bigram] = (), + last_key: int | None = None, + last_pair: Bigram | None = None, min_attempts: int = MIN_CONFIDENCE_ATTEMPTS, -) -> int: - """The unlocked key a lesson should emphasize. - - Any key that hasn't yet cleared both the skill threshold and the attempt - floor (`clears_threshold`) wins over every key that has — so focus stays - on an in-progress key across lesson builds instead of a mastered-but- - stale key's review-urgency discount stealing it away mid-calibration. - Only once every unlocked key has cleared does review-urgency-based - staleness compete for focus among the whole set (SlimStampen-style - spacing; see typing-pedagogy.md).""" - candidates = [ - cp - for cp in unlocked - if not clears_threshold( - stats.get(cp), target, threshold=threshold, min_attempts=min_attempts + min_transition_attempts: int = MIN_TRANSITION_CONFIDENCE_ATTEMPTS, +) -> tuple[int, Bigram | None]: + """Today's focus as (key, pair); pair is None for key focus, else the key + is the pair's second letter. The first rule with candidates wins, and each + rule keeps the last lesson's focus while it is still in that rule's list + and not stalled (see docs/wiki/Focus-States.md): + + 1. Keys not cleared (skill and attempt floor) -- weakest key. + 2. The newest key's `gate` pairs that block the next unlock -- weakest pair. + 3. Pairs not cleared that occur in the language -- highest + (1 - confidence) x `pair_frequency`. + 4. Cleared keys and pairs due for review -- most overdue. + 5. Otherwise the weakest key. + """ + + def key_conf(cp: int) -> float: + return confidence_from_stats(key_stats.get(cp), target, min_attempts=min_attempts) + + def pair_conf(pair: Bigram) -> float: + return confidence_from_stats( + transitions.get(pair), target, min_attempts=min_transition_attempts ) - ] - pool = candidates or unlocked - return min( - pool, - key=lambda cp: _focus_score( - cp, - stats, - target, - now, - review_penalty=review_penalty, - min_attempts=min_attempts, - ), - ) - - -def _transition_focus_score( - prev_cp: int, - next_cp: int, - stats: Mapping[Bigram, TransitionStats], - target: float, - now: float, - *, - review_penalty: float, - min_attempts: int = MIN_TRANSITION_CONFIDENCE_ATTEMPTS, -) -> float: - return _focus_score_from_entry( - stats.get(Bigram(prev_cp, next_cp)), - target, - now, - review_penalty=review_penalty, - min_attempts=min_attempts, - ) + weak_keys = [cp for cp in unlocked if not _cleared(key_stats.get(cp), target, min_attempts)] + if weak_keys: + return _pick(weak_keys, last_key, key_stats.get, min_attempts, key_conf), None -def newest_key_unmeasured_pairs( - unlocked: Sequence[int], - transitions: Mapping[Bigram, TransitionStats], - key_stats: Mapping[int, KeyStats] | None, -) -> list[Bigram]: - """Cross-key pairs between the most-recently-*practiced* unlocked key and - its other already-practiced unlocked peers, when none of those pairs - have measured transition data yet. - - Built on `domain.newest_key.newest_practiced_key_pairs`, the single - source of truth shared with `domain.unlock.newest_key_clears_transition_gate`. - Returns empty once the newest key has *any* measured transition - (ordinary weakest-pair scoring takes over from there for focus - selection), when no unlocked key has been practiced yet, or when it has - no other practiced key to pair with.""" - unmeasured, measured = newest_practiced_key_pairs(unlocked, transitions, key_stats) - return [] if measured else unmeasured - - -def select_focus_transition( - unlocked: Sequence[int], - transitions: Mapping[Bigram, TransitionStats], - target: float, - now: float, - *, - key_stats: Mapping[int, KeyStats] | None = None, - gating_candidates: Sequence[Bigram] | None = None, - review_penalty: float = 0.5, - threshold: float = 1.0, - min_attempts: int = MIN_TRANSITION_CONFIDENCE_ATTEMPTS, -) -> Bigram | None: - """Weakest unlocked bigram by transition confidence. - - Measured pairs are preferred. But when the newest practiced key has no - measured transitions of its own yet (`newest_key_unmeasured_pairs`), fall - back to its weakest-scoring unmeasured pair — so a freshly-opened letter - gets bigram focus right away instead of waiting for its pairs to appear - by chance. - - Among measured pairs (outside `gating_candidates` mode), a pair that - hasn't cleared both the skill threshold and attempt floor - (`clears_threshold`) always wins over an already-cleared, merely-stale - pair — the same stickiness `select_focus` applies to keys, so an - in-progress bigram isn't preempted by review urgency before it's done - calibrating. Once every measured pair has cleared, review-urgency-based - staleness competes for focus among all of them, same as before.""" - - def _score(pair: Bigram) -> float: - return _transition_focus_score( - pair.prev_cp, - pair.next_cp, - transitions, - target, - now, - review_penalty=review_penalty, - min_attempts=min_attempts, - ) + if gate: + pair = _pick(gate, last_pair, transitions.get, min_transition_attempts, pair_conf) + return pair.next_cp, pair - measured = [pair for pair in unlocked_cross_key_pairs(unlocked) if pair in transitions] - if gating_candidates is not None: - gating_set = frozenset(gating_candidates) - regressions = [ - pair - for pair in measured - if pair not in gating_set - and attempts_of(transitions[pair]) >= min_attempts - and skill_from_stats(transitions[pair], target) < 1.0 - ] - if regressions: - return min(regressions, key=_score) - return min(gating_candidates, key=_score) if gating_candidates else None - - candidates = newest_key_unmeasured_pairs(unlocked, transitions, key_stats) - if candidates: - return min(candidates, key=_score) - if not measured: - return None - not_cleared = [ + weak_pairs = [ pair - for pair in measured - if not clears_threshold( - transitions.get(pair), target, threshold=threshold, min_attempts=min_attempts - ) + for pair in unlocked_cross_key_pairs(unlocked) + if pair_frequency(pair) > 0 + and not _cleared(transitions.get(pair), target, min_transition_attempts) ] - return min(not_cleared or measured, key=_score) - + if weak_pairs: + pair = _pick( + weak_pairs, + last_pair, + transitions.get, + min_transition_attempts, + lambda p: -(1.0 - pair_conf(p)) * pair_frequency(p), + ) + return pair.next_cp, pair -def focus_key_from_transition(_prev_cp: int, next_cp: int) -> int: - """Endpoint key to mark as lesson focus for a transition-driven bigram.""" - return next_cp + def urgency(stats: HasConfidenceFields | None) -> float: + return review_urgency(stats.last_seen if stats else 0.0, now) + due = [(urgency(key_stats.get(cp)), cp, None) for cp in unlocked] + [ + (urgency(transitions.get(pair)), pair.next_cp, pair) + for pair in unlocked_cross_key_pairs(unlocked) + ] + most_due, key, pair = max(due, key=lambda entry: entry[0]) + if most_due > 0: + return key, pair -def remedial_focus( - lesson_alphabet: Sequence[int], - unlocked: Sequence[int], - stats: Mapping[int, KeyStats], - transitions: Mapping[Bigram, TransitionStats], - target: float, - *, - now: float, - threshold: float = 1.0, - min_attempts: int = MIN_CONFIDENCE_ATTEMPTS, - min_transition_attempts: int = MIN_TRANSITION_CONFIDENCE_ATTEMPTS, -) -> tuple[int, Bigram | None] | None: - """Weakest key or bigram confined to one already-finished lesson's own - alphabet — pulls focus back onto that lesson's own weak point when its - WPM fell short of target, instead of ordinary weakest-across-window - selection (or the newest-key transition gate) drifting focus elsewhere. - - None once every key in the pool has cleared and no bigram among them is - left wanting either — callers should fall back to ordinary focus - selection at that point (the lesson's weak point is resolved).""" - pool = [cp for cp in unlocked if cp in frozenset(lesson_alphabet)] - if not pool: - return None - if blocks_transition_focus(pool, stats, target, threshold=threshold, min_attempts=min_attempts): - focus = select_focus( - pool, stats, target, now, threshold=threshold, min_attempts=min_attempts - ) - return focus, None - bigram = select_focus_transition( - pool, - transitions, - target, - now, - key_stats=stats, - threshold=threshold, - min_attempts=min_transition_attempts, - ) - if bigram is None: - return None - return focus_key_from_transition(*bigram), bigram + return min(unlocked, key=key_conf), None def coverage_deficit_factor( diff --git a/src/keystrike/domain/markov.py b/src/keystrike/domain/markov.py index 73f1395..fd2d10a 100644 --- a/src/keystrike/domain/markov.py +++ b/src/keystrike/domain/markov.py @@ -37,6 +37,13 @@ def __post_init__(self) -> None: # dict field too so in-place mutation of its contents also raises. object.__setattr__(self, "transitions", MappingProxyType(dict(self.transitions))) + def pair_weight(self, pair: Bigram) -> int: + """Language weight of `pair` -- the order-1 row for its first letter, + else the global "" row (same back-off as `sample`). 0 when the pair + never occurs.""" + row = self.transitions.get(chr(pair.prev_cp)) or self.transitions.get("", {}) + return row.get(chr(pair.next_cp), 0) + def sample( self, context: str, diff --git a/src/keystrike/domain/models.py b/src/keystrike/domain/models.py index c143fc8..29b27f9 100644 --- a/src/keystrike/domain/models.py +++ b/src/keystrike/domain/models.py @@ -111,6 +111,10 @@ class SessionResult: # Per-key/bigram tallies (schema 5+). Empty for sessions older than the # stats retention window (see domain.retention) — their history row stays. stats: SessionStats = field(default_factory=SessionStats) + # The pair this lesson emphasized, for pair focus (`focus_key` is then its + # second letter); the next lesson keeps it while it still needs work. + # Written only when set, so schema 5 rows without it read back unchanged. + focus_pair: Bigram | None = None def __post_init__(self) -> None: # Freezing the dataclass only blocks attribute rebinding — wrap the @@ -126,8 +130,10 @@ class KeyStats: codepoint: int samples: int mean_time_ns: float - error_count: int + # Recency-weighted in windowed aggregates, so it can be fractional. + error_count: float last_seen: float + # Real presses in the window (not recency-weighted): the attempt floor. attempt_count: int = 0 @@ -137,9 +143,9 @@ class TransitionStats: next_cp: int samples: int mean_time_ns: float - error_count: int + error_count: float # see KeyStats last_seen: float - attempt_count: int = 0 + attempt_count: int = 0 # see KeyStats def _empty_transitions() -> dict[Bigram, TransitionStats]: diff --git a/src/keystrike/domain/newest_key.py b/src/keystrike/domain/newest_key.py index 4a09526..b9d2493 100644 --- a/src/keystrike/domain/newest_key.py +++ b/src/keystrike/domain/newest_key.py @@ -12,7 +12,7 @@ from collections.abc import Mapping, Sequence from .confidence import is_same_key_transition -from .models import GATING_BIGRAM_LIMIT, Bigram, KeyStats, TransitionStats +from .models import GATING_BIGRAM_LIMIT, Bigram, KeyStats _GATING_BIGRAM_MIN = 2 _GATING_BIGRAM_MAX = 4 @@ -44,39 +44,6 @@ def unlocked_cross_key_pairs(unlocked: Sequence[int]) -> list[Bigram]: ] -def newest_practiced_key_pairs( - unlocked: Sequence[int], - transitions: Mapping[Bigram, TransitionStats], - key_stats: Mapping[int, KeyStats] | None, -) -> tuple[list[Bigram], list[Bigram]]: - """(unmeasured, measured) cross-key pairs between the most-recently- - *practiced* unlocked key (skipping any further, not-yet-touched key - added by an unlock-threshold cascade) and its other already-practiced - unlocked peers. Both empty when no unlocked key has been practiced yet, - or the newest practiced key has no practiced peer to pair with. - - Single source of truth for "what does the newest key still need bigram - attention on" — shared by `domain.focus.newest_key_unmeasured_pairs` - (focus selection, lesson weighting) and - `domain.unlock.newest_key_clears_transition_gate` (unlock), so the two - can't drift on what counts as newest or which pairs are in play.""" - practiced = _practiced_unlocked(unlocked, key_stats) - if not practiced: - return [], [] - newest = practiced[-1] - practiced_set = frozenset(practiced) - pairs = [ - pair - for pair in unlocked_cross_key_pairs(unlocked) - if newest in (pair.prev_cp, pair.next_cp) - and pair.prev_cp in practiced_set - and pair.next_cp in practiced_set - ] - measured = [pair for pair in pairs if pair in transitions] - unmeasured = [pair for pair in pairs if pair not in transitions] - return unmeasured, measured - - def newest_key_gating_cohort( unlocked: Sequence[int], key_stats: Mapping[int, KeyStats] | None, diff --git a/src/keystrike/domain/session.py b/src/keystrike/domain/session.py index aa99d93..744f35b 100644 --- a/src/keystrike/domain/session.py +++ b/src/keystrike/domain/session.py @@ -10,7 +10,7 @@ from dataclasses import dataclass, field from .enums import Mode, SessionState -from .models import Keystroke +from .models import Bigram, Keystroke BACKSPACE = "\x7f" # normalized backspace codepoint marker used at edges LEADING_SKIP_KEYS = frozenset(" \t\n\r") # accidental space/enter/tab before session/word start @@ -111,6 +111,7 @@ class Session: error_positions: set[int] = field(default_factory=set[int]) # positions that needed correction state: SessionState = SessionState.RUNNING focus_key: int | None = None # adaptive mode: the key this lesson emphasized + focus_pair: Bigram | None = None # adaptive mode: the pair, for pair focus @property def finished(self) -> bool: diff --git a/src/keystrike/domain/unlock.py b/src/keystrike/domain/unlock.py index 0e7260a..95356bf 100644 --- a/src/keystrike/domain/unlock.py +++ b/src/keystrike/domain/unlock.py @@ -6,26 +6,13 @@ from collections.abc import Mapping, Sequence -from .confidence import attempts_of, clears_threshold, transition_confidence_of +from .confidence import attempts_of, clears_threshold from .models import Bigram, KeyStats, TransitionStats, UnlockTuning from .newest_key import newest_key_gating_cohort -# ponytail: fixed multiplier; upgrade to a Settings field if a stuck gating -# bigram turns out to need per-user tuning. -TRANSITION_STALL_ATTEMPTS_MULTIPLIER = 3 - _DEFAULT_UNLOCK_TUNING = UnlockTuning() -def default_transition_stall_attempts_cap(min_attempts: int) -> int: - """Default `transition_stall_attempts_cap` for `compute_unlocked`: give a - stuck gating bigram this many times the normal calibration floor before - releasing it anyway. Single source of truth so callers that need the - gate (`build_lesson`, `session_use_cases`) can't compute this - differently and drift apart.""" - return min_attempts * TRANSITION_STALL_ATTEMPTS_MULTIPLIER - - def newest_key_clears_transition_gate( unlocked: Sequence[int], transitions: Mapping[Bigram, TransitionStats], @@ -56,18 +43,13 @@ def gating_bigram_is_ready( stall_attempts_cap: int | None = None, ) -> bool: stats = transitions.get(pair) - attempts = attempts_of(stats) if stats is not None else 0 - mastered = ( - attempts >= tuning.min_transition_confidence_attempts - and transition_confidence_of( - pair.prev_cp, - pair.next_cp, - transitions, - target, - min_attempts=tuning.min_transition_confidence_attempts, - ) - >= tuning.next_letter_unlock_threshold + mastered = clears_threshold( + stats, + target, + threshold=tuning.next_letter_unlock_threshold, + min_attempts=tuning.min_transition_confidence_attempts, ) + attempts = attempts_of(stats) if stats is not None else 0 return mastered or (stall_attempts_cap is not None and attempts >= stall_attempts_cap) diff --git a/src/keystrike/domain/version.py b/src/keystrike/domain/version.py index 9aa3f90..8a124bf 100644 --- a/src/keystrike/domain/version.py +++ b/src/keystrike/domain/version.py @@ -1 +1 @@ -__version__ = "2.1.0" +__version__ = "2.2.0" diff --git a/src/keystrike/infrastructure/aggregates_cache.py b/src/keystrike/infrastructure/aggregates_cache.py index 3e83de6..91fbb37 100644 --- a/src/keystrike/infrastructure/aggregates_cache.py +++ b/src/keystrike/infrastructure/aggregates_cache.py @@ -18,6 +18,11 @@ from .json_coerce import require_float, require_int from .paths import Paths, sanitize_layout_name +# 2: attempt_count is a plain (unweighted) window count and error_count is +# fractional. Older files hold recency-weighted attempts, so `get` treats them +# as a miss and the aggregates rebuild from the session rows. +_SCHEMA_VERSION = 2 + def _coerce_samples(entry: dict[str, object]) -> int: samples = require_int(entry, "samples") @@ -26,8 +31,8 @@ def _coerce_samples(entry: dict[str, object]) -> int: def _coerce_attempt_count(entry: dict[str, object], samples: int) -> int: - errors = require_int(entry, "error_count") - stored = require_int(entry, "attempt_count", samples + errors) + errors = require_float(entry, "error_count") + stored = require_int(entry, "attempt_count", 0) return infer_key_stat_attempt_count(samples, errors, stored) @@ -41,7 +46,7 @@ def _parse_transition_entry(entry: dict[str, object]) -> tuple[Bigram, Transitio next_cp=require_int(entry, "next_cp"), samples=samples, mean_time_ns=require_float(entry, "mean_time_ns"), - error_count=require_int(entry, "error_count"), + error_count=require_float(entry, "error_count"), last_seen=require_float(entry, "last_seen"), attempt_count=_coerce_attempt_count(entry, samples), ) @@ -53,7 +58,7 @@ def _parse_key_entry(cp: str, entry: dict[str, object]) -> KeyStats: codepoint=int(cp), samples=samples, mean_time_ns=require_float(entry, "mean_time_ns"), - error_count=require_int(entry, "error_count"), + error_count=require_float(entry, "error_count"), last_seen=require_float(entry, "last_seen"), attempt_count=_coerce_attempt_count(entry, samples), ) @@ -73,6 +78,8 @@ def get(self, layout: str) -> LayoutAggregates | None: return None try: data = json.loads(file.read_text(encoding="utf-8")) + if data.get("schema_version") != _SCHEMA_VERSION: + return None keys = data.get("keys", {}) transitions = data.get("transitions", {}) parsed_transitions = dict( @@ -93,7 +100,7 @@ def get(self, layout: str) -> LayoutAggregates | None: def put(self, layout: str, aggregates: LayoutAggregates) -> None: transitions = without_same_key_transitions(aggregates.transitions) payload = { - "schema_version": 1, + "schema_version": _SCHEMA_VERSION, "layout": layout, "keys": {str(cp): asdict(k) for cp, k in aggregates.keys.items()}, "transitions": {key.chars(): asdict(t) for key, t in transitions.items()}, diff --git a/src/keystrike/infrastructure/session_repo_jsonl.py b/src/keystrike/infrastructure/session_repo_jsonl.py index 94f88d3..e2c8af7 100644 --- a/src/keystrike/infrastructure/session_repo_jsonl.py +++ b/src/keystrike/infrastructure/session_repo_jsonl.py @@ -29,7 +29,9 @@ # `[samples, time_ns, errors, attempts]`, keyed by codepoint for keys and by # "prev,next" codepoints for bigrams. Omitted entirely when the stats are empty. _STATS_KEY = "stats" +_FOCUS_PAIR_KEY = "focus_pair" _TALLY_FIELDS = 4 +_PAIR_FIELDS = 2 class JsonlSessionRepository: @@ -138,6 +140,10 @@ def header_to_row(h: SessionResult) -> dict[str, object]: del base[_STATS_KEY] else: base[_STATS_KEY] = stats_to_row(h.stats) + if h.focus_pair is None: + del base[_FOCUS_PAIR_KEY] + else: + base[_FOCUS_PAIR_KEY] = [h.focus_pair.prev_cp, h.focus_pair.next_cp] return base @@ -167,6 +173,15 @@ def _parse_key_confidence(raw: object) -> dict[int, float]: return out +def _parse_focus_pair(raw: object) -> Bigram | None: + if raw is None: + return None + if not isinstance(raw, (list, tuple)) or len(cast("list[object]", raw)) != _PAIR_FIELDS: + raise TypeError(f"expected [prev, next] for {_FOCUS_PAIR_KEY!r}, got {raw!r}") + prev, nxt = (coerce_int(v, label=_FOCUS_PAIR_KEY) for v in cast("list[object]", raw)) + return Bigram(prev, nxt) + + def header_from_row(d: dict[str, object]) -> SessionResult: session_id = require_str(d, "session_id") validate_session_id(session_id) # Reject path-traversal attempts @@ -189,4 +204,5 @@ def header_from_row(d: dict[str, object]) -> SessionResult: generated_min_len=require_int(d, "generated_min_len", GENERATED_WORD_MIN_LEN), generated_max_len=require_int(d, "generated_max_len", GENERATED_WORD_MAX_LEN), stats=stats_from_row(d.get(_STATS_KEY)), + focus_pair=_parse_focus_pair(d.get(_FOCUS_PAIR_KEY)), ) diff --git a/src/keystrike/presentation/screens/practice.py b/src/keystrike/presentation/screens/practice.py index aabfe5d..401021d 100644 --- a/src/keystrike/presentation/screens/practice.py +++ b/src/keystrike/presentation/screens/practice.py @@ -66,6 +66,7 @@ def __init__( layout=initial.layout, mode=initial.mode, focus_key=initial.focus_key, + focus_pair=initial.focus_pair, ) self._typing_area = TypingArea(self._session) self._hud = HUD( @@ -181,6 +182,7 @@ def _begin_session(self, prep: SessionPrep) -> None: layout=prep.layout, mode=prep.mode, focus_key=prep.focus_key, + focus_pair=prep.focus_pair, ) self._typing_area.set_session(self._session) self._hud.set_session( diff --git a/tests/application/test_build_lesson.py b/tests/application/test_build_lesson.py index 60586db..6c618c2 100644 --- a/tests/application/test_build_lesson.py +++ b/tests/application/test_build_lesson.py @@ -10,7 +10,7 @@ from keystrike.domain.aggregate import _combine_transition_maps_weighted, session_recency_weights from keystrike.domain.confidence import target_ms_per_char, transition_confidence_of from keystrike.domain.enums import FocusKind, Mode -from keystrike.domain.focus import FocusReason, select_focus +from keystrike.domain.focus import FocusReason from keystrike.domain.generator import weak_focus_word_quota, word_matches_focus from keystrike.domain.learn_order import keyboard_order from keystrike.domain.models import ( @@ -662,7 +662,7 @@ def test_lesson_uses_transition_review_when_stale_and_mastered(): five_days = 5 * 86_400.0 at_target = 200_000_000.0 # s→h stays weak; h still calibrating so unlock stops at a,s,h. Stale a→s - # wins transition focus via review scoring (see test_select_focus_transition_*). + # wins transition focus via review (see tests/domain/test_focus.py). slow = 380_000_000.0 keys = { a: KeyStats(a, 10, at_target, 0, now, attempt_count=10), @@ -1152,167 +1152,44 @@ def test_focus_stays_on_calibrating_key_despite_stale_mastered_peer(): assert builder("qwerty").focus_key != a -def test_remedial_focus_targets_weak_key_from_low_wpm_lesson_alphabet(): - """A lesson that finished under its own WPM target pulls the next - lesson's focus back onto that lesson's own weak point, even though a - different key is weaker across the whole rolling window.""" - layout = BUNDLED_LAYOUTS["qwerty"] - order = keyboard_order(layout) - a, s, h, d = order[0], order[1], order[2], order[3] - now = 1_700_000_000.0 - at_target = 200_000_000.0 - keys = { - a: KeyStats(a, 10, at_target, 0, now, attempt_count=10), # mastered - s: KeyStats(s, 10, at_target, 0, now, attempt_count=10), # mastered - h: KeyStats(h, 10, at_target / 0.2, 0, now, attempt_count=10), # globally weakest - d: KeyStats(d, 10, at_target / 0.6, 0, now, attempt_count=10), # weak, but not weakest - } - target = target_ms_per_char(300) - # Sanity: plain weakest-across-window selection would land on h, not d. - assert select_focus((a, s, h, d), keys, target, now) == h - - cache = FakeAggregatesCache( - by_layout={"qwerty": LayoutAggregates(keys=keys, transitions={})}, - ) - slow_session = SessionResult( - schema_version=3, - session_id="slow", - started_at=1.0, - duration_ns=60_000_000_000, - layout="qwerty", - mode=Mode.ADAPTIVE, - lesson_alphabet=(a, d), - focus_key=d, - total_keystrokes=1, - correct_keystrokes=1, - words_completed=1, - target_speed_cpm=300, - ) - session_repo = FakeSessionRepository(headers=[slow_session]) - builder = BuildLesson( - layout_repo=FakeLayoutRepository(dict(BUNDLED_LAYOUTS)), - aggregates_cache=cache, - settings_repo=FakeSettingsRepository(Settings(alphabet_size=4, target_speed_cpm=300)), - language_provider=FakeLanguageProvider(), - wordlist_store=FakeWordListStore(), - rng=Random(0), - clock=FakeClock(), - session_repo=session_repo, - ) - assert builder("qwerty").focus_key == d - - -def test_remedial_focus_uses_session_word_bounds_not_current_settings(): - """Changing word-length bounds after a session finishes must not alter - whether that session triggers remedial focus on the next lesson.""" - layout = BUNDLED_LAYOUTS["qwerty"] - order = keyboard_order(layout) - a, s, h, d = order[0], order[1], order[2], order[3] - now = 1_700_000_000.0 - at_target = 200_000_000.0 +def test_focus_keeps_the_last_session_focus_while_it_needs_work(): + order = keyboard_order(BUNDLED_LAYOUTS["qwerty"]) + a, s = order[0], order[1] keys = { - a: KeyStats(a, 10, at_target, 0, now, attempt_count=10), - s: KeyStats(s, 10, at_target, 0, now, attempt_count=10), - h: KeyStats(h, 10, at_target / 0.2, 0, now, attempt_count=10), - d: KeyStats(d, 10, at_target / 0.6, 0, now, attempt_count=10), + a: KeyStats(a, 10, 400_000_000.0, 0, 0.0, attempt_count=10), # weakest + s: KeyStats(s, 10, 300_000_000.0, 0, 0.0, attempt_count=10), } - cache = FakeAggregatesCache( - by_layout={"qwerty": LayoutAggregates(keys=keys, transitions={})}, - ) - # 50 wpm: below 100 wpm target at snapshotted 2-4 bounds, above 46 at 3-10. - slow_session = SessionResult( - schema_version=4, - session_id="slow", + last = SessionResult( + schema_version=5, + session_id="s1", started_at=1.0, - duration_ns=60_000_000_000, + duration_ns=1, layout="qwerty", mode=Mode.ADAPTIVE, - lesson_alphabet=(a, d), - focus_key=d, + lesson_alphabet=(a, s), + focus_key=s, total_keystrokes=1, correct_keystrokes=1, - words_completed=50, - target_speed_cpm=300, - generated_min_len=2, - generated_max_len=4, - ) - session_repo = FakeSessionRepository(headers=[slow_session]) - builder = BuildLesson( - layout_repo=FakeLayoutRepository(dict(BUNDLED_LAYOUTS)), - aggregates_cache=cache, - settings_repo=FakeSettingsRepository( - Settings( - alphabet_size=4, - target_speed_cpm=300, - word_gen=WordGenBounds(min_len=3, max_len=10), - ) - ), - language_provider=FakeLanguageProvider(), - wordlist_store=FakeWordListStore(), - rng=Random(0), - clock=FakeClock(), - session_repo=session_repo, ) - assert builder("qwerty").focus_key == d - -def test_remedial_focus_clears_once_lesson_wpm_meets_target(): - """Once the most recent session's own WPM meets its own target again, - normal weakest-across-window focus selection resumes.""" - layout = BUNDLED_LAYOUTS["qwerty"] - order = keyboard_order(layout) - a, s, h, d = order[0], order[1], order[2], order[3] - now = 1_700_000_000.0 - at_target = 200_000_000.0 - keys = { - a: KeyStats(a, 10, at_target, 0, now, attempt_count=10), - s: KeyStats(s, 10, at_target, 0, now, attempt_count=10), - h: KeyStats(h, 10, at_target / 0.2, 0, now, attempt_count=10), - d: KeyStats(d, 10, at_target / 0.6, 0, now, attempt_count=10), - } - cache = FakeAggregatesCache( - by_layout={"qwerty": LayoutAggregates(keys=keys, transitions={})}, - ) - slow_session = SessionResult( - schema_version=3, - session_id="slow", - started_at=1.0, - duration_ns=60_000_000_000, - layout="qwerty", - mode=Mode.ADAPTIVE, - lesson_alphabet=(a, d), - focus_key=d, - total_keystrokes=1, - correct_keystrokes=1, - words_completed=1, - target_speed_cpm=300, - ) - fast_session = SessionResult( - schema_version=3, - session_id="fast", - started_at=2.0, - duration_ns=1_000_000_000, - layout="qwerty", - mode=Mode.ADAPTIVE, - lesson_alphabet=(a, d), - focus_key=d, - total_keystrokes=1, - correct_keystrokes=1, - words_completed=10, - target_speed_cpm=300, - ) - session_repo = FakeSessionRepository(headers=[slow_session, fast_session]) - builder = BuildLesson( - layout_repo=FakeLayoutRepository(dict(BUNDLED_LAYOUTS)), - aggregates_cache=cache, - settings_repo=FakeSettingsRepository(Settings(alphabet_size=4, target_speed_cpm=300)), - language_provider=FakeLanguageProvider(), - wordlist_store=FakeWordListStore(), - rng=Random(0), - clock=FakeClock(), - session_repo=session_repo, - ) - assert builder("qwerty").focus_key == h + def build(headers: list[SessionResult]): + return BuildLesson( + layout_repo=FakeLayoutRepository(dict(BUNDLED_LAYOUTS)), + aggregates_cache=FakeAggregatesCache( + by_layout={"qwerty": LayoutAggregates(keys=keys, transitions={})}, + ), + settings_repo=FakeSettingsRepository(Settings(alphabet_size=2)), + language_provider=FakeLanguageProvider(), + wordlist_store=FakeWordListStore(), + rng=Random(0), + clock=FakeClock(), + session_repo=FakeSessionRepository(headers=headers), + )("qwerty") + + assert build([]).focus_key == a + lesson = build([last]) + assert lesson.focus_key == s + assert lesson.focus_pair is None def test_large_alphabet_unlock_advances_when_keys_rotated(): diff --git a/tests/application/test_session_use_cases.py b/tests/application/test_session_use_cases.py index 5c946d6..910e10d 100644 --- a/tests/application/test_session_use_cases.py +++ b/tests/application/test_session_use_cases.py @@ -10,7 +10,6 @@ compute_wpm, latest_session_header, previous_session_header, - session_wpm_below_target, ) from keystrike.application.session_use_cases import ( AbortSession, @@ -28,6 +27,7 @@ from keystrike.domain.learn_order import keyboard_order from keystrike.domain.models import ( CONFIDENCE_SESSION_WINDOW, + Bigram, SessionResult, Settings, UnlockTuning, @@ -437,54 +437,6 @@ def test_latest_session_header_none_when_no_sessions(): assert latest_session_header(FakeSessionRepository(), "qwerty") is None -def test_session_wpm_below_target_true_when_slower_than_target(clock, id_gen): - _, result = _drive("hello world", "hello ", clock, id_gen) - result = replace(result, words_completed=1, duration_ns=60_000_000_000, target_speed_cpm=300) - assert session_wpm_below_target(result) is True - - -def test_session_wpm_below_target_false_when_faster_than_target(clock, id_gen): - _, result = _drive("hello world", "hello ", clock, id_gen) - result = replace(result, words_completed=20, duration_ns=1_000_000_000, target_speed_cpm=300) - assert session_wpm_below_target(result) is False - - -def test_session_wpm_below_target_false_for_legacy_session_without_target(): - result = SessionResult( - session_id="s1", - total_keystrokes=1, - correct_keystrokes=1, - **_session_stats_common(), - ) - assert result.target_speed_cpm == 0 - assert session_wpm_below_target(result) is False - - -def test_session_wpm_below_target_uses_snapshotted_word_bounds(): - """WPM target conversion must use bounds recorded at finish, not caller overrides.""" - result = replace( - SessionResult( - session_id="s1", total_keystrokes=1, correct_keystrokes=1, **_session_stats_common() - ), - words_completed=50, - duration_ns=60_000_000_000, - target_speed_cpm=300, - generated_min_len=2, - generated_max_len=4, - ) - # 50 wpm vs target 100 wpm (300 cpm / 3 chars per word at 2-4 bounds) - assert session_wpm_below_target(result) is True - # Wider bounds would lower target to ~46 wpm and incorrectly clear remedial. - assert ( - session_wpm_below_target( - result, - generated_min_len=3, - generated_max_len=10, - ) - is False - ) - - def test_finish_session_persists_generated_word_bounds(clock, id_gen): settings_repo = FakeSettingsRepository(Settings(word_gen=WordGenBounds(min_len=3, max_len=8))) finish = FinishSession(clock=clock, settings_repo=settings_repo) @@ -766,3 +718,20 @@ def test_finish_session_alphabet_bump_respects_transition_gate(clock, id_gen): unlocked_cps = {k.codepoint for k in lesson.state.keys} assert unlocked_cps == set(order[:4]) assert order[4] not in unlocked_cps + + +def test_finish_session_saves_focus_pair(clock, id_gen): + repo = FakeSessionRepository() + finish = FinishSession(clock=clock, repo=repo) + pair = Bigram(ord("a"), ord("s")) + session = StartSession(clock=clock, id_gen=id_gen)( + "as", layout="qwerty", mode=Mode.ADAPTIVE, focus_key=pair.next_cp, focus_pair=pair + ) + record = RecordKeystroke(clock=clock) + for ch in "as": + clock.advance(100_000_000) + record(session, ch) + result = finish(session) + assert result.focus_key == pair.next_cp + assert result.focus_pair == pair + assert repo.headers[0].focus_pair == pair diff --git a/tests/application/test_stats_use_cases.py b/tests/application/test_stats_use_cases.py index b1d8083..2a8cd3f 100644 --- a/tests/application/test_stats_use_cases.py +++ b/tests/application/test_stats_use_cases.py @@ -6,8 +6,7 @@ GetOrRebuildAggregates, RebuildAggregates, ) -from keystrike.domain.aggregate import session_recency_weights -from keystrike.domain.confidence import MIN_CONFIDENCE_ATTEMPTS, SESSION_RECENCY_DECAY +from keystrike.domain.confidence import MIN_CONFIDENCE_ATTEMPTS from keystrike.domain.enums import Mode from keystrike.domain.models import ( CONFIDENCE_SESSION_WINDOW, @@ -267,8 +266,8 @@ def test_rebuild_aggregates_drops_sessions_outside_window(): assert ord("a") in result assert ord("z") not in result - weights = session_recency_weights(CONFIDENCE_SESSION_WINDOW, decay=SESSION_RECENCY_DECAY) - assert result[ord("a")].attempt_count == round(2 * sum(weights)) + # Attempts are real presses in the window, not recency-weighted. + assert result[ord("a")].attempt_count == 2 * CONFIDENCE_SESSION_WINDOW def test_rebuild_aggregates_respects_settings_window(): @@ -300,8 +299,7 @@ def test_rebuild_aggregates_respects_settings_window(): result = cached.keys assert ord("z") not in result - weights = session_recency_weights(window, decay=SESSION_RECENCY_DECAY) - assert result[ord("a")].attempt_count == round(2 * sum(weights)) + assert result[ord("a")].attempt_count == 2 * window def test_get_heatmap_empty_cache_returns_empty_view(): diff --git a/tests/domain/test_aggregate.py b/tests/domain/test_aggregate.py index 25008c3..737adaf 100644 --- a/tests/domain/test_aggregate.py +++ b/tests/domain/test_aggregate.py @@ -1,3 +1,7 @@ +from dataclasses import replace + +import pytest + from keystrike.domain.aggregate import ( _combine_key_maps_weighted, _combine_transition_maps_weighted, @@ -11,6 +15,7 @@ ) from keystrike.domain.confidence import ( SESSION_RECENCY_DECAY, + clears_threshold, confidence_of, transition_accuracy_of, transition_confidence_of, @@ -152,7 +157,9 @@ def test_combine_sessions_favors_recent_session(): assert out.keys[ord("a")].mean_time_ns > 200_000_000 -def test_combine_sessions_weights_recent_attempts_more(): +def test_combine_sessions_counts_attempts_unweighted(): + """Attempts are evidence for the floors, so they are real presses in the + window — only speed and accuracy carry recency weights.""" s1 = _session("s1", started_at=1.0) s2 = _session("s2", started_at=2.0) keys1 = [ @@ -163,10 +170,57 @@ def test_combine_sessions_weights_recent_attempts_more(): Keystroke(codepoint=ord("a"), typed=ord("a"), t_ns=0, correct=True), ] out = combine_sessions([(s1, tally_session(keys1)), (s2, tally_session(keys2))]) - w_old, w_new = session_recency_weights(2) - expected_attempts = round(2 * w_old + 1 * w_new) - assert out.keys[ord("a")].attempt_count == expected_attempts - assert out.keys[ord("a")].attempt_count < 3 + assert out.keys[ord("a")].attempt_count == 3 + + +def _lesson_of_pairs(keys: str, ms: int = 200) -> list[Keystroke]: + return [ + Keystroke(codepoint=ord(ch), typed=ord(ch), t_ns=i * ms * 1_000_000, correct=True) + for i, ch in enumerate(keys) + ] + + +def test_combine_sessions_attempts_reach_floor_at_two_presses_per_lesson(): + """Default window (10) and decay (0.7): a pair typed twice per lesson has + 20 real attempts, not the ~6.5 a recency-weighted count settles at.""" + sessions = [ + (_session(f"s{i}", started_at=float(i)), tally_session(_lesson_of_pairs("abab"))) + for i in range(10) + ] + ab = combine_sessions(sessions).transitions[Bigram(ord("a"), ord("b"))] + assert ab.attempt_count == 20 + assert 2 * sum(session_recency_weights(10)) < 7 + + +def test_weighted_merge_keeps_one_old_typo_fractional(): + """One typo in the previous lesson (weight 0.7) with about 7 weighted + samples counts as 0.7 of an error, not a full one.""" + ab = Bigram(ord("a"), ord("b")) + old = TransitionStats(ord("a"), ord("b"), 4, 200_000_000.0, 1, 1.0, attempt_count=5) + recent = TransitionStats(ord("a"), ord("b"), 4, 200_000_000.0, 0, 2.0, attempt_count=4) + merged = _combine_transition_maps_weighted( + [{ab: old}, {ab: recent}], session_recency_weights(2) + )[ab] + assert merged.samples == 7 # 0.7 * 4 + 4 = 6.8 + assert merged.error_count == pytest.approx(0.7) + assert merged.attempt_count == 9 + assert transition_accuracy_of(merged) == pytest.approx(7 / 7.7) + + +def test_one_old_typo_does_not_block_an_otherwise_clean_pair(): + """14 weighted samples and one typo last-but-one lesson: 14 / 14.7 is + above the 95% accuracy target. Rounded to a full error (14 / 15) it was not.""" + ab = Bigram(ord("a"), ord("b")) + old = TransitionStats(ord("a"), ord("b"), 10, 200_000_000.0, 1, 1.0, attempt_count=11) + recent = TransitionStats(ord("a"), ord("b"), 7, 200_000_000.0, 0, 2.0, attempt_count=7) + merged = _combine_transition_maps_weighted( + [{ab: old}, {ab: recent}], session_recency_weights(2) + )[ab] + assert merged.samples == 14 + target = 200.0 # 300 CPM; the pair is at exactly target speed + assert clears_threshold(merged, target, threshold=1.0, min_attempts=4) + rounded = replace(merged, error_count=1) + assert not clears_threshold(rounded, target, threshold=1.0, min_attempts=4) def test_combine_sessions_recent_errors_weigh_more_on_confidence(): diff --git a/tests/domain/test_confidence.py b/tests/domain/test_confidence.py index 6f5ba44..c9b375d 100644 --- a/tests/domain/test_confidence.py +++ b/tests/domain/test_confidence.py @@ -1,4 +1,5 @@ from keystrike.domain.confidence import ( + ACCURACY_TARGET, MIN_CONFIDENCE_ATTEMPTS, accuracy_of, confidence_of, @@ -89,8 +90,8 @@ def test_confidence_of_scales_down_with_few_attempts(): attempt_count=2, ), } - # raw min(speed, accuracy) = min(2.0, 0.5) = 0.5; only 2 attempts -> x0.2 - assert confidence_of(ord("a"), stats, target=200.0) == 0.1 + # raw min(speed, accuracy score) = min(2.0, 0.5 / 0.95); only 2 attempts -> x0.2 + assert confidence_of(ord("a"), stats, target=200.0) == 0.11 def test_skill_of_ignores_attempt_ramp(): @@ -202,9 +203,9 @@ def test_transition_confidence_infers_attempts_from_mean_when_all_counts_zeroed( def test_confidence_of_penalizes_frequent_errors(): - # Fast (2.0 speed) but wrong half the time -> min(2.0, 0.5) = 0.5, not mastered. + # Fast (2.0 speed) but wrong half the time -> min(2.0, 0.5 / 0.95), not mastered. stats = {ord("a"): _stats(ord("a"), mean_time_ns=100_000_000.0, error_count=10)} - assert confidence_of(ord("a"), stats, target=200.0) == 0.5 + assert confidence_of(ord("a"), stats, target=200.0) == 0.53 def test_confidence_of_unseen_key_is_zero(): @@ -295,3 +296,14 @@ def _transition( last_seen=last_seen, attempt_count=attempts, ) + + +def test_skill_scores_accuracy_against_target(): + """Accuracy at `ACCURACY_TARGET` (95%) is full marks; below it scales down. + Raw accuracy stays the display value.""" + assert ACCURACY_TARGET == 0.95 + at_target = KeyStats(ord("a"), 19, 200_000_000.0, 1, 0.0, attempt_count=20) + below = KeyStats(ord("a"), 15, 200_000_000.0, 1, 0.0, attempt_count=16) + assert accuracy_of(at_target) == 0.95 + assert skill_of(ord("a"), {ord("a"): at_target}, 200.0) == 1.0 + assert skill_of(ord("a"), {ord("a"): below}, 200.0) == 0.99 # 0.9375 / 0.95 diff --git a/tests/domain/test_focus.py b/tests/domain/test_focus.py index ecbbcae..b43400e 100644 --- a/tests/domain/test_focus.py +++ b/tests/domain/test_focus.py @@ -1,21 +1,7 @@ import pytest -from keystrike.domain.confidence import ( - MIN_CONFIDENCE_ATTEMPTS, - attempts_of, - skill_from_stats, - skill_of, - target_ms_per_char, -) from keystrike.domain.enums import FocusKind -from keystrike.domain.focus import ( - FocusReason, - blocks_transition_focus, - focus_key_from_transition, - remedial_focus, - select_focus, - select_focus_transition, -) +from keystrike.domain.focus import FocusReason, select_lesson_focus from keystrike.domain.models import Bigram, KeyStats, TransitionStats @@ -25,6 +11,7 @@ def _stats( error_count: int = 0, *, last_seen: float = 0.0, + attempt_count: int | None = None, ) -> KeyStats: return KeyStats( codepoint=codepoint, @@ -32,7 +19,7 @@ def _stats( mean_time_ns=mean_time_ns, error_count=error_count, last_seen=last_seen, - attempt_count=10 + error_count, + attempt_count=attempt_count if attempt_count is not None else 10 + error_count, ) @@ -84,416 +71,119 @@ def test_focus_reason_key_kind_is_not_transition(): assert reason.is_transition is False -def test_remedial_focus_picks_weakest_key_in_lesson_alphabet(): - """Weakest key within the remedial pool, not the globally weakest key.""" - now = 1_700_000_000.0 - at_target = 200_000_000.0 - a, s, h, d = ord("a"), ord("s"), ord("h"), ord("d") - stats = { - a: KeyStats(a, 10, at_target, 0, now, attempt_count=10), - s: KeyStats(s, 10, at_target, 0, now, attempt_count=10), - h: KeyStats(h, 10, at_target / 0.2, 0, now, attempt_count=10), - d: KeyStats(d, 10, at_target / 0.6, 0, now, attempt_count=10), - } - target = target_ms_per_char(300) - assert select_focus((a, s, h, d), stats, target, now) == h +TARGET = 200.0 # ms per char +FAST = 100_000_000.0 # ns: skill >= 1.0 +SLOW = 400_000_000.0 # ns: skill 0.5 +NOW = 10 * 86_400.0 +DAY = 86_400.0 +A, S, D = map(ord, "asd") + + +def _focus(keys, transitions=None, *, freq: dict[Bigram, int] | None = None, **kwargs): + weights = freq or {} + return select_lesson_focus( + (A, S, D), + keys, + transitions or {}, + TARGET, + NOW, + pair_frequency=lambda pair: weights.get(pair, 1), + min_attempts=10, + min_transition_attempts=4, + **kwargs, + ) - result = remedial_focus((a, d), (a, s, h, d), stats, {}, target, now=now) - assert result == (d, None) +def _cleared_keys(): + return {cp: _stats(cp, FAST, last_seen=NOW) for cp in (A, S, D)} -def test_remedial_focus_none_when_alphabet_disjoint_from_unlocked(): - now = 1_700_000_000.0 - at_target = 200_000_000.0 - a = ord("a") - stats = {a: KeyStats(a, 10, at_target, 0, now, attempt_count=10)} - target = target_ms_per_char(300) - assert remedial_focus((ord("z"),), (a,), stats, {}, target, now=now) is None +def _cleared_pairs(): + return { + Bigram(p, n): _transition(p, n, FAST, last_seen=NOW) + for p in (A, S, D) + for n in (A, S, D) + if p != n + } -def test_select_focus_picks_weakest_unlocked_key(): - stats = { - 1: _stats(1, mean_time_ns=100_000_000.0), # confidence 2.0 - 2: _stats(2, mean_time_ns=400_000_000.0), # confidence 0.5 - } - assert select_focus((1, 2), stats, target=200.0, now=1000.0) == 2 - - -def test_blocks_transition_focus_true_when_key_below_attempt_floor(): - stats = { - ord("t"): KeyStats( - ord("t"), - samples=9, - mean_time_ns=125_000_000.0, - error_count=0, - last_seen=0.0, - attempt_count=9, - ), - } - target = target_ms_per_char(300) - assert skill_of(ord("t"), stats, target=target) == 1.0 - assert blocks_transition_focus((ord("t"),), stats, target=target) is True +def test_weakest_key_wins_while_any_key_is_not_cleared(): + keys = _cleared_keys() + keys[S] = _stats(S, SLOW, last_seen=NOW) + assert _focus(keys) == (S, None) -def test_blocks_transition_focus_true_for_never_practiced(): - stats = {1: _stats(1, mean_time_ns=200_000_000.0)} - assert blocks_transition_focus((1, 2), stats, target=200.0) is True +def test_never_practiced_key_is_not_cleared(): + keys = _cleared_keys() + del keys[D] + assert _focus(keys) == (D, None) -def test_blocks_transition_focus_includes_never_practiced_key(): - a, s, h = ord("a"), ord("s"), ord("h") - stats = { - a: _stats(a, mean_time_ns=200_000_000.0), - s: _stats(s, mean_time_ns=200_000_000.0), - } - assert blocks_transition_focus((a, s, h), stats, target=200.0) is True - - -def test_blocks_transition_focus_true_for_measured_weak_key(): - h = ord("h") - stats = {h: _stats(h, mean_time_ns=400_000_000.0)} - assert skill_of(h, stats, target=200.0) < 1.0 - assert blocks_transition_focus((h,), stats, target=200.0) is True - - -def test_select_focus_prefers_never_practiced_key(): - stats = {1: _stats(1, mean_time_ns=100_000_000.0)} - assert select_focus((1, 2), stats, target=200.0, now=1000.0) == 2 - - -def test_select_focus_stays_on_calibrating_key_over_stale_mastered_peer(): - """A key still short of the attempt floor must not lose focus to an - already-mastered, stale peer -- that peer's review-urgency discount can - make it look "weaker" by raw score, but it has already cleared both - conditions (skill + attempts) so it must not preempt an in-progress key.""" - now = 1_700_000_000.0 - five_days = 5 * 86_400.0 - at_target = 200_000_000.0 - calibrating = KeyStats( - codepoint=1, - samples=8, - mean_time_ns=at_target, - error_count=0, - last_seen=now, - attempt_count=8, - ) - mastered_stale = KeyStats( - codepoint=2, - samples=10, - mean_time_ns=at_target, - error_count=0, - last_seen=now - five_days, - attempt_count=10, - ) - stats = {1: calibrating, 2: mastered_stale} - target = 200.0 - assert skill_of(1, stats, target) == 1.0 - assert attempts_of(calibrating) < MIN_CONFIDENCE_ATTEMPTS - assert skill_of(2, stats, target) == 1.0 - assert attempts_of(mastered_stale) >= MIN_CONFIDENCE_ATTEMPTS - # Without the stickiness fix, key 2's review-urgency discount would drop - # its score below key 1's, stealing focus mid-calibration. - assert select_focus((1, 2), stats, target=target, now=now) == 1 - - -def test_select_focus_picks_stale_over_slightly_weaker_recent(): - now = 1_000_000.0 - five_days = 5 * 86_400.0 - stats = { - 1: _stats(1, mean_time_ns=210_000_000.0, last_seen=now - five_days), # ~0.95 - 2: _stats(2, mean_time_ns=235_000_000.0, last_seen=now), # ~0.85 - } - assert select_focus((1, 2), stats, target=200.0, now=now) == 1 - - -def test_select_focus_transition_ignores_same_key_pairs(): - now = 1_700_000_000.0 - unlocked = (ord("e"), ord("a")) - transitions = { - Bigram(ord("e"), ord("e")): _transition( - ord("e"), - ord("e"), - 400_000_000.0, - last_seen=now, - attempt_count=10, - ), - Bigram(ord("a"), ord("a")): _transition( - ord("a"), - ord("a"), - 400_000_000.0, - last_seen=now, - attempt_count=10, - ), - } - assert select_focus_transition(unlocked, transitions, 200.0, now) is None - - -def test_select_focus_transition_skips_unmeasured_when_unlocked_key_unpracticed(): - now = 1_700_000_000.0 - unlocked = (ord("e"), ord("a")) - transitions = { - Bigram(ord("e"), ord("e")): _transition( - ord("e"), - ord("e"), - 400_000_000.0, - last_seen=now, - attempt_count=10, - ), - } - assert select_focus_transition(unlocked, transitions, 200.0, now) is None - - -def test_select_focus_transition_never_picks_same_key_even_when_weakest(): - now = 1_700_000_000.0 - unlocked = (ord("a"), ord("b")) - fast = 100_000_000.0 - transitions = { - Bigram(ord("a"), ord("a")): _transition( - ord("a"), - ord("a"), - 400_000_000.0, - last_seen=now, - attempt_count=10, - ), - Bigram(ord("a"), ord("b")): _transition( - ord("a"), - ord("b"), - fast, - last_seen=now, - attempt_count=10, - ), - Bigram(ord("b"), ord("a")): _transition( - ord("b"), - ord("a"), - fast, - last_seen=now, - attempt_count=10, - ), - Bigram(ord("b"), ord("b")): _transition( - ord("b"), - ord("b"), - 400_000_000.0, - last_seen=now, - attempt_count=10, - ), - } - result = select_focus_transition(unlocked, transitions, 200.0, now) - assert result is not None - assert result[0] != result[1] - - -def test_select_focus_transition_picks_stale_over_slightly_weaker_recent(): - now = 1_000_000.0 - five_days = 5 * 86_400.0 - unlocked = (ord("a"), ord("b")) - fast = 100_000_000.0 - transitions = { - Bigram(ord("a"), ord("a")): _transition(ord("a"), ord("a"), fast, last_seen=now), - Bigram(ord("a"), ord("b")): _transition( - ord("a"), - ord("b"), - 210_000_000.0, - last_seen=now - five_days, - ), - Bigram(ord("b"), ord("a")): _transition(ord("b"), ord("a"), 235_000_000.0, last_seen=now), - Bigram(ord("b"), ord("b")): _transition(ord("b"), ord("b"), fast, last_seen=now), - } - assert select_focus_transition(unlocked, transitions, 200.0, now) == (ord("a"), ord("b")) - - -def test_select_focus_transition_stays_on_calibrating_pair_over_stale_mastered_peer(): - """Mirrors `test_select_focus_stays_on_calibrating_key_over_stale_mastered_peer` - for bigrams: an in-progress pair (below the attempt floor) must not lose - focus to an already-cleared, merely-stale measured pair.""" - now = 1_700_000_000.0 - five_days = 5 * 86_400.0 - unlocked = (ord("a"), ord("b"), ord("c")) - calibrating = Bigram(ord("a"), ord("b")) - mastered_stale = Bigram(ord("b"), ord("c")) - transitions = { - calibrating: _transition(*calibrating, 200_000_000.0, last_seen=now, attempt_count=2), - mastered_stale: _transition( - *mastered_stale, 200_000_000.0, last_seen=now - five_days, attempt_count=10 - ), - } - target = 200.0 - assert skill_from_stats(transitions[calibrating], target) == 1.0 - assert attempts_of(transitions[calibrating]) < 4 # below MIN_TRANSITION_CONFIDENCE_ATTEMPTS - assert skill_from_stats(transitions[mastered_stale], target) == 1.0 - assert attempts_of(transitions[mastered_stale]) >= 4 - assert select_focus_transition(unlocked, transitions, target, now) == calibrating +def test_last_key_is_kept_while_it_still_needs_work(): + keys = _cleared_keys() + keys[A] = _stats(A, SLOW, last_seen=NOW) + keys[S] = _stats(S, 300_000_000.0, last_seen=NOW) + assert _focus(keys) == (A, None) + assert _focus(keys, last_key=S) == (S, None) -def test_select_focus_transition_returns_none_without_unlocked_pairs(): - assert select_focus_transition((), {}, target=200.0, now=1000.0) is None - assert select_focus_transition((ord("a"),), {}, target=200.0, now=1000.0) is None +def test_stalled_last_key_goes_back_to_the_pool(): + keys = _cleared_keys() + keys[A] = _stats(A, SLOW, last_seen=NOW) + keys[S] = _stats(S, 300_000_000.0, last_seen=NOW, attempt_count=30) + assert _focus(keys, last_key=S) == (A, None) -def test_select_focus_transition_picks_unmeasured_pair_when_all_keys_practiced(): - stats = { - 1: KeyStats(1, 10, 200_000_000.0, 0, 1.0, attempt_count=10), - 2: KeyStats(2, 10, 200_000_000.0, 0, 1.0, attempt_count=10), - } - assert select_focus_transition((1, 2), {}, target=200.0, now=1000.0, key_stats=stats) == ( - 1, - 2, - ) +def test_cleared_last_key_is_not_kept(): + keys = _cleared_keys() + keys[A] = _stats(A, SLOW, last_seen=NOW) + assert _focus(keys, last_key=S) == (A, None) -def test_select_focus_transition_skips_unmeasured_pair_on_cold_start(): - assert select_focus_transition((1, 2), {}, target=200.0, now=1000.0) is None - - -def test_select_focus_transition_ignores_unmeasured_pairs(): - unlocked = (ord("a"), ord("b"), ord("c")) - now = 1_700_000_000.0 - fast = 100_000_000.0 - transitions = { - Bigram(ord("a"), ord("b")): _transition( - ord("a"), - ord("b"), - 400_000_000.0, - last_seen=now, - attempt_count=10, - ), - Bigram(ord("b"), ord("c")): _transition( - ord("b"), - ord("c"), - fast, - last_seen=now, - attempt_count=10, - ), - } - assert select_focus_transition(unlocked, transitions, 200.0, now) == (ord("a"), ord("b")) - - -def test_select_focus_transition_prefers_newest_key_over_older_measured_pair(): - """Once a third key is unlocked, its bigrams should win focus over an - already-measured (and merely weak, not unmeasured) older pair, so a - freshly-opened letter gets bigram practice right away.""" - unlocked = (ord("a"), ord("b"), ord("c")) - now = 1_700_000_000.0 - stats = { - ord("a"): KeyStats(ord("a"), 10, 200_000_000.0, 0, now, attempt_count=10), - ord("b"): KeyStats(ord("b"), 10, 200_000_000.0, 0, now, attempt_count=10), - ord("c"): KeyStats(ord("c"), 10, 200_000_000.0, 0, now, attempt_count=10), - } - transitions = { - Bigram(ord("a"), ord("b")): _transition( - ord("a"), - ord("b"), - 400_000_000.0, - last_seen=now, - attempt_count=10, - ), - } - # c is unlocked and practiced solo, but none of its transitions are - # measured yet -- it should still win over the already-measured (a, b). - result = select_focus_transition(unlocked, transitions, 200.0, now, key_stats=stats) - assert result is not None - assert ord("c") in result - - -def test_select_focus_transition_uses_measured_pairs_once_newest_key_has_data(): - """Once the newest key has its own measured transition, ordinary - weakest-pair scoring resumes -- no permanent bias toward the newest key.""" - unlocked = (ord("a"), ord("b"), ord("c")) - now = 1_700_000_000.0 - fast = 100_000_000.0 - stats = { - ord("a"): KeyStats(ord("a"), 10, 200_000_000.0, 0, now, attempt_count=10), - ord("b"): KeyStats(ord("b"), 10, 200_000_000.0, 0, now, attempt_count=10), - ord("c"): KeyStats(ord("c"), 10, 200_000_000.0, 0, now, attempt_count=10), - } - transitions = { - Bigram(ord("a"), ord("b")): _transition( - ord("a"), - ord("b"), - 400_000_000.0, - last_seen=now, - attempt_count=10, - ), - Bigram(ord("b"), ord("c")): _transition( - ord("b"), - ord("c"), - fast, - last_seen=now, - attempt_count=10, - ), - } - assert select_focus_transition(unlocked, transitions, 200.0, now, key_stats=stats) == ( - ord("a"), - ord("b"), - ) +def test_gate_pair_wins_over_other_weak_pairs(): + gate = Bigram(S, D) + transitions = _cleared_pairs() + transitions[Bigram(A, S)] = _transition(A, S, SLOW, last_seen=NOW) + assert _focus(_cleared_keys(), transitions, gate=(gate,)) == (D, gate) -def test_select_focus_transition_falls_back_to_unmeasured_unlocked_pair(): - unlocked = (ord("a"), ord("b")) - stats = { - ord("a"): KeyStats(ord("a"), 10, 200_000_000.0, 0, 1.0, attempt_count=10), - ord("b"): KeyStats(ord("b"), 10, 200_000_000.0, 0, 1.0, attempt_count=10), - } - transitions = { - Bigram(ord("y"), ord("z")): _transition( - ord("y"), - ord("z"), - 100_000_000.0, - last_seen=1_700_000_000.0, - ), - } - assert select_focus_transition( - unlocked, transitions, 200.0, now=1_700_000_000.0, key_stats=stats - ) == (ord("a"), ord("b")) +def test_weak_pairs_rank_by_weakness_times_language_frequency(): + common, rare = Bigram(A, S), Bigram(S, A) + transitions = _cleared_pairs() + del transitions[common] + del transitions[rare] + freq = {common: 100, rare: 1} + assert _focus(_cleared_keys(), transitions, freq=freq) == (S, common) -def test_focus_key_from_transition_uses_next_endpoint(): - assert focus_key_from_transition(ord("t"), ord("h")) == ord("h") +def test_pair_that_never_occurs_in_the_language_is_skipped(): + never, real = Bigram(A, S), Bigram(S, A) + transitions = _cleared_pairs() + del transitions[never] + transitions[real] = _transition(S, A, SLOW, last_seen=NOW) + assert _focus(_cleared_keys(), transitions, freq={never: 0}) == (A, real) -def test_key_attempt_floor_blocks_transition_focus(): - cp = ord("a") - stats = { - cp: KeyStats(cp, 9, 100_000_000.0, 0, 1.0, attempt_count=9), - } - assert blocks_transition_focus((cp,), stats, 200.0, min_attempts=10) +def test_last_pair_is_kept_over_a_more_frequent_weak_pair(): + common, last = Bigram(A, S), Bigram(S, D) + transitions = _cleared_pairs() + del transitions[common] + transitions[last] = _transition(S, D, SLOW, last_seen=NOW) + freq = {common: 100} + assert _focus(_cleared_keys(), transitions, freq=freq) == (S, common) + assert _focus(_cleared_keys(), transitions, freq=freq, last_pair=last) == (D, last) -def test_sparse_old_transition_does_not_preempt_gating_cohort(): - a, b, c = map(ord, "abc") - gate = Bigram(b, c) - old = Bigram(a, b) - transitions = { - old: _transition(a, b, 400_000_000.0, attempt_count=3), - } - assert ( - select_focus_transition( - (a, b, c), - transitions, - 200.0, - 1_000.0, - gating_candidates=(gate,), - min_attempts=4, - ) - == gate - ) +def test_most_overdue_key_or_pair_is_reviewed_when_everything_cleared(): + stale_pair = Bigram(D, A) + transitions = _cleared_pairs() + transitions[stale_pair] = _transition(D, A, FAST, last_seen=NOW - 3 * DAY) + keys = _cleared_keys() + keys[S] = _stats(S, FAST, last_seen=NOW - 2 * DAY) + assert _focus(keys, transitions) == (A, stale_pair) + keys[S] = _stats(S, FAST, last_seen=NOW - 5 * DAY) # ties the pair: keys first + assert _focus(keys, transitions) == (S, None) -def test_genuine_old_transition_regression_preempts_gating_cohort(): - a, b, c = map(ord, "abc") - gate = Bigram(b, c) - old = Bigram(a, b) - transitions = { - old: _transition(a, b, 400_000_000.0, attempt_count=4), - } - assert ( - select_focus_transition( - (a, b, c), - transitions, - 200.0, - 1_000.0, - gating_candidates=(gate,), - min_attempts=4, - ) - == old - ) +def test_first_unlocked_key_when_nothing_needs_work(): + assert _focus(_cleared_keys(), _cleared_pairs()) == (A, None) diff --git a/tests/domain/test_markov.py b/tests/domain/test_markov.py index 0a3083d..8ac12cb 100644 --- a/tests/domain/test_markov.py +++ b/tests/domain/test_markov.py @@ -18,6 +18,13 @@ def test_sample_falls_back_to_shorter_context(): assert table.sample("xb", frozenset("bcxz"), rng) == "c" +def test_pair_weight_reads_the_first_letter_row_then_the_global_row(): + table = TransitionTable(order=2, transitions={"a": {"s": 7}, "": {"s": 2}}) + assert table.pair_weight(Bigram(ord("a"), ord("s"))) == 7 + assert table.pair_weight(Bigram(ord("a"), ord("d"))) == 0 + assert table.pair_weight(Bigram(ord("d"), ord("s"))) == 2 + + def test_sample_falls_back_to_global_distribution(): table = TransitionTable(order=2, transitions={"": {"e": 1}}) rng = Random(1) diff --git a/tests/domain/test_newest_key.py b/tests/domain/test_newest_key.py index 86e6979..313897e 100644 --- a/tests/domain/test_newest_key.py +++ b/tests/domain/test_newest_key.py @@ -1,5 +1,4 @@ -from keystrike.domain.focus import newest_key_unmeasured_pairs -from keystrike.domain.models import Bigram, KeyStats, TransitionStats +from keystrike.domain.models import Bigram, KeyStats from keystrike.domain.newest_key import effective_gating_bigram_limit, newest_key_gating_cohort @@ -20,57 +19,6 @@ def _stats( ) -def _transition( - prev_cp: int, - next_cp: int, - mean_time_ns: float, - *, - last_seen: float = 0.0, - error_count: int = 0, - attempt_count: int | None = None, -) -> TransitionStats: - attempts = attempt_count if attempt_count is not None else 10 + error_count - return TransitionStats( - prev_cp=prev_cp, - next_cp=next_cp, - samples=10, - mean_time_ns=mean_time_ns, - error_count=error_count, - last_seen=last_seen, - attempt_count=attempts, - ) - - -def test_newest_key_unmeasured_pairs_excludes_unpracticed_cascade_key(): - """A key unlocked by cascade (no KeyStats yet) shouldn't get bigram - candidacy just for sitting next to the newest practiced key.""" - unlocked = (ord("a"), ord("s"), ord("h"), ord("d"), ord("l")) - now = 1_700_000_000.0 - stats = { - cp: KeyStats(cp, 10, 200_000_000.0, 0, now, attempt_count=10) - for cp in (ord("a"), ord("s"), ord("h"), ord("d")) - } - pairs = newest_key_unmeasured_pairs(unlocked, {}, stats) - assert pairs - assert all(ord("l") not in (p.prev_cp, p.next_cp) for p in pairs) - assert all(ord("d") in (p.prev_cp, p.next_cp) for p in pairs) - - -def test_newest_key_unmeasured_pairs_empty_once_newest_has_any_measured_pair(): - unlocked = (ord("a"), ord("s"), ord("d")) - now = 1_700_000_000.0 - stats = { - cp: KeyStats(cp, 10, 200_000_000.0, 0, now, attempt_count=10) - for cp in (ord("a"), ord("s"), ord("d")) - } - transitions = { - Bigram(ord("s"), ord("d")): _transition( - ord("s"), ord("d"), 200_000_000.0, last_seen=now, attempt_count=10 - ), - } - assert newest_key_unmeasured_pairs(unlocked, transitions, stats) == [] - - def test_gating_cohort_is_bounded_deterministic_and_measurement_independent(): unlocked = tuple(map(ord, "abcd")) stats = {cp: _stats(cp, mean_time_ns=100_000_000.0) for cp in unlocked} diff --git a/tests/domain/test_unlock.py b/tests/domain/test_unlock.py index c1b071a..f0bd7c5 100644 --- a/tests/domain/test_unlock.py +++ b/tests/domain/test_unlock.py @@ -1,8 +1,18 @@ -from keystrike.domain.confidence import confidence_of, skill_of -from keystrike.domain.models import Bigram, KeyStats, TransitionStats, UnlockTuning +from keystrike.domain.aggregate import combine_sessions, tally_session +from keystrike.domain.confidence import confidence_of, skill_of, stall_attempts_cap +from keystrike.domain.enums import Mode +from keystrike.domain.models import ( + Bigram, + KeyStats, + Keystroke, + SessionResult, + TransitionStats, + UnlockTuning, +) from keystrike.domain.newest_key import newest_key_gating_cohort from keystrike.domain.unlock import ( compute_unlocked, + gating_bigram_is_ready, newest_key_clears_transition_gate, newest_key_transition_gate_progress, ) @@ -366,3 +376,84 @@ def test_newest_key_clears_transition_gate_stall_cap_overrides_weak_pair(): ) is False ) + + +def _header(i: int) -> SessionResult: + return SessionResult( + schema_version=5, + session_id=f"s{i}", + started_at=float(i), + duration_ns=1_000_000_000, + layout="qwerty", + mode=Mode.ADAPTIVE, + lesson_alphabet=(), + focus_key=None, + total_keystrokes=0, + correct_keystrokes=0, + ) + + +def _lesson(text: str, ms: int) -> list[Keystroke]: + return [ + Keystroke(codepoint=ord(ch), typed=ord(ch), t_ns=i * ms * 1_000_000, correct=True) + for i, ch in enumerate(text) + ] + + +def test_compute_unlocked_sixteen_keys_reach_attempt_floor_over_window(): + """16 unlocked keys and 12-word lessons (about 2 presses per key per + lesson): over the default 10-session window every key has 20 real + presses, so the default floor of 10 is met. Recency-weighted attempts + settle near 6.5 and held the next letter back forever.""" + learn_order = tuple(ord(ch) for ch in "abcdefghijklmnopq") + sixteen = "abcdefghijklmnop" + sessions = [(_header(i), tally_session(_lesson(sixteen * 2, 200))) for i in range(10)] + stats = combine_sessions(sessions).keys + assert all(stats[cp].attempt_count == 20 for cp in learn_order[:16]) + tuning = UnlockTuning() + assert tuning.min_confidence_attempts == 10 + unlocked = compute_unlocked(learn_order, 16, stats, target=200.0, tuning=tuning) + assert unlocked == learn_order + + +def test_transition_stall_cap_reachable_at_two_attempts_per_lesson(): + """A stuck cohort pair that gets 2 of the weak-focus coverage slots per + lesson reaches the default stall cap (3 x 4 = 12 real attempts) on the + 6th lesson in the window, instead of settling near 6.5 weighted ones.""" + ea = Bigram(ord("e"), ord("a")) + tuning = UnlockTuning() + cap = stall_attempts_cap(tuning.min_transition_confidence_attempts) + assert cap == 12 + + def ready_after(lessons: int) -> bool: + # "eaea" types e->a twice, slowly (400 ms vs a 200 ms target). + sessions = [(_header(i), tally_session(_lesson("eaea", 400))) for i in range(lessons)] + transitions = combine_sessions(sessions).transitions + return gating_bigram_is_ready(ea, transitions, 200.0, tuning=tuning, stall_attempts_cap=cap) + + assert not ready_after(5) + assert ready_after(6) + assert ready_after(10) + + +def test_gate_pair_clears_at_accuracy_target(): + """The gate uses the same `clears_threshold` rule as key unlock: a pair + at target speed with 95% accuracy is ready — no near-perfect bar.""" + ea = Bigram(ord("e"), ord("a")) + pair = TransitionStats(ord("e"), ord("a"), 19, 200_000_000.0, 1, 0.0, attempt_count=20) + assert gating_bigram_is_ready(ea, {ea: pair}, 200.0) + sloppy = TransitionStats(ord("e"), ord("a"), 15, 200_000_000.0, 1, 0.0, attempt_count=16) + assert not gating_bigram_is_ready(ea, {ea: sloppy}, 200.0) + + +def test_compute_unlocked_allows_one_recent_typo_on_a_well_practiced_key(): + """Key s: 60 weighted samples, speed 1.3, one typo last lesson. Accuracy + 60 / 61 = 0.984 is above the 95% target, so s no longer blocks the unlock + (it read as skill 0.98 against a 100% accuracy bar).""" + a, s, d = ord("a"), ord("s"), ord("d") + stats = { + a: _stats(a, mean_time_ns=200_000_000.0 / 1.3), + s: KeyStats(s, 60, 200_000_000.0 / 1.3, 1, 0.0, attempt_count=61), + } + assert skill_of(s, stats, 200.0) == 1.0 + assert compute_unlocked((a, s, d), 2, stats, target=200.0) == (a, s, d) diff --git a/tests/infrastructure/test_aggregates_cache.py b/tests/infrastructure/test_aggregates_cache.py index 71bf993..99d272d 100644 --- a/tests/infrastructure/test_aggregates_cache.py +++ b/tests/infrastructure/test_aggregates_cache.py @@ -60,7 +60,7 @@ def test_get_pre_transition_cache_sets_transitions_computed_false(paths): cache._file("qwerty").write_text( json.dumps( { - "schema_version": 1, + "schema_version": 2, "layout": "qwerty", "keys": { "97": { @@ -118,7 +118,7 @@ def test_get_strips_same_key_transitions_from_stale_cache(paths): cache._file("qwerty").write_text( json.dumps( { - "schema_version": 1, + "schema_version": 2, "layout": "qwerty", "keys": {}, "transitions": { @@ -157,7 +157,7 @@ def test_get_repairs_zero_samples_when_mean_time_present(paths): cache._file("qwerty").write_text( json.dumps( { - "schema_version": 1, + "schema_version": 2, "layout": "qwerty", "keys": {}, "transitions": { @@ -197,7 +197,7 @@ def test_get_repairs_zero_attempt_count_when_samples_present(paths): cache._file("qwerty").write_text( json.dumps( { - "schema_version": 1, + "schema_version": 2, "layout": "qwerty", "keys": {}, "transitions": { @@ -229,3 +229,25 @@ def test_get_repairs_zero_attempt_count_when_samples_present(paths): ) > 0.0 ) + + +def test_get_treats_schema_1_cache_as_miss(paths): + """Schema 1 stored recency-weighted attempt counts; it must rebuild.""" + cache = FileAggregatesCache(paths) + cache._file("qwerty").write_text( + json.dumps({"schema_version": 1, "layout": "qwerty", "keys": {}, "transitions": {}}), + encoding="utf-8", + ) + assert cache.get("qwerty") is None + + +def test_put_then_get_keeps_fractional_errors(paths): + cache = FileAggregatesCache(paths) + aggregates = LayoutAggregates( + keys={ord("a"): KeyStats(ord("a"), 7, 200_000_000.0, 0.7, 1.0, attempt_count=9)}, + ) + cache.put("qwerty", aggregates) + loaded = cache.get("qwerty") + assert loaded is not None + assert loaded.keys[ord("a")].error_count == 0.7 + assert json.loads(cache._file("qwerty").read_text(encoding="utf-8"))["schema_version"] == 2 diff --git a/tests/infrastructure/test_session_repo_jsonl.py b/tests/infrastructure/test_session_repo_jsonl.py index 9c65287..d282177 100644 --- a/tests/infrastructure/test_session_repo_jsonl.py +++ b/tests/infrastructure/test_session_repo_jsonl.py @@ -66,6 +66,20 @@ def test_round_trip_single_session(paths): assert headers[0].stats.transitions[Bigram(ord("a"), ord("b"))] == KeyTally(1, 100, 0, 1) +def test_round_trip_focus_pair(paths): + repo = JsonlSessionRepository(paths) + header = replace(_header(), focus_pair=Bigram(ord("s"), ord("a"))) + repo.save_header(header) + assert list(JsonlSessionRepository(paths).iter_headers("qwerty")) == [header] + + +def test_row_without_focus_pair_reads_back_as_none(paths): + repo = JsonlSessionRepository(paths) + repo.save_header(_header()) + assert "focus_pair" not in paths.sessions_index.read_text(encoding="utf-8") + assert next(iter(repo.iter_headers("qwerty"))).focus_pair is None + + def test_iter_headers_filters_by_layout(paths): repo = JsonlSessionRepository(paths) repo.save_header(_header(sid=_VALID_ULID_A, layout="qwerty")) @@ -240,6 +254,7 @@ def test_legacy_row_without_stats_reads_as_empty_stats(paths): headers = list(JsonlSessionRepository(paths).iter_headers("qwerty")) assert headers[0].session_id == _VALID_ULID_D assert headers[0].stats.is_empty + assert headers[0].focus_pair is None @pytest.mark.parametrize( @@ -268,6 +283,31 @@ def test_malformed_stats_skip_the_row_not_the_index(paths, stats_json): assert headers == [_VALID_ULID_A, _VALID_ULID_B] +@pytest.mark.parametrize( + "pair_json", + [ + "[97]", # wrong arity + "[97, 98, 99]", # wrong arity + '"97,98"', # not a list + '[97, "x"]', # not a codepoint + ], +) +def test_malformed_focus_pair_skips_the_row_not_the_index(paths, pair_json): + repo = JsonlSessionRepository(paths) + repo.save_header(_header(sid=_VALID_ULID_A)) + with paths.sessions_index.open("a", encoding="utf-8") as fh: + fh.write( + f'{{"schema_version": 5, "session_id": "{_VALID_ULID_E}", "layout": "qwerty", ' + '"started_at": 1700000000.0, "duration_ns": 1000000000, "mode": "adaptive", ' + '"lesson_alphabet": [], "focus_key": null, "total_keystrokes": 0, ' + f'"correct_keystrokes": 0, "focus_pair": {pair_json}}}\n' + ) + repo.save_header(_header(sid=_VALID_ULID_B)) + + headers = [h.session_id for h in JsonlSessionRepository(paths).iter_headers("qwerty")] + assert headers == [_VALID_ULID_A, _VALID_ULID_B] + + def test_replace_all_headers_rewrites_index_in_place(paths): repo = JsonlSessionRepository(paths) a = replace(_header(sid=_VALID_ULID_A, started_at=1.0), stats=_stats()) diff --git a/uv.lock b/uv.lock index 7f54163..5ebb915 100644 --- a/uv.lock +++ b/uv.lock @@ -79,7 +79,7 @@ wheels = [ [[package]] name = "keystrike" -version = "2.1.0" +version = "2.2.0" source = { editable = "." } dependencies = [ { name = "platformdirs" },