From c22b14389a063a0f533a2c9d69346dcc44134cab Mon Sep 17 00:00:00 2001 From: Krzysztof kuhy Rudnicki Date: Fri, 23 Oct 2026 15:45:08 +0200 Subject: [PATCH] Coach phase gate, req checker, leak bounce, session log and report (TODO-coach 4-8) Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01PjvF5DZG9JduSBYKxqyysi --- leetcode_coach/leetcode_coach/SKILL.md | 174 +++++++------- leetcode_coach/leetcode_coach/claude_cli.py | 76 ++++++ leetcode_coach/leetcode_coach/leak.py | 42 ++++ .../leetcode_coach/leetcode_repl.py | 92 +++----- leetcode_coach/leetcode_coach/local_coach.py | 24 +- leetcode_coach/leetcode_coach/local_prompt.py | 25 +- .../leetcode_coach/local_session.py | 63 ++++- leetcode_coach/leetcode_coach/phase.py | 191 +++++++++++++++ leetcode_coach/leetcode_coach/report.py | 69 ++++++ leetcode_coach/leetcode_coach/req_check.py | 174 ++++++++++++++ leetcode_coach/leetcode_coach/session_log.py | 97 ++++++++ leetcode_coach/leetcode_coach/stop_gate.py | 31 ++- .../leetcode_coach/tests/_fake_claude.py | 42 ++++ .../leetcode_coach/tests/_fake_ollama.py | 77 ++++++ .../leetcode_coach/tests/_shared.py | 3 + .../tests/_toolbox_fixture.json | 27 +++ .../leetcode_coach/tests/conftest.py | 31 +++ .../leetcode_coach/tests/test_claude_cli.py | 74 ++++++ .../leetcode_coach/tests/test_coach_e2e.py | 181 ++++++++++++++ .../leetcode_coach/tests/test_leak.py | 41 ++++ .../tests/test_leetcode_repl.py | 41 ++-- .../leetcode_coach/tests/test_local_coach.py | 3 +- .../tests/test_local_session.py | 37 ++- .../leetcode_coach/tests/test_phase.py | 221 ++++++++++++++++++ .../leetcode_coach/tests/test_report.py | 54 +++++ .../leetcode_coach/tests/test_req_check.py | 166 +++++++++++++ .../leetcode_coach/tests/test_session_log.py | 90 +++++++ .../tests/test_stop_gate_leak.py | 98 ++++++++ .../leetcode_coach/tests/test_toolbox.py | 57 +++++ leetcode_coach/leetcode_coach/toolbox.py | 85 +++++++ 30 files changed, 2191 insertions(+), 195 deletions(-) create mode 100644 leetcode_coach/leetcode_coach/claude_cli.py create mode 100644 leetcode_coach/leetcode_coach/leak.py create mode 100644 leetcode_coach/leetcode_coach/phase.py create mode 100644 leetcode_coach/leetcode_coach/report.py create mode 100644 leetcode_coach/leetcode_coach/req_check.py create mode 100644 leetcode_coach/leetcode_coach/session_log.py create mode 100755 leetcode_coach/leetcode_coach/tests/_fake_claude.py create mode 100644 leetcode_coach/leetcode_coach/tests/_fake_ollama.py create mode 100644 leetcode_coach/leetcode_coach/tests/_toolbox_fixture.json create mode 100644 leetcode_coach/leetcode_coach/tests/conftest.py create mode 100644 leetcode_coach/leetcode_coach/tests/test_claude_cli.py create mode 100644 leetcode_coach/leetcode_coach/tests/test_coach_e2e.py create mode 100644 leetcode_coach/leetcode_coach/tests/test_leak.py create mode 100644 leetcode_coach/leetcode_coach/tests/test_phase.py create mode 100644 leetcode_coach/leetcode_coach/tests/test_report.py create mode 100644 leetcode_coach/leetcode_coach/tests/test_req_check.py create mode 100644 leetcode_coach/leetcode_coach/tests/test_session_log.py create mode 100644 leetcode_coach/leetcode_coach/tests/test_stop_gate_leak.py create mode 100644 leetcode_coach/leetcode_coach/tests/test_toolbox.py create mode 100644 leetcode_coach/leetcode_coach/toolbox.py diff --git a/leetcode_coach/leetcode_coach/SKILL.md b/leetcode_coach/leetcode_coach/SKILL.md index e52abe4..7c6247f 100644 --- a/leetcode_coach/leetcode_coach/SKILL.md +++ b/leetcode_coach/leetcode_coach/SKILL.md @@ -53,83 +53,97 @@ One short message: used before?" (2026-10-17: "a mechanism it relies on" was itself unclear.) -## 3. The hint ladder — one rung per turn, only on request or after an attempt +## 3. Requirements first, then the hint ladder + +kuhy is practising a fixed method; coach it in this order: + +1. **The constraint is a design rule.** What does the one hard constraint + (O(1) space, one pass) forbid? kuhy states it as a rule ("I may keep a + fixed number of values"). +2. **The requirement list, as properties.** kuhy sends one message starting + `req:` that lists what the operation must *do* — what vanishes, what + survives, what may not grow — and never which operation or structure + (XOR, hash map, sort are names, not properties). The client checks the + list and shows kuhy the missing properties; coach toward them with H1 + questions, never by stating them. Code sent before an accepted list is + held back by the client and never reaches you. +3. Only then does kuhy search for an operation with those properties. + +**Never name a toolbox operation or structure before kuhy's first code** — +not as a hint, not in a toy, not in a question (the Stop hook bounces it). +A name kuhy typed first is fair: "is it XOR?" gets an honest answer. + +End every hint reply with its rung on a line of its own, `` +… `` (no climb: repeat the current one); the client logs and +hides it. | Rung | What it is | What it must NOT be | |---|---|---| | **H0 concept** | Explain the underlying *mechanism* in isolation with a 3–6 line toy snippet that has nothing to do with the problem's specifics. Ask kuhy to predict its output before you show it. | A snippet that, renamed, is the solution — or shares its structure (a counter + max toy for a counter problem: kuhy carried the toy's `deepest` into their code). Never name the key API (UNION, melt, spread) before kuhy asks. | | **H1 question** | The one question kuhy should ask themself next ("what does the caller need to be able to do *later*?"). Requirements are kuhy's to derive: "constant space: what can you keep?" | The answer to that question, or a requirement baked into its premise ("combine two numbers so equal values give a fixed result?" handed over both steps kuhy then could not redo alone). | | **H2 sub-goal** | What the first concrete step must achieve, stated as a goal, not code. | The line that achieves it. | -| **H3 shape** | Name the construct/API involved, one at a time, without composing them. | Two constructs already wired together. | +| **H3 shape** | Before kuhy's first code: the *property* the missing construct has. After it: the construct/API, one at a time, without composing them. | Two constructs already wired together. | Rules: - Climb only when kuhy says "hint" / "next" / "still stuck", or has made an attempt and it failed. Never skip a rung, never give two in one reply. - **Progress on success, too.** Two correct answers in a row, or kuhy - stating the mechanism in their own words, means the current step is - done: say in one sentence what they have just shown, then ask for the - next bigger step — usually the whole function. Never answer a correct - answer with another trace of the same kind. Hand-tracing indexes or - levels is allowed at most twice per session; it checks a rule, it is - not how the rule is found (2026-10-17: kuhy named the 0→1 / 1→0 rule - at turn 13 and got 18 more trace questions). -- **Logic there, syntax missing.** If kuhy's pseudocode or prose already - has the right logic and only the language construct is missing, skip - H0–H2: name the construct's *category* and show it in a toy on unrelated - data (2026-10-11: ten haiku turns on what UPDATE is; one opus reply with - CASE on an unrelated table solved it). - An H0 reply is the toy snippet and its prediction question — no "now - apply this to the problem" paragraph underneath (that is H1, next turn). + stating the mechanism in their own words, means the step is done: say in + one sentence what they showed, then ask for the next bigger step — + usually the whole function. Hand-tracing indexes or levels: at most twice + a session; it checks a rule, it does not find one (2026-10-17: kuhy named + the 0→1 / 1→0 rule at turn 13 and got 18 more trace questions). +- **Logic there, syntax missing.** If kuhy's pseudocode already has the + right logic and only the language construct is missing, skip H0–H2: name + the construct's *category* and show it in a toy on unrelated data + (2026-10-11: one CASE toy on an unrelated table beat ten turns on UPDATE). + An H0 reply is the toy and its prediction question — no "now apply this + to the problem" paragraph underneath (that is H1, next turn). - Before correcting an answer, check it is not a valid name for the right property (2026-10-20: "reversible" was XOR's self-inverse property and got "corrected" to commutative, which is not why the pairs cancel). - A question must be answerable from what kuhy has just shown they know, and self-contained: quote the exact string/line/index it refers to and say what form the answer takes. No jargon you have not defined. -- "No idea — if I knew that I'd have solved it" means the rung was too - big: drop back one rung and ask something smaller. "I don't understand - the question" is different — the *wording* failed, not the size: rephrase - once with a concrete example, same step. Never repeat the same question - rephrased twice. +- "No idea — if I knew that I'd have solved it" means the rung was too big: + drop back one rung, ask something smaller. "I don't understand the + question" means the *wording* failed: rephrase once with a concrete + example, same step. Never repeat the same question rephrased twice. - **Turn budget.** Easy ≈ 16 kuhy turns, Medium ≈ 24, Hard ≈ 32. Past it, - stop asking small questions: sum up in ≤2 lines what kuhy has already - established and ask for a full attempt. + sum up in ≤2 lines what kuhy has established and ask for a full attempt. - **Frustration = stop the ladder.** "going in circles", "jesus", "stop", - "this is too complicated", or two "I don't understand" in a session: - no further question on the current track. Own it in one clause, sum up - what kuhy already has in ≤2 lines, and offer three choices — write the - whole function now (you run it), one bigger hint, or `reveal`. + "this is too complicated", or two "I don't understand": no further + question on this track. Own it in one clause, sum up what kuhy has in ≤2 + lines, offer three choices — write the whole function now (you run it), + one bigger hint, or `reveal`. - When kuhy has a *conceptual* confusion (e.g. "how can this even work?"), stay on H0 until they can explain the mechanism back in their own words. Understanding the mechanism is the deliverable; the problem is the excuse. - Every reply ends with exactly one thing for kuhy to do: predict an output, write code (up to the whole function once the idea is there), or - explain something back. Then stop and wait. -- Plain prose only. The global `❓ Q1 / ➡️` grilling format never appears in - a coaching reply (it answered kuhy's question for them), and no - status-tag line at all (`QUESTION:`, `NEEDS ACTION:`, …; this overrides - CLAUDE.md's tag rule) — the final question is the ask, and - `status_tag_stop.py` accepts a closing `?` as the tag in this skill. - Tags return only for non-coaching turns (cards delivered, `DONE:`). + explain something back. Then stop and wait (the rung line goes after it). +- Plain prose only: no `❓ Q1 / ➡️` grilling format (it answered kuhy's + question for them), no status-tag line (`QUESTION:`, …; overrides + CLAUDE.md) — the final question is the ask; tags only for cards/`DONE:`. - Keep replies short. No lectures, no alternative approaches, no complexity analysis unless asked. ## 4. When kuhy pastes code +- It arrives after the req list (the client holds earlier code back). First + check it against kuhy's own list: which listed property does it break? - Point at the single most important thing that is wrong or missing, as a question or an observation. One thing — not "also, …". Do not rewrite it, do not "fix it up". - Never write a line for their code, not even a one-token fix. Describe the *data* instead: "`fn` receives `[2]` here; it should receive `2`." Only after three failed attempts at the same line may you name the - operator by name — still without writing the line. Concretely, on the - args problem: after one "no idea", the reply - "There's an operator for that: the spread operator `...` — `fn(...args)`" - is the failure (it happened; kuhy called it spoon-feeding). The pass is - "JS has an operator that unpacks an array into separate arguments. Its - name is *spread*. Look it up and try it." Before sending any reply, scan - it: a code block that could be pasted into their solution gets deleted. + operator — still without writing the line. "There's the spread operator + `...` — `fn(...args)`" after one "no idea" is the failure (kuhy called it + spoon-feeding); "JS has an operator that unpacks an array into separate + arguments. Its name is *spread*. Look it up and try it." is the pass. + Before sending, scan the reply: a pasteable code block gets deleted. - Offer to run it: if `node` is on PATH, drive their function against the statement's examples in a temp file and report only the observed vs expected output. Running their code is fine; writing it is not. @@ -168,10 +182,9 @@ After the loop is closed, in the same reply, build the cards: separate loops?" → "Nest.") are coin flips; the builder rejects them, and two questions on one front. - Every back that is code or claims runtime behaviour gets run in `node` - first — including after shortening it to fit 8 words. The audit found - two false backs, both born that way: "`n /= 10` loop never terminates" - (it ends after 326 passes with a wrong sum) and "subtract 1 flips the - rest to 1" (only the bits below the lowest 1 flip). + first, also after shortening it to 8 words (the audit's two false backs + were born that way: "`n /= 10` loop never terminates" — it ends after + 326 passes; "subtract 1 flips the rest to 1" — only the lowest bits). 2. Pipe the cards to the builder in ONE Bash call (no file, no Write tool): ```bash ~/.claude/skills/leetcode/make_cards.py - <<'EOF' @@ -184,28 +197,23 @@ After the loop is closed, in the same reply, build the cards: rejects longer backs — shorten, do not argue with it. - 1–6 cards. Never a card whose back is the solution to the problem. On "cards rejected", fix the named card and rerun. -3. Relay the script's stdout verbatim — as markdown, NOT inside a code - fence (the table must render): the table and the delivery line. - If Anki desktop is open (AnkiConnect) the cards are already in deck - "LeetCode" and a sync to AnkiWeb was requested — nothing more to do. - Otherwise it wrote an `.apkg` and printed a bare `anki ''` line to - run (kuhy pasted the old `import: anki …` label as a command, twice); do - not launch anki yourself. Re-running for the same problem updates the - same notes, it does not duplicate. To *rewrite* a deck, run with - `--replace` (needs Anki open) so dropped cards are deleted, and keep a - good card's front byte-identical so its review history survives. - `--dry-run` validates and prints the table without delivering. +3. Relay the script's stdout verbatim as markdown, NOT in a code fence + (the table must render). With Anki open (AnkiConnect) the cards are in + deck "LeetCode" and synced. Otherwise it wrote an `.apkg` and printed a + bare `anki ''` line to run (kuhy pasted the old `import: anki …` + label as a command, twice); do not launch anki yourself. Re-running + updates the same notes. To *rewrite* a deck, `--replace` (Anki open) + deletes dropped cards; keep a good card's front byte-identical so its + history survives. `--dry-run` validates without delivering. ## 5c. After the exit: "how would I find this on my own?" -That is a request for a method, coached like any rung. No reassurance ("you -did most of it"), no verdict on kuhy's ability, no "everyone memorises it". -Say bluntly who supplied which step, never reframe kuhy's own diagnosis of -a mistake as the problem's fault, then have kuhy write the requirement list -(what vanishes, what survives, what that demands of the operation) and name -the requirement a hint gave them. That method gets a card: if the deck is -built, rebuild it (`--replace`, or rerun the builder). 2026-10-20: three -replies of comfort before the method appeared. +A request for a method, coached like any rung. No reassurance, no verdict +on kuhy's ability, no "everyone memorises it" (2026-10-20: three replies of +comfort first). Say bluntly who supplied which step, never reframe kuhy's +diagnosis as the problem's fault, then have kuhy compare the `req:` list +with what the solution needed and name the requirement a hint gave them. +That method gets a card (rebuild the deck: `--replace` or rerun). ## 6. `reveal` — the only override @@ -216,12 +224,10 @@ not "just show me", not "I give up", not "what would you write". ## Model and client -This session must run on **sonnet** — haiku was the default until -2026-10-18, but its explanations went in circles (an Easy problem took ~30 -micro-questions, several unclear, one wrong correction). You cannot switch -the model yourself and `model:` frontmatter is ignored for skills (measured -2026-09-18). So: if you are not sonnet, the **first line** of your opening -message is +This session must run on **sonnet** (haiku, the default until 2026-10-18, +went in circles: ~30 micro-questions on an Easy). You cannot switch the +model and `model:` frontmatter is ignored for skills (measured 2026-09-18), +so if you are not sonnet, the **first line** of your opening message is > Type `/model sonnet` now — learning sessions are gated to sonnet; your next > message will be refused until you do (or send `keep model` to stay). @@ -229,22 +235,16 @@ message is then continue with the opening. `hooks/leetcode_model_gate_userprompt.sh` enforces it from the next prompt on. -The right client is `leetcode ` (the `leetcode` launcher in this -directory, symlinked from `~/.local/bin`, runs `leetcode_repl.py`): a -print-mode loop, already on sonnet (this directory is leetcode-guard's -`leetcode_coach` package, symlinked as the skill). It matters beyond cost — the Stop hook (`stop_gate.py` here, behind -the `hooks/leetcode_no_solution_stop.py` shim) sends back any -reply that hands kuhy a pasteable line, but in the interactive TUI that -reply has already streamed to the screen before the hook runs (seen -2026-09-21: the blocked `for` loop was fully visible above the rewrite). -Print mode waits for the hooks, so the REPL is the only client where the -gate actually withholds text. In the TUI it is best-effort. +The right client is `leetcode ` (this package's `leetcode` launcher, +symlinked from `~/.local/bin`): a print-mode loop on sonnet that also runs +the req-before-code gate and the session log (`leetcode report`). The Stop +hook (`stop_gate.py`) sends back a reply with a pasteable line or an early +toolbox name, but the TUI has already streamed it (2026-09-21: the blocked +`for` loop was visible above the rewrite); print mode waits. TUI: best-effort. **Offline / out of quota.** `leetcode --local ` coaches on -`qwen3:30b-a3b-instruct-2507` through Ollama (free, no network after the fetch), and the -REPL switches there by itself, history intact, when `claude -p` answers with -"You've hit your ... limit". The local path has no hooks, so -`local_coach.py` runs the Stop hook's detector (`gate_detector.py`) on every -reply before printing; three leaky drafts in a row are withheld. Everything -above `## Model and client` is its system prompt, minus section 1. -`python3 -m leetcode_coach.eval_local`: scripted acceptance run (two-sum). +`qwen3:30b-a3b-instruct-2507` via Ollama, and the REPL switches there by +itself, history intact, on "You've hit your ... limit". With no hooks there, +`local_coach.py` runs the same detectors on every draft; three leaky drafts +are withheld. Everything above `## Model and client` is its system prompt, +minus section 1. `python3 -m leetcode_coach.eval_local`: acceptance run. diff --git a/leetcode_coach/leetcode_coach/claude_cli.py b/leetcode_coach/leetcode_coach/claude_cli.py new file mode 100644 index 0000000..6390c2b --- /dev/null +++ b/leetcode_coach/leetcode_coach/claude_cli.py @@ -0,0 +1,76 @@ +"""The one `claude -p` call, shared by the coaching REPL and the req checker. + +The binary is looked up per call (``LEETCODE_CLAUDE_BIN``, else PATH), so the +E2E tests can point a running REPL at a fake. A binary that cannot start or +hangs past `timeout` comes back as an error reply, never an exception: the +checker treats it as "haiku did not answer" and falls back. + +The signal for a spent plan is the synthetic reply "You've hit your session +limit · resets 3pm (Europe/Warsaw)" (error rate_limit, HTTP 429), as recorded +in real transcripts on 2026-09-26. +""" + +from __future__ import annotations + +import json +import logging +import os +from pathlib import Path +import re +import shutil +import subprocess +from typing import Any + +_LOG = logging.getLogger(__name__) + +LIMIT_RE = re.compile(r"hit your (?:\w+ )?limit|usage limit reached", re.IGNORECASE) +RATE_LIMIT_STATUS = 429 + + +def claude_bin() -> str: + """``LEETCODE_CLAUDE_BIN``, else ``claude`` from PATH. + + PATH lookup, not a fixed path: claude lives in ~/.local/bin here, and the + old /usr/bin/claude default made every REPL turn crash (2026-10-20). + """ + return os.environ.get("LEETCODE_CLAUDE_BIN") or shutil.which("claude") or "claude" + + +def ask( + args: list[str], prompt: str, *, timeout: float | None = None +) -> dict[str, Any]: + """One `claude -p` turn; the JSON result, or a stand-in carrying stderr.""" + env = {**os.environ, "CLAUDE_TTY_FIX_OFF": "1"} + try: + proc = subprocess.run( + [claude_bin(), "-p", *args, "--output-format", "json", prompt], + capture_output=True, + stdin=subprocess.DEVNULL, # claude -p would otherwise drain REPL input + text=True, + env=env, + check=False, + cwd=Path.home(), + timeout=timeout, + ) + except (OSError, subprocess.TimeoutExpired) as exc: + _LOG.warning("claude -p could not run: %s", exc) + return {"result": f"[claude could not run: {exc}]", "_stderr": str(exc)} + reply: object + try: + reply = json.loads(proc.stdout, strict=False) if proc.stdout.strip() else {} + except json.JSONDecodeError as exc: + _LOG.warning("claude -p printed something that is not JSON: %s", exc) + reply = {} + result: dict[str, Any] = reply if isinstance(reply, dict) else {} + result["_stderr"] = proc.stderr.strip() + if proc.returncode != 0 and "result" not in result: + result["result"] = f"[claude exited {proc.returncode}]\n{proc.stderr.strip()}" + return result + + +def is_limit_hit(reply: dict[str, Any]) -> bool: + """Whether Claude refused this turn because the plan's usage limit is spent.""" + if reply.get("api_error_status") == RATE_LIMIT_STATUS: + return True + text = f"{reply.get('result', '')}\n{reply.get('_stderr', '')}" + return bool(LIMIT_RE.search(text)) diff --git a/leetcode_coach/leetcode_coach/leak.py b/leetcode_coach/leetcode_coach/leak.py new file mode 100644 index 0000000..506e774 --- /dev/null +++ b/leetcode_coach/leetcode_coach/leak.py @@ -0,0 +1,42 @@ +r"""Which toolbox entries a coach reply names (TODO-coach.md decision 7). + +Until kuhy's first code, a reply that names an operation from the toolbox -- +its name or any alias, e.g. "XOR", "^", "hash map" -- hands over step 3 of the +method (searching the toolbox by properties). Shared by the Stop hook +(`stop_gate`) and the offline coach (`local_coach`), so both judge alike. + +An entry kuhy (or the problem statement) already named is free for the whole +session, under every alias: after "is it XOR?" the honest answer has to be +able to say XOR, or the rewrite can never pass and the gate loops. + +Matching is case-insensitive on whole tokens. ``\b`` does not work next to a +symbol alias like ``^``, so the boundary is "no word character either side": +"a ^ b" matches, "2^3" (a power) and "xorshift" do not. +""" + +from __future__ import annotations + +import re +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from leetcode_coach.toolbox import Toolbox + + +def _pattern(name: str) -> re.Pattern[str]: + words = r"\s+".join(re.escape(w) for w in name.split()) + return re.compile(rf"(? bool: + return any(_pattern(n).search(text) for n in names if n.strip()) + + +def leaked_names(reply: str, allowed: list[str], box: Toolbox) -> list[str]: + """Entry names in `reply` that nothing in `allowed` already named.""" + seen = "\n".join(allowed) + return [ + e.name + for e in box.entries + if _named(reply, e.names()) and not _named(seen, e.names()) + ] diff --git a/leetcode_coach/leetcode_coach/leetcode_repl.py b/leetcode_coach/leetcode_coach/leetcode_repl.py index 0e17d16..a23f5ee 100755 --- a/leetcode_coach/leetcode_coach/leetcode_repl.py +++ b/leetcode_coach/leetcode_coach/leetcode_repl.py @@ -7,75 +7,34 @@ before the rewrite appears. `claude -p` returns nothing until the hooks are done, so this loop is the only client where the no-solution gate actually withholds text from the human. Each turn is one `claude -p --resume` call. -When Claude answers with its usage-limit message the session continues on the -local model (`local_session.py`) with the conversation so far; `--local` starts -there. The signal is the synthetic reply "You've hit your session limit · -resets 3pm (Europe/Warsaw)" (error rate_limit, HTTP 429), as recorded in real -transcripts on 2026-09-26. +Every block first passes the req-before-code phase gate (`phase.py`): code +sent before an accepted ``req:`` list is refused here and never reaches the +model. Replies lose their ```` marker (logged) before printing. + +When Claude answers with its usage-limit message (`claude_cli.is_limit_hit`) +the session continues on the local model (`local_session.py`) with the +conversation so far and the same gate; `--local` starts there. Input: a block ends on an empty line (paste code, then press Enter twice). `quit` or Ctrl-D leaves. `reveal` is passed through like any other message. +`leetcode report` prints the per-problem method report (`report.py`). Started by the `leetcode` launcher as `python3 -m leetcode_coach.leetcode_repl`; its prints ARE the session (T201 exemption in pyproject.toml). """ from __future__ import annotations -import json import logging import os -from pathlib import Path -import re -import shutil -import subprocess import sys -from typing import Any -from leetcode_coach import local_session +from leetcode_coach import local_session, phase, report +from leetcode_coach.claude_cli import ask, is_limit_hit # Named, not __name__: under `-m` this module is `__main__`, which would escape # the package logger's NullHandler and print the warning to the session. _LOG = logging.getLogger("leetcode_coach.leetcode_repl") -# PATH lookup, not a fixed path: claude lives in ~/.local/bin here, and the -# old /usr/bin/claude default made every REPL turn crash (2026-10-20). -CLAUDE = os.environ.get("LEETCODE_CLAUDE_BIN") or shutil.which("claude") or "claude" -LIMIT_RE = re.compile(r"hit your (?:\w+ )?limit|usage limit reached", re.IGNORECASE) -RATE_LIMIT_STATUS = 429 - - -def ask(args: list[str], prompt: str) -> dict[str, Any]: - """One `claude -p` turn; the JSON result, or a stand-in carrying stderr.""" - env = {**os.environ, "CLAUDE_TTY_FIX_OFF": "1"} - proc = subprocess.run( - [CLAUDE, "-p", *args, "--output-format", "json", prompt], - capture_output=True, - stdin=subprocess.DEVNULL, # claude -p would otherwise drain our REPL input - text=True, - env=env, - check=False, - cwd=Path.home(), - ) - reply: object - try: - reply = json.loads(proc.stdout, strict=False) if proc.stdout.strip() else {} - except json.JSONDecodeError as exc: - _LOG.warning("claude -p printed something that is not JSON: %s", exc) - reply = {} - result: dict[str, Any] = reply if isinstance(reply, dict) else {} - result["_stderr"] = proc.stderr.strip() - if proc.returncode != 0 and "result" not in result: - result["result"] = f"[claude exited {proc.returncode}]\n{proc.stderr.strip()}" - return result - - -def is_limit_hit(reply: dict[str, Any]) -> bool: - """Whether Claude refused this turn because the plan's usage limit is spent.""" - if reply.get("api_error_status") == RATE_LIMIT_STATUS: - return True - text = f"{reply.get('result', '')}\n{reply.get('_stderr', '')}" - return bool(LIMIT_RE.search(text)) - def read_block() -> str | None: """Lines until an empty one; None on EOF with nothing typed.""" @@ -96,15 +55,24 @@ def read_block() -> str | None: lines.append(line) +def _screen(gate: phase.PhaseGate, block: str) -> str | None: + """The block to send the coach, or None when the phase gate refuses it.""" + sent, note = gate.screen(block) + if note: + print(note) + return sent + + def run_claude(model: str, problem: list[str]) -> int: """The Claude session; hands over to the local model on a usage limit.""" + gate = phase.PhaseGate.open(problem[0]) opener = local_session.opening_message(problem[1:]) first = ask(["--model", model], "/leetcode " + " ".join(problem)) if is_limit_hit(first): print(f"[{first.get('result', 'usage limit')} -- switching to local model]") - return local_session.run(problem[0], [], opener, read_block) + return local_session.run(problem[0], [], opener, read_block, gate=gate) session = first.get("session_id") - print(first.get("result", "")) + print(gate.after_reply(str(first.get("result", "")))) if not session: return 1 history = [ @@ -115,30 +83,36 @@ def run_claude(model: str, problem: list[str]) -> int: block = read_block() if block is None or local_session.is_quit(block): return 0 + if (sent := _screen(gate, block)) is None: + continue print("\n...", flush=True) - reply = ask(["--model", model, "--resume", session], block) + reply = ask(["--model", model, "--resume", session], sent) if is_limit_hit(reply): print(f"[{reply.get('result', 'usage limit')} -- continuing locally]") - return local_session.run(problem[0], history, block, read_block) - print(reply.get("result", "")) + return local_session.run(problem[0], history, sent, read_block, gate=gate) + print(gate.after_reply(str(reply.get("result", "")))) history += [ - {"role": "user", "content": block}, + {"role": "user", "content": sent}, {"role": "assistant", "content": str(reply.get("result", ""))}, ] def main(argv: list[str]) -> int: - """Parse `[--local] [stuck on...]` and run a session.""" + """Parse `[--local] [stuck on...]` (or `report`) and run it.""" local = "--local" in argv args = [a for a in argv if a != "--local"] + if args[:1] == ["report"]: + return report.main(args[1:]) if not args or args[0] in {"-h", "--help"}: print("Usage: leetcode [--local] [what you are stuck on...]") print(" LEETCODE_MODEL=opus leetcode ... to use a stronger model") print(" --local coach on the local Ollama model (free, offline)") + print(" leetcode report req-before-code, max rung, reveal per problem") return 0 if local: opener = local_session.opening_message(args[1:]) - return local_session.run(args[0], [], opener, read_block) + gate = phase.PhaseGate.open(args[0]) + return local_session.run(args[0], [], opener, read_block, gate=gate) return run_claude(os.environ.get("LEETCODE_MODEL", "sonnet"), args) diff --git a/leetcode_coach/leetcode_coach/local_coach.py b/leetcode_coach/leetcode_coach/local_coach.py index ec4cae3..3630ea8 100644 --- a/leetcode_coach/leetcode_coach/local_coach.py +++ b/leetcode_coach/leetcode_coach/local_coach.py @@ -62,6 +62,8 @@ class LocalCoach: chat: Callable[[list[dict[str, str]]], local_backend.LocalReply] = ( local_backend.chat ) + leak_guard: Callable[[str, list[str]], list[str]] | None = None + """Toolbox entries a draft names too early (`PhaseGate.leaks`); off if None.""" @classmethod def start( @@ -95,14 +97,18 @@ class LocalCoach: total += reply.seconds display, payload = local_cards.split_cards(reply.text) bad = gate_findings(display, self.statement, self.user_texts()) - if not bad: - last_clean = reply.text - if payload is not None or local_prompt.ends_with_one_task(display): - return self._accept(reply.text, total, attempt) - reason = local_prompt.GATE_REASON_SHAPE - else: + named = self._leaks(display) if not bad else [] + if bad: snippets = "; ".join(f"`{b}`" for b in bad) reason = local_prompt.GATE_REASON_CODE.format(snippets=snippets) + elif named: + reason = local_prompt.GATE_REASON_LEAK.format(names=", ".join(named)) + else: + last_clean = reply.text + prose, _rung = local_prompt.split_rung(display) + if payload is not None or local_prompt.ends_with_one_task(prose): + return self._accept(reply.text, total, attempt) + reason = local_prompt.GATE_REASON_SHAPE scratch += [ {"role": "assistant", "content": reply.text}, {"role": "user", "content": reason}, @@ -112,6 +118,12 @@ class LocalCoach: self.history.append({"role": "assistant", "content": local_prompt.WITHHELD}) return Turn(local_prompt.WITHHELD, total, MAX_TRIES, withheld=True) + def _leaks(self, display: str) -> list[str]: + """Entry names the draft gives away, per the session's leak guard.""" + if self.leak_guard is None: + return [] + return self.leak_guard(display, self.user_texts()) + def _accept(self, text: str, seconds: float, attempts: int) -> Turn: """Record `text` as the assistant's reply and wrap it as a Turn.""" self.history.append({"role": "assistant", "content": text}) diff --git a/leetcode_coach/leetcode_coach/local_prompt.py b/leetcode_coach/leetcode_coach/local_prompt.py index 8c9a013..c235589 100644 --- a/leetcode_coach/leetcode_coach/local_prompt.py +++ b/leetcode_coach/leetcode_coach/local_prompt.py @@ -30,12 +30,12 @@ LOCAL_NOTES = """ - Write plain markdown for a terminal. No emoji. - Never write code that uses the starter's function or parameter names unless kuhy has typed `reveal`. A client-side gate rejects such replies. -- Never name the data structure or algorithm that solves the problem before - rung H3 -- not even inside a calibration question. Ask about what kuhy - already said instead. +- Never name an operation, data structure or algorithm before kuhy has + written code (unless kuhy named it first) -- not even inside a calibration + question. Ask about the properties kuhy listed instead. - The LAST line of every reply is exactly one question or one instruction for - kuhy. Nothing after it. ONE question mark in the whole closing paragraph: - never stack questions ("What happens? Why?"). + kuhy; only the `` line may follow it. ONE question mark in + the closing paragraph: never stack questions ("What happens? Why?"). - Never state the key insight, the fix, or what the code "must" do. On a buggy paste, describe what the data does on one concrete input from the examples ("with [3,2,4] and target 6, at i=0 `seen` already holds 3 when the lookup @@ -59,11 +59,18 @@ GATE_REASON_CODE = ( "Rewrite it without that snippet: describe the data or " "name the construct, and keep the one question at the end." ) +GATE_REASON_LEAK = ( + "leetcode gate: your reply names {names} before kuhy has written any code. " + "Rewrite it without naming an operation or structure: ask about the " + "properties the operation needs, and keep the one question at the end." +) GATE_REASON_SHAPE = ( "leetcode gate: end with exactly ONE question or ONE " "instruction -- your closing paragraph asks more than one " "thing or ends on neither. Rewrite it." ) +RUNG_RE = re.compile(r"") +"""The coach's trailing hint-rung marker (SKILL.md section 3); logged, never shown.""" IMPERATIVE = re.compile( r"^(try|write|predict|explain|tell|describe|trace|run|paste|list|sketch|" r"walk|show|give|say|name|work|think|check|change|fix|look|answer|type|" @@ -99,3 +106,11 @@ def ends_with_one_task(text: str) -> bool: sentences = [x for x in re.split(r"(?<=[.?!])\s+", last) if x.strip()] final = sentences[-1].strip() return final.endswith("?") or bool(IMPERATIVE.match(final)) + + +def split_rung(text: str) -> tuple[str, str | None]: + """`text` without its ```` markers, and the last rung named.""" + rungs = RUNG_RE.findall(text) + if not rungs: + return text, None + return RUNG_RE.sub("", text).strip(), rungs[-1] diff --git a/leetcode_coach/leetcode_coach/local_session.py b/leetcode_coach/leetcode_coach/local_session.py index e7a8503..d1906d0 100644 --- a/leetcode_coach/leetcode_coach/local_session.py +++ b/leetcode_coach/leetcode_coach/local_session.py @@ -19,6 +19,8 @@ from leetcode_coach import local_backend, local_cards, local_coach if TYPE_CHECKING: from collections.abc import Callable + from leetcode_coach import phase + _LOG = logging.getLogger(__name__) CARD_RETRIES = 2 @@ -38,18 +40,33 @@ def is_quit(block: str) -> bool: return block.strip().lower() in {"quit", "exit"} +def _run_fetcher( + problem: str, runner: Callable[..., subprocess.CompletedProcess[str]] +) -> subprocess.CompletedProcess[str]: + return runner( + [sys.executable, str(local_coach.SKILL_DIR / "fetch_problem.py"), problem], + capture_output=True, + text=True, + check=False, + ) + + +def fetched_statement( + problem: str, + runner: Callable[..., subprocess.CompletedProcess[str]] = subprocess.run, +) -> str | None: + """Statement + starter code, or None when it cannot be fetched (no prompt).""" + proc = _run_fetcher(problem, runner) + return proc.stdout if proc.returncode == 0 else None + + def fetch_statement( problem: str, read: Callable[[], str | None], runner: Callable[..., subprocess.CompletedProcess[str]] = subprocess.run, ) -> str | None: """Statement + starter code, or whatever kuhy pastes when it cannot be fetched.""" - proc = runner( - [sys.executable, str(local_coach.SKILL_DIR / "fetch_problem.py"), problem], - capture_output=True, - text=True, - check=False, - ) + proc = _run_fetcher(problem, runner) if proc.returncode == 0: return proc.stdout why = "premium or unknown problem" if proc.returncode == FETCH_PREMIUM else "" @@ -72,11 +89,15 @@ def confirm_gpu(ask: Callable[[str], str] = input) -> bool: return answer.strip().lower() in {"y", "yes"} -def show(turn: local_coach.Turn, coach: local_coach.LocalCoach) -> None: +def show( + turn: local_coach.Turn, + coach: local_coach.LocalCoach, + gate: phase.PhaseGate | None = None, +) -> None: """Print one reply; build its cards deck, feeding rejections back.""" for _ in range(CARD_RETRIES + 1): display, payload = local_cards.split_cards(turn.text) - print(display) + print(gate.after_reply(display) if gate else display) if payload is None: return ok, out = local_cards.deliver_cards(payload) @@ -90,13 +111,28 @@ def show(turn: local_coach.Turn, coach: local_coach.LocalCoach) -> None: ) +def _screen(gate: phase.PhaseGate | None, block: str) -> str | None: + """The block to send the coach (None when the phase gate refuses it).""" + if gate is None: + return block + sent, note = gate.screen(block) + if note: + print(note) + return sent + + def run( problem: str, history: list[dict[str, str]], first: str, read: Callable[[], str | None], + gate: phase.PhaseGate | None = None, ) -> int: - """Coach locally until kuhy quits. `first` is sent before reading input.""" + """Coach locally until kuhy quits. `first` is sent before reading input. + + With a `gate`, every block passes the req-before-code phase gate first and + every reply the toolbox leak check (`first` was screened by the caller). + """ statement = fetch_statement(problem, read) if not statement: return 1 @@ -105,11 +141,16 @@ def run( print("[not loading the local model]") return 1 coach = local_coach.LocalCoach.start(statement, history) + if gate is not None: + gate.known = statement + coach.leak_guard = gate.leaks print(f"[local: {local_backend.MODEL} via Ollama]\n\n...", flush=True) - show(coach.respond(first), coach) + show(coach.respond(first), coach, gate) while (block := read()) is not None and not is_quit(block): + if (sent := _screen(gate, block)) is None: + continue print("\n...", flush=True) - show(coach.respond(block), coach) + show(coach.respond(sent), coach, gate) except (urllib.error.URLError, ConnectionError) as exc: _LOG.warning("Ollama is not reachable: %s", exc) print(f"[Ollama is not reachable at {local_backend.OLLAMA_URL}: {exc}]") diff --git a/leetcode_coach/leetcode_coach/phase.py b/leetcode_coach/leetcode_coach/phase.py new file mode 100644 index 0000000..8f5dffa --- /dev/null +++ b/leetcode_coach/leetcode_coach/phase.py @@ -0,0 +1,191 @@ +"""The req-before-code phase gate, run by the REPL on both paths. + +TODO-coach.md decisions 4-6: a message starting ``req:`` is kuhy's +requirement list. Until one is accepted, a block that is code is refused +here, with a one-line reason, and never reaches the coach model. A list is +accepted when the checker passes it or cannot judge it (fail open); after the +3rd list that still comes back incomplete, code is let through anyway and its +``code`` event carries ``req_incomplete: true``. + +State is derived from the session log, not held in memory, so it survives the +mid-session hand-over to the local model and a restarted REPL the same day. +The ``code`` event is appended BEFORE the block is sent: the Stop hook judges +the reply to that very block and reads the log to know code exists. + +Is a block code? A regex decides the clear cases; only a block it cannot +decide goes to the model (`req_check.looks_like_code`), and an unanswered +question is "not code" -- the gate never refuses kuhy on a guess. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +import os +import re +from typing import TYPE_CHECKING, Any + +from leetcode_coach import ( + leak, + local_prompt, + local_session, + req_check, + session_log, + toolbox, +) + +if TYPE_CHECKING: + from collections.abc import Callable + from pathlib import Path + +MAX_LISTS = 3 +REQ_RE = re.compile(r"^\s*req\s*:", re.IGNORECASE) +STRONG_RE = re.compile( + r"```|=>|^\s*(?:function\b|def\s+\w+\s*\(|class\s+\w+|" + r"(?:const|let|var)\s+\w+\s*=|(?:for|while|if)\s*\(|#include|import\s+\w)" + r"|[;{]\s*$|^\s*\}", + re.MULTILINE, +) +OPERATOR_RE = re.compile( + r"\w\s*(?:[=!]==?|<=|>=|[-+*/%^&|]=|\^|&&|\|\||<<|>>|%|=)\s*\w|\w\(|\w\[" +) + + +def regex_verdict(block: str) -> bool | None: + """True for clear code, False for clear prose, None when it cannot tell.""" + if REQ_RE.match(block): + return False + if STRONG_RE.search(block): + return True + return None if OPERATOR_RE.search(block) else False + + +@dataclass(frozen=True) +class ReqState: + """Where the requirement phase stands, derived from the log.""" + + accepted: bool + forced: bool + tries: int + + @property + def open(self) -> bool: + """Whether code may reach the coach.""" + return self.accepted or self.forced + + +def req_state(events: list[dict[str, Any]]) -> ReqState: + """Accepted on a pass/unverified check; forced after `MAX_LISTS` incompletes.""" + verdicts = [e.get("verdict") for e in events if e.get("event") == "req_check"] + tries = verdicts.count("incomplete") + accepted = "pass" in verdicts or "unverified" in verdicts + return ReqState(accepted, not accepted and tries >= MAX_LISTS, tries) + + +def _req_note(result: req_check.Check, state: ReqState, box: toolbox.Toolbox) -> str: + if result.verdict == "pass": + return "[req list accepted -- code is open]" + if result.verdict == "unverified": + return "[req check unavailable -- continuing unverified, code is open]" + if state.forced: + missing = ", ".join(result.missing) + return f"[req list still incomplete ({missing}) -- code accepted anyway]" + props = "\n".join(f" - {p}: {box.properties[p]}" for p in result.missing) + return ( + f"[req list incomplete -- it does not yet say these properties:\n{props}\n" + f" revise with req: ({state.tries} of {MAX_LISTS})]" + ) + + +@dataclass +class PhaseGate: + """One problem's gate: its log, toolbox and (lazily fetched) statement.""" + + log: Path + toolbox: toolbox.Toolbox + source: Callable[[], str] + ask: Callable[..., dict[str, Any]] | None = None + known: str | None = field(default=None, repr=False) + + @classmethod + def open(cls, problem: str) -> PhaseGate: + """The gate for `problem`; exports its log path for the Stop hook.""" + log = session_log.log_path(problem) + os.environ[session_log.LOG_ENV] = str(log) + + def source() -> str: + return local_session.fetched_statement(problem) or session_log.slug(problem) + + return cls(log, toolbox.load(), source) + + def statement(self) -> str: + """The problem statement, fetched once on first use.""" + if self.known is None: + self.known = self.source() + return self.known + + def code_seen(self) -> bool: + """Whether a code block has been let through this session.""" + return session_log.has_code(self.log) + + def screen(self, block: str) -> tuple[str | None, str]: + """(what to send the coach, or None to refuse; a note for kuhy).""" + text = block.strip() + if text.lower() == "reveal": + session_log.append(self.log, "reveal") + return block, "" + if REQ_RE.match(text): + return self._req(block) + verdict = regex_verdict(text) + if verdict is None: + verdict = req_check.looks_like_code(text, self.ask) + if not verdict: + return block, "" + state = req_state(session_log.read(self.log)) + if not state.open: + return None, self._refusal(state) + extra = {"req_incomplete": True} if state.forced else {} + session_log.append(self.log, "code", **extra) + return block, "" + + def _req(self, block: str) -> tuple[str, str]: + session_log.append(self.log, "req", text=block.strip()) + result = req_check.check(self.statement(), block, self.toolbox, self.ask) + session_log.append( + self.log, + "req_check", + verdict=result.verdict, + missing=list(result.missing), + via=result.via, + ) + state = req_state(session_log.read(self.log)) + tag = result.verdict + if result.missing: + tag += "; missing properties: " + ", ".join(result.missing) + return f"{block}\n\n[req check: {tag}]", _req_note(result, state, self.toolbox) + + @staticmethod + def _refusal(state: ReqState) -> str: + if not state.tries: + return ( + "[code held back: write the requirement list first -- one message " + "starting with req: that lists the properties the operation must " + "have, without naming an operation or structure]" + ) + return ( + "[code held back: the req list is still incomplete -- revise it " + f"with req: ({state.tries} of {MAX_LISTS} so far)]" + ) + + def after_reply(self, text: str) -> str: + """Log the reply's rung marker and return the reply without it.""" + clean, rung = local_prompt.split_rung(text) + if rung is not None: + session_log.append(self.log, "rung", rung=rung) + return clean + + def leaks(self, reply: str, user_texts: list[str]) -> list[str]: + """Toolbox entries `reply` names too early (the local path's bounce).""" + if any(t.strip() == "reveal" for t in user_texts) or self.code_seen(): + return [] + allowed = [*user_texts, self.statement()] + return leak.leaked_names(reply, allowed, self.toolbox) diff --git a/leetcode_coach/leetcode_coach/report.py b/leetcode_coach/leetcode_coach/report.py new file mode 100644 index 0000000..e30d304 --- /dev/null +++ b/leetcode_coach/leetcode_coach/report.py @@ -0,0 +1,69 @@ +"""`leetcode report`: one row per session file, for kuhy to read the method back. + +Columns: req-before-code (``y``/``n``, ``-`` while there is no code yet), +the highest hint rung, whether ``reveal`` was typed, and every ``req_check`` +verdict in order -- all of them, because "incomplete,incomplete,pass" is the +story, where the status app's reader keeps only the last one. Its stdout IS +its interface (T201 exemption in pyproject.toml); tests assert on it. +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING, Any + +from leetcode_coach import session_log + +if TYPE_CHECKING: + from pathlib import Path + +RUNGS = ("H0", "H1", "H2", "H3") +WIDTHS = (36, 17, 10, 8) +HEADER = ("problem", "req-before-code", "max-rung", "reveal", "req_check") + + +def _line(cols: tuple[str, ...]) -> str: + padded = [c.ljust(w) for c, w in zip(cols, WIDTHS, strict=False)] + return "".join(padded) + cols[-1] + + +def _req_before_code(events: list[dict[str, Any]]) -> str: + seen_req = False + for event in events: + if event.get("event") == "req": + seen_req = True + elif event.get("event") == "code": + return "y" if seen_req else "n" + return "-" + + +def row(path: Path) -> str: + """The report line for one session file.""" + events = session_log.read(path) + rungs = [ + RUNGS.index(e["rung"]) + for e in events + if e.get("event") == "rung" and e.get("rung") in RUNGS + ] + verdicts = [str(e.get("verdict")) for e in events if e.get("event") == "req_check"] + return _line( + ( + path.stem, + _req_before_code(events), + RUNGS[max(rungs)] if rungs else "-", + "y" if any(e.get("event") == "reveal" for e in events) else "n", + ",".join(verdicts) or "-", + ) + ) + + +def main(_argv: list[str]) -> int: + """Print the report for every session file.""" + folder = session_log.sessions_dir() + paths = sorted(folder.glob("*.jsonl")) + if not paths: + print(f"no coach sessions yet in {folder}") + return 0 + print(_line(HEADER)) + for path in paths: + print(row(path)) + return 0 diff --git a/leetcode_coach/leetcode_coach/req_check.py b/leetcode_coach/leetcode_coach/req_check.py new file mode 100644 index 0000000..4a45cbb --- /dev/null +++ b/leetcode_coach/leetcode_coach/req_check.py @@ -0,0 +1,174 @@ +"""Judge kuhy's requirement list, and ambiguous code, on a cheap model. + +TODO-coach.md decision 5: a separate ``claude -p --model haiku`` call (the +coaching session stays on sonnet), Ollama through `local_backend` when haiku +is out of quota, errors or answers something that is not the JSON asked for, +and when neither answers the gate fails OPEN: ``unverified``, kuhy continues. + +The checker may only name *missing properties* from the toolbox vocabulary. +That is enforced here, not trusted to the prompt: every ``missing`` item that +is not a vocabulary key is dropped (an entry name like "XOR" is a leak, and an +invented property is noise). An ``incomplete`` whose every item was dropped +names nothing kuhy could fix, so it counts as no answer and falls through. + +The haiku call runs with no tools, no MCP, no hooks and no session file: the +global CLAUDE.md still loads (no OAuth-compatible way to skip it), which is +why the answer is extracted and validated rather than parsed whole. +""" + +from __future__ import annotations + +from dataclasses import dataclass +import json +import logging +from typing import TYPE_CHECKING, Any +import urllib.error + +from leetcode_coach import claude_cli, local_backend + +if TYPE_CHECKING: + from collections.abc import Callable + + from leetcode_coach.toolbox import Toolbox + +_LOG = logging.getLogger(__name__) + +HAIKU_TIMEOUT_S = 50.0 +HAIKU_ARGS = [ + "--model", + "haiku", + "--tools", + "", + "--strict-mcp-config", + "--no-session-persistence", + "--disable-slash-commands", + "--settings", + '{"disableAllHooks": true}', +] +CHECK_SYSTEM = ( + "You check a student's requirement list for a programming problem. You are " + "given the problem, the list, and a vocabulary of properties an operation " + "can have. Decide which vocabulary properties the operation that solves " + "the problem must have, and whether the list states each of them in any " + "words. Reply with ONE JSON object and nothing else: " + '{"verdict": "pass" | "incomplete", "missing": []}. ' + "Never name an operation, a data structure or an algorithm; only the " + "property names from the vocabulary may appear in missing." +) +CODE_SYSTEM = ( + "You decide whether a chat message is program source code (a line or " + "block someone would paste into a solution) or prose (a question or an " + 'idea, even one that mentions a symbol). Reply with ONE JSON object: {"code": ' + 'true} or {"code": false}.' +) + + +@dataclass(frozen=True) +class Check: + """A checker answer: verdict, the missing properties, and who answered.""" + + verdict: str + missing: tuple[str, ...] = () + via: str = "none" + + +def extract_json(text: str) -> dict[str, Any] | None: + """The JSON object starting at the first ``{`` (fences and chatter ignored).""" + start = text.find("{") + if start == -1: + return None + try: + obj, _end = json.JSONDecoder().raw_decode(text, start) + except ValueError as exc: + _LOG.warning("checker answer is not a JSON object: %s", exc) + return None + return obj if isinstance(obj, dict) else None + + +def parse_verdict(obj: object, vocabulary: dict[str, str]) -> Check | None: + """A validated verdict, or None when the answer is unusable.""" + if not isinstance(obj, dict) or obj.get("verdict") not in {"pass", "incomplete"}: + return None + missing = obj.get("missing", []) + if not isinstance(missing, list): + return None + kept = tuple(m for m in missing if isinstance(m, str) and m in vocabulary) + if obj["verdict"] == "pass": + return Check("pass") + return Check("incomplete", kept) if kept else None + + +def check_prompt(statement: str, req: str, box: Toolbox) -> str: + """The checker's user message: problem, list, vocabulary, toolbox.""" + vocab = "\n".join( + f"- {name}: {meaning}" for name, meaning in box.properties.items() + ) + toolbox = "\n".join( + f"- {', '.join(e.properties)}" for e in box.entries if e.properties + ) + return ( + f"## Problem\n{statement.strip()}\n\n## Requirement list\n{req.strip()}\n\n" + f"## Property vocabulary\n{vocab}\n\n" + f"## Property sets the known operations offer\n{toolbox}\n" + ) + + +def _ask_models[T]( + system: str, + prompt: str, + validate: Callable[[dict[str, Any] | None], T | None], + ask: Callable[..., dict[str, Any]] | None, +) -> tuple[T, str] | None: + """Haiku, then Ollama; the first validated answer and who gave it.""" + asker = claude_cli.ask if ask is None else ask + reply = asker( + [*HAIKU_ARGS, "--system-prompt", system], prompt, timeout=HAIKU_TIMEOUT_S + ) + if not claude_cli.is_limit_hit(reply): + got = validate(extract_json(str(reply.get("result", "")))) + if got is not None: + return got, "haiku" + try: + local = local_backend.chat( + [{"role": "system", "content": system}, {"role": "user", "content": prompt}] + ) + except (urllib.error.URLError, OSError, local_backend.ContextOverflowError) as exc: + _LOG.warning("Ollama could not answer the check either: %s", exc) + return None + got = validate(extract_json(local.text)) + return None if got is None else (got, "ollama") + + +def check( + statement: str, + req: str, + box: Toolbox, + ask: Callable[..., dict[str, Any]] | None = None, +) -> Check: + """Judge `req` against the problem; ``unverified`` when nobody can.""" + if not box.properties: + _LOG.warning("no property vocabulary: the req list stays unverified") + return Check("unverified") + got = _ask_models( + CHECK_SYSTEM, + check_prompt(statement, req, box), + lambda obj: parse_verdict(obj, box.properties), + ask, + ) + if got is None: + return Check("unverified") + verdict, via = got + return Check(verdict.verdict, verdict.missing, via) + + +def _code_flag(obj: dict[str, Any] | None) -> bool | None: + flag = None if obj is None else obj.get("code") + return flag if isinstance(flag, bool) else None + + +def looks_like_code( + block: str, ask: Callable[..., dict[str, Any]] | None = None +) -> bool: + """The model's call on a block the regex could not decide; prose if unsure.""" + got = _ask_models(CODE_SYSTEM, block, _code_flag, ask) + return False if got is None else got[0] diff --git a/leetcode_coach/leetcode_coach/session_log.py b/leetcode_coach/leetcode_coach/session_log.py new file mode 100644 index 0000000..2921a56 --- /dev/null +++ b/leetcode_coach/leetcode_coach/session_log.py @@ -0,0 +1,97 @@ +"""The coach's per-problem session log (TODO-coach.md decision 8). + +One JSONL file per problem and day, +``~/.local/share/leetcode-coach/sessions/-.jsonl`` (the +directory follows ``LEETCODE_COACH_SESSIONS`` when set, for tests). Every line +is one object with ``event`` and ``ts``; ``req_check`` carries ``verdict``, +``rung`` carries ``rung``. That contract is FIXED: ``leetcode_guard._coach_log`` +reads it for the status app, so a new kind of fact rides as an extra field on +an existing event (``code`` + ``req_incomplete``), never as a new event name. + +Appends are one ``os.write`` on an ``O_APPEND`` descriptor, so a concurrent +reader sees whole lines, and two writers (REPL and a hook) cannot interleave. +""" + +from __future__ import annotations + +from datetime import UTC, date, datetime +import json +import logging +import os +from pathlib import Path +import re +from typing import Any + +from leetcode_coach import fetch_problem + +_LOG = logging.getLogger(__name__) + +SESSIONS_ENV = "LEETCODE_COACH_SESSIONS" +LOG_ENV = "LEETCODE_COACH_LOG" +"""Set by the REPL so the Stop hook (a grandchild) reads the same log.""" +DEFAULT_DIR = Path.home() / ".local" / "share" / "leetcode-coach" / "sessions" +EVENTS = frozenset( + {"req", "req_check", "code", "rung", "reveal", "postmortem", "retest"} +) + + +def sessions_dir() -> Path: + """Where the session files live.""" + return Path(os.environ.get(SESSIONS_ENV) or DEFAULT_DIR) + + +def slug(problem: str) -> str: + """The LeetCode slug in `problem`, or a filesystem-safe stand-in.""" + try: + return fetch_problem.slug_from(problem) + except ValueError as exc: + _LOG.warning("%s; naming the log after a sanitised copy", exc) + return re.sub(r"[^a-z0-9]+", "-", problem.lower()).strip("-") or "problem" + + +def log_path(problem: str, day: date | None = None) -> Path: + """The log file for `problem` (URL or slug) on `day` (default today).""" + when = day or datetime.now(tz=UTC).astimezone().date() + return sessions_dir() / f"{when.isoformat()}-{slug(problem)}.jsonl" + + +def append(path: Path, event: str, **fields: object) -> None: + """Append one event line atomically. + + Raises: + ValueError: for an event name outside the fixed contract. + """ + if event not in EVENTS: + msg = f"{event!r} is not a session-log event; add a field instead" + raise ValueError(msg) + stamp = datetime.now(tz=UTC).astimezone().isoformat(timespec="seconds") + record = {"event": event, "ts": stamp, **fields} + line = (json.dumps(record, ensure_ascii=False) + "\n").encode() + path.parent.mkdir(parents=True, exist_ok=True) + fd = os.open(path, os.O_WRONLY | os.O_APPEND | os.O_CREAT, 0o644) + try: + os.write(fd, line) + finally: + os.close(fd) + + +def _parse(line: str) -> object: + try: + return json.loads(line) + except ValueError as exc: + _LOG.warning("session log line is not JSON, skipped: %s", exc) + return None + + +def read(path: Path) -> list[dict[str, Any]]: + """Every object line in `path`; nothing when it does not exist yet.""" + if not path.is_file(): + return [] + lines = path.read_text(encoding="utf-8").splitlines() + parsed = (_parse(line) for line in lines if line.strip()) + return [obj for obj in parsed if isinstance(obj, dict)] + + +def has_code(path: Path) -> bool: + """Whether kuhy's code has reached the coach in this session.""" + return any(e.get("event") == "code" for e in read(path)) diff --git a/leetcode_coach/leetcode_coach/stop_gate.py b/leetcode_coach/leetcode_coach/stop_gate.py index 8a82d1f..72b2868 100644 --- a/leetcode_coach/leetcode_coach/stop_gate.py +++ b/leetcode_coach/leetcode_coach/stop_gate.py @@ -1,5 +1,8 @@ """Stop hook: in a /leetcode session, refuse a reply that hands kuhy a line of code. +Before kuhy's first code it also bounces a reply that names a toolbox entry +(`leak`), except an entry kuhy or the fetched problem already named. + The prose rule ("never write a line for their code") held in one haiku role-play out of two, so it is adjudicated here instead. A reply is blocked when any code span or fenced block in it uses an identifier from the @@ -19,10 +22,18 @@ IS the hook's interface (hence the T201 exemption in pyproject.toml). from __future__ import annotations import json +import os from pathlib import Path import sys -from leetcode_coach import gate_detector +from leetcode_coach import ( + gate_detector, + leak, + local_prompt, + phase, + session_log, + toolbox, +) SKILL_LINK = Path.home() / ".claude" / "skills" / "leetcode" # A /leetcode slash command, or a model-invoked Skill(leetcode) load (which @@ -49,6 +60,11 @@ def verdict(raw: str, reply: str) -> str | None: users = [gate_detector.kuhy_text(e) for e in entries if e.get("type") == "user"] if any(u.strip() == "reveal" for u in users): return None + if not _code_seen(users): + quoted = [gate_detector.allowed_text(entries)] + named = leak.leaked_names(reply, quoted, toolbox.load()) + if named: + return local_prompt.GATE_REASON_LEAK.format(names=", ".join(named)) idents = gate_detector.starter_identifiers(entries) if not idents: return None @@ -64,6 +80,19 @@ def verdict(raw: str, reply: str) -> str | None: ) +def _code_seen(users: list[str]) -> bool: + """Whether kuhy's code has reached the coach: the REPL's log says so. + + The REPL exports the log path (`session_log.LOG_ENV`). The TUI has no REPL + and no log, so there a code block among kuhy's own turns stands in for the + ``code`` event (regex only: a hook must not wait on a model). + """ + log = os.environ.get(session_log.LOG_ENV) + if log: + return session_log.has_code(Path(log)) + return any(phase.regex_verdict(u.strip()) is True for u in users) + + def blocks_so_far(session_id: str, prompt_id: str) -> int: """How many times this prompt's reply was already sent back. diff --git a/leetcode_coach/leetcode_coach/tests/_fake_claude.py b/leetcode_coach/leetcode_coach/tests/_fake_claude.py new file mode 100755 index 0000000..4975307 --- /dev/null +++ b/leetcode_coach/leetcode_coach/tests/_fake_claude.py @@ -0,0 +1,42 @@ +#!/usr/bin/env python3 +"""A scripted stand-in for `claude -p`, reached through LEETCODE_CLAUDE_BIN. + +Reads ``$FAKE_CLAUDE_DIR/script.json``: ``{"haiku": [...], "coach": [...]}``. +A call made with ``--model haiku`` is the checker; any other is the coach. +Each call takes the next answer of its kind -- ``{"result": "..."}`` is printed +the way ``--output-format json`` prints it, ``{"exit": n, "stderr": "..."}`` +fails -- and appends ``{"kind", "args", "prompt"}`` to ``calls.jsonl`` so a +test can prove what did and did not reach the model. +""" + +from __future__ import annotations + +import json +import os +from pathlib import Path +import sys + + +def main() -> int: + """Answer one scripted call.""" + root = Path(os.environ["FAKE_CLAUDE_DIR"]) + args = sys.argv[1:] + model = args[args.index("--model") + 1] if "--model" in args else "" + kind = "haiku" if model == "haiku" else "coach" + with (root / "calls.jsonl").open("a", encoding="utf-8") as fh: + record = {"kind": kind, "args": args[:-1], "prompt": args[-1]} + fh.write(json.dumps(record) + "\n") + counter = root / f"{kind}.used" + used = int(counter.read_text(encoding="utf-8")) if counter.exists() else 0 + counter.write_text(str(used + 1), encoding="utf-8") + script = json.loads((root / "script.json").read_text(encoding="utf-8")) + answer = script[kind][used] + if "exit" in answer: + sys.stderr.write(answer.get("stderr", "")) + return int(answer["exit"]) + sys.stdout.write(json.dumps({"result": answer["result"], "session_id": "fake"})) + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/leetcode_coach/leetcode_coach/tests/_fake_ollama.py b/leetcode_coach/leetcode_coach/tests/_fake_ollama.py new file mode 100644 index 0000000..e293906 --- /dev/null +++ b/leetcode_coach/leetcode_coach/tests/_fake_ollama.py @@ -0,0 +1,77 @@ +"""A scripted Ollama on a free localhost port, for the coach E2E tests. + +Speaks the two endpoints `local_backend` uses: ``GET /api/ps`` (the model is +always "loaded", so no nvidia-smi probe runs) and ``POST /api/chat``. A chat +whose system prompt is the requirement checker's is answered from the +``checker`` queue; every other chat is the coach and comes from ``coach``. +""" + +from __future__ import annotations + +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +import json +import threading +from typing import Any + +from leetcode_coach import local_backend, req_check + +Seen = list[tuple[str, list[dict[str, str]]]] + + +class _Server(ThreadingHTTPServer): + """Holds the scripted answers and a record of every chat request.""" + + queues: dict[str, list[str]] + seen: Seen + + +class _Handler(BaseHTTPRequestHandler): + def _owner(self) -> _Server: + assert isinstance(self.server, _Server) + return self.server + + def _send(self, body: dict[str, Any]) -> None: + data = json.dumps(body).encode() + self.send_response(200) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(data))) + self.end_headers() + self.wfile.write(data) + + def do_GET(self) -> None: + """`/api/ps`: the model is resident.""" + self._send({"models": [{"name": local_backend.MODEL}]}) + + def do_POST(self) -> None: + """`/api/chat`: the next scripted answer for this kind of call.""" + length = int(self.headers["Content-Length"]) + messages = json.loads(self.rfile.read(length))["messages"] + system = messages[0]["content"] if messages else "" + kind = "checker" if system == req_check.CHECK_SYSTEM else "coach" + owner = self._owner() + owner.seen.append((kind, messages)) + text = owner.queues[kind].pop(0) + self._send({"message": {"content": text}, "prompt_eval_count": 50}) + + +class FakeOllama: + """Start with scripted queues; `url` goes into `local_backend.OLLAMA_URL`.""" + + def __init__(self, checker: list[str], coach: list[str]) -> None: + self.server = _Server(("127.0.0.1", 0), _Handler) + self.server.queues = {"checker": list(checker), "coach": list(coach)} + self.server.seen = [] + self.thread = threading.Thread(target=self.server.serve_forever, daemon=True) + self.thread.start() + self.url = f"http://127.0.0.1:{self.server.server_address[1]}" + + @property + def seen(self) -> Seen: + """Every chat request so far, tagged checker/coach.""" + return self.server.seen + + def close(self) -> None: + """Stop serving and release the socket.""" + self.server.shutdown() + self.server.server_close() + self.thread.join() diff --git a/leetcode_coach/leetcode_coach/tests/_shared.py b/leetcode_coach/leetcode_coach/tests/_shared.py index 5fe57a1..83a3bc5 100644 --- a/leetcode_coach/leetcode_coach/tests/_shared.py +++ b/leetcode_coach/leetcode_coach/tests/_shared.py @@ -2,11 +2,14 @@ from __future__ import annotations +from pathlib import Path import subprocess from typing import Any from leetcode_coach import local_backend +FIXTURE_TOOLBOX = Path(__file__).with_name("_toolbox_fixture.json") +"""Three entries and six properties: the toolbox every coach test runs on.""" STATEMENT = """# 1. Two Sum (Easy) Return indices of two numbers adding up to target. diff --git a/leetcode_coach/leetcode_coach/tests/_toolbox_fixture.json b/leetcode_coach/leetcode_coach/tests/_toolbox_fixture.json new file mode 100644 index 0000000..d0ef8cc --- /dev/null +++ b/leetcode_coach/leetcode_coach/tests/_toolbox_fixture.json @@ -0,0 +1,27 @@ +{ + "entries": [ + { + "name": "XOR", + "aliases": ["^", "exclusive or", "bitwise xor"], + "properties": ["self-inverse", "commutative", "associative"] + }, + { + "name": "hash map", + "aliases": ["hashmap", "dictionary"], + "properties": ["constant-time-lookup"] + }, + { + "name": "sum formula", + "aliases": ["Gauss sum"], + "properties": ["constant-space", "closed-form-total"] + } + ], + "properties": { + "self-inverse": "applying it twice with the same value undoes it", + "commutative": "the order of the operands does not change the result", + "associative": "the grouping of the operands does not change the result", + "constant-time-lookup": "answers 'have I seen this?' without scanning", + "constant-space": "memory does not grow with the input", + "closed-form-total": "the expected total is known without iterating" + } +} diff --git a/leetcode_coach/leetcode_coach/tests/conftest.py b/leetcode_coach/leetcode_coach/tests/conftest.py new file mode 100644 index 0000000..40b47a7 --- /dev/null +++ b/leetcode_coach/leetcode_coach/tests/conftest.py @@ -0,0 +1,31 @@ +"""Keep every coach test off the real machine. + +No session log under ~/.local, no real `claude` binary, no Ollama on :11434, +and the fixture toolbox instead of whatever `toolbox.json` holds today +(another session fills it in). Env vars are *set* (to empty or a dead path) +rather than deleted, so teardown also removes the `LEETCODE_COACH_LOG` the +REPL exports for its hooks. +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +import pytest + +from leetcode_coach import local_backend, toolbox +from leetcode_coach.tests._shared import FIXTURE_TOOLBOX + +if TYPE_CHECKING: + from pathlib import Path + +DEAD_OLLAMA = "http://127.0.0.1:9" # the discard port: connection refused at once + + +@pytest.fixture(autouse=True) +def _hermetic_coach(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("LEETCODE_COACH_SESSIONS", str(tmp_path / "sessions")) + monkeypatch.setenv("LEETCODE_COACH_LOG", "") + monkeypatch.setenv("LEETCODE_CLAUDE_BIN", str(tmp_path / "no-such-claude")) + monkeypatch.setattr(local_backend, "OLLAMA_URL", DEAD_OLLAMA) + monkeypatch.setattr(toolbox, "TOOLBOX_PATH", FIXTURE_TOOLBOX) diff --git a/leetcode_coach/leetcode_coach/tests/test_claude_cli.py b/leetcode_coach/leetcode_coach/tests/test_claude_cli.py new file mode 100644 index 0000000..5c751ed --- /dev/null +++ b/leetcode_coach/leetcode_coach/tests/test_claude_cli.py @@ -0,0 +1,74 @@ +"""Tests for the one `claude -p` call shared by the REPL and the checker.""" + +from __future__ import annotations + +import json +import shutil +import subprocess +from typing import TYPE_CHECKING + +from leetcode_coach import claude_cli + +if TYPE_CHECKING: + from pathlib import Path + + import pytest + + +def _stub(tmp_path: Path, monkeypatch: pytest.MonkeyPatch, body: str) -> Path: + stub = tmp_path / "claude" + stub.write_text(body, encoding="utf-8") + stub.chmod(0o755) + monkeypatch.setenv("LEETCODE_CLAUDE_BIN", str(stub)) + return stub + + +def test_ask_parses_a_failing_claude_stub( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + body = json.dumps({"is_error": True, "result": "You've hit your session limit"}) + (tmp_path / "out.json").write_text(body, encoding="utf-8") + stub = _stub( + tmp_path, + monkeypatch, + f"#!/bin/sh\ncat {tmp_path / 'out.json'}\necho boom >&2\nexit 1\n", + ) + reply = claude_cli.ask([], "hi") + assert claude_cli.is_limit_hit(reply) + assert reply["_stderr"] == "boom" + stub.write_text("#!/bin/sh\necho not-json\nexit 3\n", encoding="utf-8") + assert claude_cli.ask([], "hi")["result"].startswith("[claude exited 3]") + stub.write_text("#!/bin/sh\necho '[1]'\n", encoding="utf-8") + assert claude_cli.ask([], "hi") == {"_stderr": ""} + + +def test_ask_passes_args_then_the_prompt( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + _stub(tmp_path, monkeypatch, '#!/bin/sh\nprintf \'{"result": "%s"}\' "$*"\n') + got = claude_cli.ask(["--model", "haiku"], "go")["result"] + assert got == "-p --model haiku --output-format json go" + + +def test_a_missing_binary_is_an_error_reply_not_a_crash() -> None: + reply = claude_cli.ask([], "hi") + assert reply["result"].startswith("[claude could not run") + + +def test_a_hung_claude_times_out(monkeypatch: pytest.MonkeyPatch) -> None: + def hang(*_a: object, **_kw: object) -> subprocess.CompletedProcess[str]: + msg = "claude" + raise subprocess.TimeoutExpired(msg, 1.0) + + monkeypatch.setattr(subprocess, "run", hang) + assert claude_cli.ask([], "hi", timeout=1.0)["result"].startswith( + "[claude could not run" + ) + + +def test_the_binary_is_looked_up_per_call(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("LEETCODE_CLAUDE_BIN", "") + monkeypatch.setattr(shutil, "which", lambda _n: "/opt/claude") + assert claude_cli.claude_bin() == "/opt/claude" + monkeypatch.setattr(shutil, "which", lambda _n: None) + assert claude_cli.claude_bin() == "claude" diff --git a/leetcode_coach/leetcode_coach/tests/test_coach_e2e.py b/leetcode_coach/leetcode_coach/tests/test_coach_e2e.py new file mode 100644 index 0000000..613f132 --- /dev/null +++ b/leetcode_coach/leetcode_coach/tests/test_coach_e2e.py @@ -0,0 +1,181 @@ +"""Scripted end-to-end runs of the req-first coach (TODO-coach.md "Build done"). + +The REPL runs in-process against a fake `claude` binary (LEETCODE_CLAUDE_BIN, +a real subprocess per turn) and a fake Ollama on a localhost port, with +kuhy's input scripted. Only the problem fetch is stubbed. +""" + +from __future__ import annotations + +import json +from pathlib import Path +from typing import TYPE_CHECKING, Any + +import pytest + +from leetcode_coach import leetcode_repl, local_backend, local_session, report +from leetcode_coach.tests._fake_ollama import FakeOllama + +if TYPE_CHECKING: + from collections.abc import Iterator + +FAKE_CLAUDE = Path(__file__).with_name("_fake_claude.py") +STATEMENT = ( + "# 268. Missing Number (Easy)\nn distinct numbers in [0, n]; return the missing " + "one.\n\n## Starter code (javascript)\n```\n" + "var missingNumber = function(nums) {\n};\n```\n" +) +CODE = "var missingNumber = function(nums) {\n return 0;\n};" + + +def _script(tmp_path: Path, monkeypatch: pytest.MonkeyPatch, **script: Any) -> Path: + root = tmp_path / "fake-claude" + root.mkdir(exist_ok=True) + for used in root.glob("*.used"): + used.unlink() + (root / "calls.jsonl").write_text("", encoding="utf-8") + (root / "script.json").write_text(json.dumps(script), encoding="utf-8") + monkeypatch.setenv("LEETCODE_CLAUDE_BIN", str(FAKE_CLAUDE)) + monkeypatch.setenv("FAKE_CLAUDE_DIR", str(root)) + return root + + +def _calls(root: Path, kind: str) -> list[str]: + lines = (root / "calls.jsonl").read_text(encoding="utf-8").splitlines() + return [c["prompt"] for c in map(json.loads, lines) if c["kind"] == kind] + + +def _type(monkeypatch: pytest.MonkeyPatch, *blocks: str) -> None: + it: Iterator[str] = iter(blocks) + monkeypatch.setattr(leetcode_repl, "read_block", lambda: next(it, None)) + + +def _events(tmp_path: Path, slug: str) -> list[dict[str, Any]]: + (path,) = (tmp_path / "sessions").glob(f"*-{slug}.jsonl") + return [json.loads(x) for x in path.read_text(encoding="utf-8").splitlines()] + + +def _coach(*texts: str) -> list[dict[str, str]]: + return [{"result": t} for t in texts] + + +def _incomplete(*missing: str) -> dict[str, str]: + return {"result": json.dumps({"verdict": "incomplete", "missing": list(missing)})} + + +def test_claude_path_req_first( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + capsys: pytest.CaptureFixture[str], +) -> None: + monkeypatch.setattr(local_session, "fetched_statement", lambda _p: STATEMENT) + root = _script( + tmp_path, + monkeypatch, + haiku=[ + _incomplete("self-inverse", "XOR"), + _incomplete("self-inverse"), + _incomplete("closed-form-total"), + ], + coach=_coach( + "Restated. What feels unclear?\n", + "What must the operation do?", + "Closer. What is still open?", + "Try the code?", + "Run it on [3,0,1]?\n\n", + ), + ) + _type(monkeypatch, CODE, "req: find it", CODE, "req: v2", "req: v3", CODE, "quit") + assert leetcode_repl.main(["missing-number"]) == 0 + out = capsys.readouterr().out + assert "[code held back: write the requirement list first" in out + assert "self-inverse: applying it twice with the same value undoes it" in out + assert "[code held back: the req list is still incomplete" in out + assert "accepted anyway" in out + assert "XOR" not in out + assert ""), + ) + _type(monkeypatch, "req: each value twice but one", CODE) + assert leetcode_repl.main(["single-number"]) == 0 + assert "unverified" in capsys.readouterr().out + events = _events(tmp_path, "single-number") + assert [e["event"] for e in events] == ["req", "req_check", "code", "rung"] + assert events[1]["verdict"] == "unverified" + assert "req_incomplete" not in events[2] + assert report.main([]) == 0 + rows = capsys.readouterr().out.splitlines() + name = next(p.stem for p in (tmp_path / "sessions").glob("*.jsonl")) + assert rows[1] == f"{name:<36}{'y':<17}{'H1':<10}{'n':<8}unverified" + + +@pytest.fixture +def ollama(monkeypatch: pytest.MonkeyPatch) -> Iterator[FakeOllama]: + server = FakeOllama( + checker=['{"verdict": "pass", "missing": []}'], + coach=[ + "Restated. What feels unclear?", + "Good list. What happens to a pair?", + "Think XOR. What is a ^ a?", + "What does a value do to itself under your rule?\n", + "Yes, XOR. What is a ^ a?", + ], + ) + monkeypatch.setattr(local_backend, "OLLAMA_URL", server.url) + yield server + server.close() + + +def test_local_path_bounces_a_leak_but_not_a_typed_name( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + capsys: pytest.CaptureFixture[str], + ollama: FakeOllama, +) -> None: + monkeypatch.setattr(local_session, "fetch_statement", lambda *_a: STATEMENT) + _script(tmp_path, monkeypatch, haiku=[{"exit": 1, "stderr": "usage limit reached"}]) + _type(monkeypatch, "req: pairs vanish, one survives", "no idea", "is it XOR?") + assert leetcode_repl.main(["--local", "single-number"]) == 0 + out = capsys.readouterr().out + assert "[req list accepted -- code is open]" in out + assert "Think XOR" not in out + assert "What does a value do to itself under your rule?" in out + assert "Yes, XOR. What is a ^ a?" in out + assert "" in prompt assert prompt.rstrip().endswith("```") assert "var twoSum" in prompt diff --git a/leetcode_coach/leetcode_coach/tests/test_local_session.py b/leetcode_coach/leetcode_coach/tests/test_local_session.py index 95b946c..6cce164 100644 --- a/leetcode_coach/leetcode_coach/tests/test_local_session.py +++ b/leetcode_coach/leetcode_coach/tests/test_local_session.py @@ -3,14 +3,17 @@ from __future__ import annotations import sys -from typing import Any +from typing import TYPE_CHECKING, Any import urllib.error import pytest -from leetcode_coach import local_backend, local_coach, local_session +from leetcode_coach import local_backend, local_coach, local_session, phase, toolbox from leetcode_coach.tests._shared import SAFE, STATEMENT, FakeChat, completed +if TYPE_CHECKING: + from pathlib import Path + def test_opening_message_and_quit() -> None: assert local_session.opening_message(["loops"]) == ( @@ -126,7 +129,7 @@ def test_run_reports_an_unreachable_ollama( _session(monkeypatch, chat=down) assert local_session.run("x", [], "Start.", lambda: None) == 1 out = capsys.readouterr().out - assert "[Ollama is not reachable at http://localhost:11434: " in out + assert f"[Ollama is not reachable at {local_backend.OLLAMA_URL}: " in out assert out.endswith("[start it with: systemctl start ollama]\n") @@ -140,3 +143,31 @@ def test_run_reports_a_context_overflow( _session(monkeypatch, chat=full) assert local_session.run("x", [], "Start.", lambda: None) == 1 assert capsys.readouterr().out.endswith("[too long]\n") + + +def test_fetched_statement_never_asks_for_a_paste() -> None: + ok = local_session.fetched_statement( + "x", runner=lambda *_a, **_k: completed(0, STATEMENT) + ) + assert ok == STATEMENT + assert ( + local_session.fetched_statement("x", runner=lambda *_a, **_k: completed(2)) + is None + ) + + +def test_run_with_a_gate_holds_code_back( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + capsys: pytest.CaptureFixture[str], +) -> None: + fake = FakeChat(SAFE, SAFE) + _session(monkeypatch, chat=fake) + gate = phase.PhaseGate(tmp_path / "log.jsonl", toolbox.load(), lambda: "s") + blocks = iter(["function f(nums) {\n return 0;\n}", "hint", "quit"]) + assert local_session.run("x", [], "Start.", lambda: next(blocks), gate=gate) == 0 + out = capsys.readouterr().out + assert "[code held back: write the requirement list first" in out + assert out.count(SAFE) == 2 + assert len(fake.prompts) == 2 # the code never reached the model + assert gate.known == STATEMENT diff --git a/leetcode_coach/leetcode_coach/tests/test_phase.py b/leetcode_coach/leetcode_coach/tests/test_phase.py new file mode 100644 index 0000000..ab1a755 --- /dev/null +++ b/leetcode_coach/leetcode_coach/tests/test_phase.py @@ -0,0 +1,221 @@ +"""Tests for the REPL's req-before-code phase gate and the rung marker.""" + +from __future__ import annotations + +import os +from typing import TYPE_CHECKING + +import pytest + +from leetcode_coach import ( + local_prompt, + local_session, + phase, + req_check, + session_log, + toolbox, +) + +if TYPE_CHECKING: + from pathlib import Path + +CODE = "function missingNumber(nums) {\n return 0;\n}" + + +@pytest.mark.parametrize( + ("block", "verdict"), + [ + ("```js\nlet x = 1\n```", True), + (CODE, True), + ("return acc;", True), + ("const seen = new Set()", True), + ("nums.map(x => x * 2)", True), + ("def f(nums):", True), + ("for (let i = 0; i < n; i++)", True), + ("is it XOR?", False), + ("I think (not sure) it is about pairs", False), + ("req: a = b", False), + ("is it x ^ y?", None), + ("a = b", None), + ("what does f(3) give", None), + ], +) +def test_regex_verdict(block: str, verdict: bool | None) -> None: + assert phase.regex_verdict(block) is verdict + + +@pytest.mark.parametrize( + ("text", "clean", "rung"), + [ + ("What next?\n\n\n", "What next?", "H1"), + ("A B?", "A B?", "H3"), + ("No marker?", "No marker?", None), + ("Bad ?", "Bad ?", None), + ], +) +def test_split_rung(text: str, clean: str, rung: str | None) -> None: + assert local_prompt.split_rung(text) == (clean, rung) + + +def _verdicts(*verdicts: str) -> list[dict[str, object]]: + return [{"event": "req_check", "verdict": v} for v in verdicts] + + +def test_req_state() -> None: + assert phase.req_state([]) == phase.ReqState(accepted=False, forced=False, tries=0) + assert phase.req_state(_verdicts("incomplete", "pass")).accepted + assert phase.req_state(_verdicts("unverified")).accepted + forced = phase.req_state(_verdicts("incomplete", "incomplete", "incomplete")) + assert forced == phase.ReqState(accepted=False, forced=True, tries=3) + assert forced.open + + +class _Checks: + """Scripted `req_check.check` results; counts statement fetches.""" + + def __init__(self, *results: req_check.Check) -> None: + self.results = list(results) + self.statements: list[str] = [] + + def __call__( + self, statement: str, _req: str, _box: object, _ask: object = None + ) -> req_check.Check: + self.statements.append(statement) + return self.results.pop(0) + + +def _gate( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, *results: req_check.Check +) -> tuple[phase.PhaseGate, _Checks]: + checks = _Checks(*results) + monkeypatch.setattr(req_check, "check", checks) + fetched: list[int] = [] + + def statement() -> str: + fetched.append(1) + return f"statement {len(fetched)}" + + log = tmp_path / "log.jsonl" + return phase.PhaseGate(log, toolbox.load(), statement), checks + + +def _events(gate: phase.PhaseGate) -> list[object]: + return [e["event"] for e in session_log.read(gate.log)] + + +def test_code_before_req_is_refused_and_never_logged( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + gate, _ = _gate(tmp_path, monkeypatch) + sent, note = gate.screen(CODE) + assert sent is None + assert "req:" in note + assert _events(gate) == [] + assert gate.screen("what is the constraint?") == ("what is the constraint?", "") + + +def test_a_passing_list_opens_code( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + passed = req_check.Check("pass", (), "haiku") + gate, checks = _gate(tmp_path, monkeypatch, passed, passed) + sent, note = gate.screen("req: one pass, no extra memory") + assert sent == "req: one pass, no extra memory\n\n[req check: pass]" + assert note == "[req list accepted -- code is open]" + assert gate.screen(CODE) == (CODE, "") + assert gate.code_seen() + events = session_log.read(gate.log) + assert [e["event"] for e in events] == ["req", "req_check", "code"] + assert events[1]["verdict"] == "pass" + assert "req_incomplete" not in events[2] + gate.screen("req: again") + assert checks.statements == ["statement 1", "statement 1"] # fetched once + + +def test_incomplete_blocks_until_the_third_list( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + missing = req_check.Check("incomplete", ("self-inverse",), "haiku") + gate, _ = _gate(tmp_path, monkeypatch, missing, missing, missing) + sent, note = gate.screen("req: loop over it") + assert sent is not None + assert sent.endswith("[req check: incomplete; missing properties: self-inverse]") + assert "self-inverse: applying it twice with the same value undoes it" in note + assert "XOR" not in note + assert "1 of 3" in note + refused, why = gate.screen(CODE) + assert refused is None + assert "still incomplete" in why + gate.screen("req: v2") + _, last = gate.screen("req: v3") + assert "accepted anyway" in last + assert gate.screen(CODE) == (CODE, "") + code = session_log.read(gate.log)[-1] + assert code["event"] == "code" + assert code["req_incomplete"] is True + + +def test_unverified_fails_open(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + gate, _ = _gate(tmp_path, monkeypatch, req_check.Check("unverified")) + _, note = gate.screen("req: something") + assert "unverified" in note + assert gate.screen(CODE) == (CODE, "") + + +def test_ambiguous_blocks_go_to_the_model( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + gate, _ = _gate(tmp_path, monkeypatch) + asked: list[str] = [] + + def classify(block: str, _ask: object = None) -> bool: + asked.append(block) + return block == "x ^= y" + + monkeypatch.setattr(req_check, "looks_like_code", classify) + assert gate.screen("is it x ^ y?") == ("is it x ^ y?", "") + assert gate.screen("x ^= y")[0] is None + assert asked == ["is it x ^ y?", "x ^= y"] + + +def test_reveal_and_rungs_are_logged( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + gate, _ = _gate(tmp_path, monkeypatch) + assert gate.screen(" reveal ") == (" reveal ", "") + assert gate.after_reply("Try it?\n") == "Try it?" + assert gate.after_reply("Plain?") == "Plain?" + events = session_log.read(gate.log) + assert [e["event"] for e in events] == ["reveal", "rung"] + assert events[1]["rung"] == "H2" + + +def test_leaks_only_before_code_and_never_what_kuhy_typed( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + gate, _ = _gate(tmp_path, monkeypatch, req_check.Check("pass")) + reply = "Think about XOR and a hash map." + assert gate.leaks(reply, ["hm"]) == ["XOR", "hash map"] + assert gate.leaks(reply, ["is it XOR?"]) == ["hash map"] + assert gate.leaks(reply, ["reveal"]) == [] + gate.screen("req: x") + gate.screen(CODE) + assert gate.leaks(reply, []) == [] + + +def test_a_statement_naming_an_entry_lets_the_coach_restate_it( + tmp_path: Path, +) -> None: + gate = phase.PhaseGate(tmp_path / "l.jsonl", toolbox.load(), lambda: "Use XOR.") + assert gate.leaks("XOR here", []) == [] + + +def test_open_exports_the_log_path_for_the_hooks( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setattr(local_session, "fetched_statement", lambda _p: None) + gate = phase.PhaseGate.open("https://leetcode.com/problems/missing-number/") + assert os.environ[session_log.LOG_ENV] == str(gate.log) + assert gate.log.name.endswith("-missing-number.jsonl") + assert gate.statement() == "missing-number" + assert gate.toolbox.entries diff --git a/leetcode_coach/leetcode_coach/tests/test_report.py b/leetcode_coach/leetcode_coach/tests/test_report.py new file mode 100644 index 0000000..e12ab9c --- /dev/null +++ b/leetcode_coach/leetcode_coach/tests/test_report.py @@ -0,0 +1,54 @@ +"""Tests for `leetcode report` (its stdout is the interface).""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +from leetcode_coach import report, session_log + +if TYPE_CHECKING: + from pathlib import Path + + import pytest + +HEADER = "problem" + " " * 29 + "req-before-code max-rung reveal req_check" + + +def _row(*cols: str) -> str: + return f"{cols[0]:<36}{cols[1]:<17}{cols[2]:<10}{cols[3]:<8}{cols[4]}" + + +def test_report_prints_one_row_per_problem( + tmp_path: Path, capsys: pytest.CaptureFixture[str] +) -> None: + folder = tmp_path / "sessions" + good = folder / "2026-10-22-missing-number.jsonl" + session_log.append(good, "req", text="req: a") + session_log.append(good, "req_check", verdict="incomplete", missing=["p"]) + session_log.append(good, "req", text="req: b") + session_log.append(good, "req_check", verdict="pass", missing=[]) + session_log.append(good, "code") + session_log.append(good, "rung", rung="H1") + session_log.append(good, "rung", rung="H0") + bad = folder / "2026-10-23-single-number.jsonl" + session_log.append(bad, "code") + session_log.append(bad, "rung", rung="H3") + session_log.append(bad, "reveal") + fresh = folder / "2026-10-23-two-sum.jsonl" + session_log.append(fresh, "rung", rung="H9") + assert report.main([]) == 0 + assert capsys.readouterr().out.splitlines() == [ + HEADER, + _row("2026-10-22-missing-number", "y", "H1", "n", "incomplete,pass"), + _row("2026-10-23-single-number", "n", "H3", "y", "-"), + _row("2026-10-23-two-sum", "-", "-", "n", "-"), + ] + + +def test_report_without_sessions_says_so( + tmp_path: Path, capsys: pytest.CaptureFixture[str] +) -> None: + assert report.main([]) == 0 + assert capsys.readouterr().out == ( + f"no coach sessions yet in {tmp_path / 'sessions'}\n" + ) diff --git a/leetcode_coach/leetcode_coach/tests/test_req_check.py b/leetcode_coach/leetcode_coach/tests/test_req_check.py new file mode 100644 index 0000000..d784fee --- /dev/null +++ b/leetcode_coach/leetcode_coach/tests/test_req_check.py @@ -0,0 +1,166 @@ +"""Tests for the haiku -> Ollama -> fail-open requirement-list checker.""" + +from __future__ import annotations + +import json +from typing import Any +import urllib.error + +import pytest + +from leetcode_coach import claude_cli, local_backend, req_check, toolbox +from leetcode_coach.tests._shared import FIXTURE_TOOLBOX + +STATEMENT = "# 268. Missing Number\nn distinct numbers in [0, n]; find the missing one." + + +class _Haiku: + """A stand-in for `claude_cli.ask` that always gives the same answer.""" + + def __init__(self, answer: object) -> None: + self.text = answer if isinstance(answer, str) else json.dumps(answer) + self.calls: list[tuple[list[str], str]] = [] + + def __call__(self, args: list[str], prompt: str, **_kw: object) -> dict[str, Any]: + self.calls.append((args, prompt)) + return {"result": self.text, "_stderr": ""} + + +def _haiku(answer: object) -> _Haiku: + return _Haiku(answer) + + +def _ollama(monkeypatch: pytest.MonkeyPatch, text: str | Exception) -> list[Any]: + sent: list[Any] = [] + + def chat(messages: list[dict[str, str]]) -> local_backend.LocalReply: + sent.append(messages) + if isinstance(text, Exception): + raise text + return local_backend.LocalReply(text, 1.0, 10) + + monkeypatch.setattr(local_backend, "chat", chat) + return sent + + +@pytest.mark.parametrize( + ("text", "obj"), + [ + ('{"verdict": "pass"}', {"verdict": "pass"}), + ('```json\n{"code": true}\n```', {"code": True}), + ('Sure! {"a": 1} hope that helps {"b": 2}', {"a": 1}), + ("no json here", None), + ("{broken", None), + ("[1, 2]", None), + ], +) +def test_extract_json(text: str, obj: object) -> None: + assert req_check.extract_json(text) == obj + + +VOCAB = toolbox.load(FIXTURE_TOOLBOX).properties + + +@pytest.mark.parametrize( + ("obj", "expected"), + [ + ({"verdict": "pass", "missing": []}, ("pass", ())), + ({"verdict": "pass"}, ("pass", ())), + ( + {"verdict": "incomplete", "missing": ["self-inverse", "XOR", 4, "nope"]}, + ("incomplete", ("self-inverse",)), + ), + ({"verdict": "incomplete", "missing": ["XOR"]}, None), + ({"verdict": "maybe", "missing": []}, None), + ({"verdict": "incomplete", "missing": "self-inverse"}, None), + (None, None), + ], +) +def test_parse_verdict_keeps_only_vocabulary_properties( + obj: object, expected: tuple[str, tuple[str, ...]] | None +) -> None: + got = req_check.parse_verdict(obj, VOCAB) + assert (got if got is None else (got.verdict, got.missing)) == expected + + +def test_the_prompt_carries_statement_list_and_vocabulary() -> None: + prompt = req_check.check_prompt(STATEMENT, "req: one pass", toolbox.load()) + assert STATEMENT in prompt + assert "req: one pass" in prompt + assert "self-inverse: applying it twice" in prompt + assert "leetcode.com/problems" not in prompt + + +def test_haiku_answers_first(monkeypatch: pytest.MonkeyPatch) -> None: + sent = _ollama(monkeypatch, "unused") + ask = _haiku({"verdict": "incomplete", "missing": ["constant-space", "XOR"]}) + got = req_check.check(STATEMENT, "req: loop", toolbox.load(), ask) + assert got == req_check.Check("incomplete", ("constant-space",), "haiku") + args, _prompt = ask.calls[0] + assert args[:2] == ["--model", "haiku"] + assert "--resume" not in args + assert sent == [] + + +@pytest.mark.parametrize( + "haiku_says", + [ + "You've hit your session limit · resets 3pm", + "not json at all", + {"verdict": "incomplete", "missing": ["XOR"]}, + ], +) +def test_ollama_steps_in_when_haiku_cannot_answer( + monkeypatch: pytest.MonkeyPatch, haiku_says: object +) -> None: + _ollama(monkeypatch, '{"verdict": "pass", "missing": []}') + got = req_check.check(STATEMENT, "req: x", toolbox.load(), _haiku(haiku_says)) + assert got == req_check.Check("pass", (), "ollama") + + +@pytest.mark.parametrize( + "ollama", [urllib.error.URLError("refused"), "garbage", ConnectionError("x")] +) +def test_both_down_fails_open_unverified( + monkeypatch: pytest.MonkeyPatch, ollama: str | Exception +) -> None: + _ollama(monkeypatch, ollama) + got = req_check.check(STATEMENT, "req: x", toolbox.load(), _haiku("nope")) + assert got == req_check.Check("unverified", (), "none") + + +def test_an_empty_vocabulary_cannot_be_checked(monkeypatch: pytest.MonkeyPatch) -> None: + sent = _ollama(monkeypatch, "unused") + ask = _haiku("unused") + got = req_check.check(STATEMENT, "req: x", toolbox.Toolbox(), ask) + assert got.verdict == "unverified" + assert ask.calls == [] + assert sent == [] + + +def test_the_default_asker_is_the_claude_cli(monkeypatch: pytest.MonkeyPatch) -> None: + ask = _haiku({"verdict": "pass"}) + monkeypatch.setattr(claude_cli, "ask", ask) + assert req_check.check(STATEMENT, "req: x", toolbox.load()).via == "haiku" + assert ask.calls[0][0][:2] == ["--model", "haiku"] + + +@pytest.mark.parametrize( + ("haiku_says", "ollama", "is_code"), + [ + ({"code": True}, "unused", True), + ({"code": False}, "unused", False), + ("broken", '{"code": true}', True), + ("broken", '{"code": "yes"}', False), + ("broken", urllib.error.URLError("down"), False), + ], +) +def test_model_code_classifier_fails_open_to_not_code( + monkeypatch: pytest.MonkeyPatch, + haiku_says: object, + ollama: str | Exception, + *, + is_code: bool, +) -> None: + _ollama(monkeypatch, ollama) + assert req_check.looks_like_code("x ^ y", _haiku(haiku_says)) is is_code diff --git a/leetcode_coach/leetcode_coach/tests/test_session_log.py b/leetcode_coach/leetcode_coach/tests/test_session_log.py new file mode 100644 index 0000000..05dbf27 --- /dev/null +++ b/leetcode_coach/leetcode_coach/tests/test_session_log.py @@ -0,0 +1,90 @@ +"""Tests for the coach's JSONL session log (contract read by leetcode_guard).""" + +from __future__ import annotations + +from datetime import date, datetime +import json +import logging +from typing import TYPE_CHECKING + +import pytest + +from leetcode_coach import session_log +from leetcode_guard import _coach_log + +if TYPE_CHECKING: + from pathlib import Path + + +def test_sessions_dir_follows_the_env( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + assert session_log.sessions_dir() == tmp_path / "sessions" + monkeypatch.setenv(session_log.SESSIONS_ENV, "") + assert session_log.sessions_dir() == session_log.DEFAULT_DIR + + +@pytest.mark.parametrize( + ("problem", "slug"), + [ + ("https://leetcode.com/problems/missing-number/description/", "missing-number"), + ("single-number", "single-number"), + ("Weird Name!", "weird-name"), + ("!!!", "problem"), + ], +) +def test_log_path_is_dated_and_slugged(tmp_path: Path, problem: str, slug: str) -> None: + path = session_log.log_path(problem, date(2026, 10, 23)) + assert path == tmp_path / "sessions" / f"2026-10-23-{slug}.jsonl" + + +def test_log_path_defaults_to_today() -> None: + assert session_log.log_path("x").name.startswith(date.today().isoformat()) + + +def test_append_writes_one_timestamped_object_per_line(tmp_path: Path) -> None: + path = tmp_path / "deep" / "log.jsonl" + session_log.append(path, "req", text="req: a") + session_log.append(path, "req_check", verdict="pass", missing=[]) + lines = path.read_text(encoding="utf-8").splitlines() + first = json.loads(lines[0]) + assert first["event"] == "req" + assert first["text"] == "req: a" + assert datetime.fromisoformat(first["ts"]).tzinfo is not None + assert json.loads(lines[1])["verdict"] == "pass" + + +def test_append_refuses_an_event_outside_the_contract(tmp_path: Path) -> None: + with pytest.raises(ValueError, match="req_incomplete"): + session_log.append(tmp_path / "x.jsonl", "req_incomplete") + + +def test_read_tolerates_a_missing_file_and_torn_lines( + tmp_path: Path, caplog: pytest.LogCaptureFixture +) -> None: + path = tmp_path / "log.jsonl" + assert session_log.read(path) == [] + path.write_text('{"event": "code"}\n{"event": \n[1]\n\n', encoding="utf-8") + with caplog.at_level(logging.WARNING, logger="leetcode_coach.session_log"): + assert session_log.read(path) == [{"event": "code"}] + assert "not JSON" in caplog.text + assert session_log.has_code(path) + assert not session_log.has_code(tmp_path / "other.jsonl") + + +def test_the_status_reader_understands_what_is_written(tmp_path: Path) -> None: + path = tmp_path / "2026-10-23-missing-number.jsonl" + session_log.append(path, "req", text="req: x", cluster="bit-cancellation") + session_log.append(path, "req_check", verdict="incomplete", missing=["p"]) + session_log.append(path, "code", req_incomplete=True) + session_log.append(path, "rung", rung="H2") + problem = _coach_log.read_session(path) + assert problem == _coach_log.CoachProblem( + slug="missing-number", + cluster="bit-cancellation", + req_before_code=True, + max_rung="H2", + revealed=False, + req_check="incomplete", + day="2026-10-23", + ) diff --git a/leetcode_coach/leetcode_coach/tests/test_stop_gate_leak.py b/leetcode_coach/leetcode_coach/tests/test_stop_gate_leak.py new file mode 100644 index 0000000..132d1cb --- /dev/null +++ b/leetcode_coach/leetcode_coach/tests/test_stop_gate_leak.py @@ -0,0 +1,98 @@ +"""Tests for the Stop hook's leak bounce: no toolbox name before kuhy's first code.""" + +from __future__ import annotations + +import io +import json +from typing import TYPE_CHECKING, Any + +from leetcode_coach import session_log, stop_gate + +if TYPE_CHECKING: + from pathlib import Path + + import pytest + +SLASH_LOAD = { + "type": "user", + "message": { + "content": "leetcode\n" + "/leetcode" + }, +} +LEAKY = "Think about XOR: what does a value do to itself?" + + +def _user(text: str) -> dict[str, Any]: + return {"type": "user", "message": {"content": [{"type": "text", "text": text}]}} + + +def _raw(*entries: dict[str, Any]) -> str: + return "".join(json.dumps(e, separators=(",", ":")) + "\n" for e in entries) + + +def test_a_reply_naming_an_entry_before_code_is_bounced() -> None: + reason = stop_gate.verdict(_raw(SLASH_LOAD, _user("no idea")), LEAKY) + assert reason is not None + assert reason.startswith("leetcode gate: your reply names XOR") + assert "properties" in reason + + +def test_is_it_xor_gets_an_honest_answer() -> None: + raw = _raw(SLASH_LOAD, _user("is it XOR?")) + assert stop_gate.verdict(raw, "Yes, XOR. What is x ^ x?") is None + + +def test_the_bounce_reason_does_not_whitelist_itself() -> None: + feedback = _user("Stop hook feedback: leetcode gate: your reply names XOR") + assert stop_gate.verdict(_raw(SLASH_LOAD, feedback), LEAKY) is not None + + +def test_names_in_the_fetched_statement_may_be_restated() -> None: + tool = { + "type": "user", + "message": { + "content": [{"type": "tool_result", "content": "Solve it with XOR."}] + }, + } + assert stop_gate.verdict(_raw(SLASH_LOAD, tool), LEAKY) is None + + +def test_the_session_log_lifts_the_bounce_once_code_exists( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + log = tmp_path / "log.jsonl" + monkeypatch.setenv(session_log.LOG_ENV, str(log)) + raw = _raw(SLASH_LOAD, _user("hmm")) + assert stop_gate.verdict(raw, LEAKY) is not None + session_log.append(log, "code") + assert stop_gate.verdict(raw, LEAKY) is None + + +def test_without_a_log_kuhys_own_code_lifts_it() -> None: + """The TUI has no REPL and no log: kuhy's code in the transcript counts.""" + raw = _raw(SLASH_LOAD, _user("function f(nums) {\n return 0;\n}")) + assert stop_gate.verdict(raw, LEAKY) is None + + +def test_reveal_lifts_the_bounce() -> None: + assert stop_gate.verdict(_raw(SLASH_LOAD, _user("reveal")), LEAKY) is None + + +def test_main_prints_the_leak_block( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + capsys: pytest.CaptureFixture[str], +) -> None: + monkeypatch.setattr(stop_gate, "STATE_DIR", tmp_path / "state") + transcript = tmp_path / "t.jsonl" + transcript.write_text(_raw(SLASH_LOAD, _user("hmm")), encoding="utf-8") + event = { + "transcript_path": str(transcript), + "last_assistant_message": LEAKY, + "session_id": "s", + "prompt_id": "p", + } + monkeypatch.setattr("sys.stdin", io.StringIO(json.dumps(event))) + assert stop_gate.main() == 0 + assert json.loads(capsys.readouterr().out)["decision"] == "block" diff --git a/leetcode_coach/leetcode_coach/tests/test_toolbox.py b/leetcode_coach/leetcode_coach/tests/test_toolbox.py new file mode 100644 index 0000000..bc35edc --- /dev/null +++ b/leetcode_coach/leetcode_coach/tests/test_toolbox.py @@ -0,0 +1,57 @@ +"""Tests for the toolbox loader (the data itself belongs to TODO-coach.md item 9).""" + +from __future__ import annotations + +import json +import logging +from typing import TYPE_CHECKING + +import pytest + +from leetcode_coach import toolbox + +if TYPE_CHECKING: + from pathlib import Path + + +def test_load_reads_the_configured_toolbox() -> None: + box = toolbox.load() + xor = box.entries[0] + assert xor.name == "XOR" + assert xor.names() == ("XOR", "^", "exclusive or", "bitwise xor") + assert xor.properties == ("self-inverse", "commutative", "associative") + assert box.properties["constant-space"] == "memory does not grow with the input" + + +def test_aliases_and_properties_are_optional(tmp_path: Path) -> None: + path = tmp_path / "box.json" + path.write_text(json.dumps({"entries": [{"name": "stack"}]}), encoding="utf-8") + box = toolbox.load(path) + assert box.entries[0].names() == ("stack",) + assert box.entries[0].properties == () + assert box.properties == {} + + +@pytest.mark.parametrize( + "body", + [ + "not json", + json.dumps([]), + json.dumps({"entries": "XOR"}), + json.dumps({"entries": [{"aliases": []}]}), + json.dumps({"entries": [{"name": "x", "aliases": [1]}]}), + json.dumps({"entries": [], "properties": {"p": 3}}), + ], +) +def test_a_broken_toolbox_is_empty_and_logged( + tmp_path: Path, body: str, caplog: pytest.LogCaptureFixture +) -> None: + path = tmp_path / "box.json" + path.write_text(body, encoding="utf-8") + with caplog.at_level(logging.WARNING, logger="leetcode_coach.toolbox"): + assert toolbox.load(path) == toolbox.Toolbox() + assert "unusable" in caplog.text + + +def test_a_missing_toolbox_is_empty(tmp_path: Path) -> None: + assert toolbox.load(tmp_path / "nope.json").entries == () diff --git a/leetcode_coach/leetcode_coach/toolbox.py b/leetcode_coach/leetcode_coach/toolbox.py new file mode 100644 index 0000000..7e8267f --- /dev/null +++ b/leetcode_coach/leetcode_coach/toolbox.py @@ -0,0 +1,85 @@ +"""The toolbox of operations, loaded from ``toolbox.json`` next to this file. + +Schema: ``{"entries": [{"name", "aliases", "properties"}], "properties": +{"": ""}}``. The entries feed the leak bounce (no +entry may be named before kuhy's first code) and the property vocabulary is +the only thing the requirement checker may say back. The data is +TODO-coach.md item 9's; this module only reads it. + +An unreadable or malformed file is an empty toolbox plus a warning: both +consumers then fail open (nothing bounced, req check ``unverified``) rather +than locking kuhy out of a session over a typo in a JSON file. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +import json +import logging +from pathlib import Path + +_LOG = logging.getLogger(__name__) + +TOOLBOX_PATH = Path(__file__).with_name("toolbox.json") + + +@dataclass(frozen=True) +class Entry: + """One operation: how it is called and which properties it has.""" + + name: str + aliases: tuple[str, ...] = () + properties: tuple[str, ...] = () + + def names(self) -> tuple[str, ...]: + """The name first, then every alias.""" + return (self.name, *self.aliases) + + +@dataclass(frozen=True) +class Toolbox: + """All entries plus the property vocabulary (name -> meaning).""" + + entries: tuple[Entry, ...] = () + properties: dict[str, str] = field(default_factory=dict) + + +def _strings(value: object, what: str) -> tuple[str, ...]: + if not isinstance(value, list) or not all(isinstance(v, str) for v in value): + msg = f"{what} must be a list of strings" + raise TypeError(msg) + return tuple(value) + + +def _entry(raw: object) -> Entry: + if not isinstance(raw, dict) or not isinstance(raw.get("name"), str): + msg = "every entry needs a string name" + raise TypeError(msg) + return Entry( + name=raw["name"], + aliases=_strings(raw.get("aliases", []), "aliases"), + properties=_strings(raw.get("properties", []), "properties"), + ) + + +def _parse(raw: object) -> Toolbox: + if not isinstance(raw, dict) or not isinstance(raw.get("entries"), list): + msg = "the toolbox must be an object with an entries list" + raise TypeError(msg) + props = raw.get("properties", {}) + if not isinstance(props, dict) or not all( + isinstance(v, str) for v in props.values() + ): + msg = "properties must map each name to a one-line meaning" + raise TypeError(msg) + return Toolbox(tuple(_entry(e) for e in raw["entries"]), dict(props)) + + +def load(path: Path | None = None) -> Toolbox: + """The toolbox at `path` (default ``toolbox.json``), or an empty one.""" + target = TOOLBOX_PATH if path is None else path + try: + return _parse(json.loads(target.read_text(encoding="utf-8"))) + except (OSError, ValueError, TypeError) as exc: + _LOG.warning("toolbox %s is unusable, checks fail open: %s", target, exc) + return Toolbox()