kuhyx/utils
1
0
Fork 0
mirror of https://github.com/kuhyx/utils.git synced 2026-11-10 19:07:17 +01:00
Code Issues Actions 14 Packages Projects Releases Wiki Activity

Coach phase gate, req checker, leak bounce, session log and report (TODO-coach 4-8)

Browse Source 
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01PjvF5DZG9JduSBYKxqyysi
This commit is contained in:
Krzysztof Rudnicki 2026-10-23 15:45:08 +02:00
parent 96a0d8af07
commit c22b14389a
30 changed files with 2191 additions and 195 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

174
leetcode_coach/leetcode_coach/SKILL.md
View File

@ -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, `<!-- rung:H0 -->`
… `<!-- rung:H3 -->` (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 '<path>'` 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 '<path>'` 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 <url>` (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 <url>` (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 <url>` 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.

76
leetcode_coach/leetcode_coach/claude_cli.py Normal file
View File

@ -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))

42
leetcode_coach/leetcode_coach/leak.py Normal file
View File

@ -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"(?<!\w){words}(?!\w)", re.IGNORECASE)
def _named(text: str, names: tuple[str, ...]) -> 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())
]

92
leetcode_coach/leetcode_coach/leetcode_repl.py
View File

@ -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 ``<!-- rung:Hn -->`` 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] <url-or-slug> [stuck on...]` and run a session."""
"""Parse `[--local] <url-or-slug> [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] <url-or-slug> [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)

24
leetcode_coach/leetcode_coach/local_coach.py
View File

@ -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})

25
leetcode_coach/leetcode_coach/local_prompt.py
View File

@ -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 `<!-- rung:Hn -->` 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"<!--\s*rung:\s*(H[0-3])\s*-->")
"""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 ``<!-- rung:Hn -->`` 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]

63
leetcode_coach/leetcode_coach/local_session.py
View File

@ -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}]")

191
leetcode_coach/leetcode_coach/phase.py Normal file
View File

@ -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)

69
leetcode_coach/leetcode_coach/report.py Normal file
View File

@ -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

174
leetcode_coach/leetcode_coach/req_check.py Normal file
View File

@ -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": [<vocabulary property names>]}. '
"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]

97
leetcode_coach/leetcode_coach/session_log.py Normal file
View File

@ -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/<YYYY-MM-DD>-<slug>.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))

31
leetcode_coach/leetcode_coach/stop_gate.py
View File

@ -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.

42
leetcode_coach/leetcode_coach/tests/_fake_claude.py Executable file
View File

@ -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())

77
leetcode_coach/leetcode_coach/tests/_fake_ollama.py Normal file
View File

@ -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()

3
leetcode_coach/leetcode_coach/tests/_shared.py
View File

@ -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.

27
leetcode_coach/leetcode_coach/tests/_toolbox_fixture.json Normal file
View File

@ -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"
}
}

31
leetcode_coach/leetcode_coach/tests/conftest.py Normal file
View File

@ -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)

74
leetcode_coach/leetcode_coach/tests/test_claude_cli.py Normal file
View File

@ -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"

181
leetcode_coach/leetcode_coach/tests/test_coach_e2e.py Normal file
View File

@ -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<!-- rung:H0 -->",
"What must the operation do?",
"Closer. What is still open?",
"Try the code?",
"Run it on [3,0,1]?\n\n<!-- rung:H2 -->",
),
)
_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 "<!--" not in out
assert out.rstrip().endswith("Run it on [3,0,1]?")
coach = _calls(root, "coach")
assert coach[0] == "/leetcode missing-number"
assert [c.split("\n")[0] for c in coach[1:4]] == [
"req: find it",
"req: v2",
"req: v3",
]
assert coach[4] == CODE # the first code that reached the model: after 3 lists
assert len(_calls(root, "haiku")) == 3
events = _events(tmp_path, "missing-number")
kinds = [e["event"] for e in events]
assert kinds == ["rung", *["req", "req_check"] * 3, "code", "rung"]
assert [e["verdict"] for e in events if e["event"] == "req_check"] == [
"incomplete"
] * 3
assert events[7]["req_incomplete"] is True
assert events[-1]["rung"] == "H2"
def test_checker_down_fails_open_and_report_rows(
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
capsys: pytest.CaptureFixture[str],
) -> None:
monkeypatch.setattr(local_session, "fetched_statement", lambda _p: STATEMENT)
_script(
tmp_path,
monkeypatch,
haiku=[{"exit": 1, "stderr": "API Error: 500"}],
coach=_coach("Open?", "List noted. Next?", "Run it?\n<!-- rung:H1 -->"),
)
_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<!-- rung:H1 -->",
"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 "<!--" not in out
coach = [m for kind, m in ollama.seen if kind == "coach"]
assert len(coach) == len(["open", "req", "leak", "rewrite", "typed"])
assert "names XOR" in coach[3][-1]["content"]
assert [k for k, _m in ollama.seen].count("checker") == 1
events = _events(tmp_path, "single-number")
assert [e["event"] for e in events] == ["req", "req_check", "rung"]
assert events[1]["verdict"] == "pass"

41
leetcode_coach/leetcode_coach/tests/test_leak.py Normal file
View File

@ -0,0 +1,41 @@
"""Tests for the toolbox-name leak detector shared by the Stop hook and --local."""
from __future__ import annotations
import pytest
from leetcode_coach import leak, toolbox
from leetcode_coach.tests._shared import FIXTURE_TOOLBOX
BOX = toolbox.load(FIXTURE_TOOLBOX)
@pytest.mark.parametrize(
("reply", "names"),
[
("Have you tried XOR?", ["XOR"]),
("what does a ^ b give when a == b?", ["XOR"]),
("Use a Hash\n Map for it.", ["hash map"]),
("A dictionary would help, or the gauss SUM.", ["hash map", "sum formula"]),
("xorshift is a random generator?", []),
("What survives when every pair cancels?", []),
("2^3 is eight?", []),
],
)
def test_leaked_names(reply: str, names: list[str]) -> None:
assert leak.leaked_names(reply, [], BOX) == names
def test_is_it_xor_is_not_bounced() -> None:
"""kuhy asked by name, so the honest answer may say it (else it loops)."""
reply = "Yes -- XOR is the one. What does x ^ x give?"
assert leak.leaked_names(reply, ["is it XOR?"], BOX) == []
def test_any_alias_kuhy_typed_frees_the_whole_entry() -> None:
assert leak.leaked_names("XOR it", ["what about exclusive OR"], BOX) == []
assert leak.leaked_names("XOR it", ["XORing is weird"], BOX) == ["XOR"]
def test_an_empty_toolbox_never_bounces() -> None:
assert leak.leaked_names("XOR", [], toolbox.Toolbox()) == []

41
leetcode_coach/leetcode_coach/tests/test_leetcode_repl.py
View File

@ -2,7 +2,6 @@
from __future__ import annotations
import json
import os
from pathlib import Path
import subprocess
@ -10,7 +9,7 @@ from typing import Any
import pytest
from leetcode_coach import leetcode_repl, local_session
from leetcode_coach import claude_cli, leetcode_repl, local_session, report
LAUNCHER = Path(leetcode_repl.__file__).with_name("leetcode")
@ -29,12 +28,16 @@ LAUNCHER = Path(leetcode_repl.__file__).with_name("leetcode")
],
)
def test_is_limit_hit(reply: dict[str, Any], *, hit: bool) -> None:
assert leetcode_repl.is_limit_hit(reply) is hit
assert claude_cli.is_limit_hit(reply) is hit
def _fake_run(handed: dict[str, Any]) -> Any:
def fake_run(
problem: str, history: list[dict[str, str]], first: str, _r: object
problem: str,
history: list[dict[str, str]],
first: str,
_r: object,
**_kw: object,
) -> int:
handed.update(problem=problem, history=history, first=first)
return 0
@ -126,24 +129,6 @@ def test_read_block_joins_lines_until_a_blank(
assert leetcode_repl.read_block() == "x"
def test_ask_parses_a_failing_claude_stub(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
stub = tmp_path / "claude"
body = json.dumps({"is_error": True, "result": "You've hit your session limit"})
(tmp_path / "out.json").write_text(body)
stub.write_text(f"#!/bin/sh\ncat {tmp_path / 'out.json'}\necho boom >&2\nexit 1\n")
stub.chmod(0o755)
monkeypatch.setattr(leetcode_repl, "CLAUDE", str(stub))
reply = leetcode_repl.ask([], "hi")
assert leetcode_repl.is_limit_hit(reply)
assert reply["_stderr"] == "boom"
stub.write_text("#!/bin/sh\necho not-json\nexit 3\n")
assert leetcode_repl.ask([], "hi")["result"].startswith("[claude exited 3]")
stub.write_text("#!/bin/sh\necho '[1]'\n")
assert leetcode_repl.ask([], "hi") == {"_stderr": ""}
def test_repl_defaults_to_sonnet(monkeypatch: pytest.MonkeyPatch) -> None:
seen: list[str] = []
monkeypatch.delenv("LEETCODE_MODEL", raising=False)
@ -160,6 +145,18 @@ def test_local_flag_skips_claude(monkeypatch: pytest.MonkeyPatch) -> None:
assert handed["first"].endswith("stuck on: loops")
def test_report_is_a_subcommand(monkeypatch: pytest.MonkeyPatch) -> None:
seen: list[list[str]] = []
def fake_report(argv: list[str]) -> int:
seen.append(argv)
return 0
monkeypatch.setattr(report, "main", fake_report)
assert leetcode_repl.main(["report"]) == 0
assert seen == [[]]
def test_help_prints_usage(capsys: pytest.CaptureFixture[str]) -> None:
assert leetcode_repl.main([]) == 0
assert capsys.readouterr().out.startswith(

3
leetcode_coach/leetcode_coach/tests/test_local_coach.py
View File

@ -30,7 +30,8 @@ def test_system_prompt_drops_fetch_step_and_model_gate() -> None:
prompt = local_prompt.system_prompt(skill, STATEMENT)
assert "fetch_problem.py" not in prompt
assert "/model sonnet" not in prompt
assert "## 3. The hint ladder" in prompt
assert "## 3. Requirements first, then the hint ladder" in prompt
assert "<!-- rung:H0 -->" in prompt
assert prompt.rstrip().endswith("```")
assert "var twoSum" in prompt

37
leetcode_coach/leetcode_coach/tests/test_local_session.py
View File

@ -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

221
leetcode_coach/leetcode_coach/tests/test_phase.py Normal file
View File

@ -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<!-- rung:H1 -->\n", "What next?", "H1"),
("A <!-- rung:H0 -->B?<!--rung: H3-->", "A B?", "H3"),
("No marker?", "No marker?", None),
("Bad <!-- rung:H7 -->?", "Bad <!-- rung:H7 -->?", 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<!-- rung:H2 -->") == "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

54
leetcode_coach/leetcode_coach/tests/test_report.py Normal file
View File

@ -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"
)

166
leetcode_coach/leetcode_coach/tests/test_req_check.py Normal file
View File

@ -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

90
leetcode_coach/leetcode_coach/tests/test_session_log.py Normal file
View File

@ -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",
)

98
leetcode_coach/leetcode_coach/tests/test_stop_gate_leak.py Normal file
View File

@ -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": "<command-message>leetcode</command-message>\n"
"<command-name>/leetcode</command-name>"
},
}
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"

57
leetcode_coach/leetcode_coach/tests/test_toolbox.py Normal file
View File

@ -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 == ()

85
leetcode_coach/leetcode_coach/toolbox.py Normal file
View File

@ -0,0 +1,85 @@
"""The toolbox of operations, loaded from ``toolbox.json`` next to this file.
Schema: ``{"entries": [{"name", "aliases", "properties"}], "properties":
{"<prop>": "<one-line meaning>"}}``. 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()