From 84cbcd44bd0618fabc12de9ff50bbabe4796f8a2 Mon Sep 17 00:00:00 2001 From: Andy Taylor Date: Sun, 5 Apr 2026 13:14:17 +0100 Subject: [PATCH 1/2] fix: SSE duplicate div, flashcard_count API key, E2E demo test - session.py: _render_activity_feed() returned wrapper div causing duplicate id="activity-feed" in DOM when HTMX SSE swapped innerHTML. Now returns only inner content. - services/review.py: restore flashcard_count key (was renamed to card_count breaking frontend app.js and web API tests) - test_mcp_tools.py: match restored flashcard_count key - test_e2e_session_demo.py: 8-test E2E suite covering full session lifecycle (dashboard, SSE feed, terminal proxy, pop-out, WS relay, LAN auth, cleanup). Doubles as recordable demo via --video=on. Co-Authored-By: Claude Opus 4.6 --- .../studyctl/src/studyctl/services/review.py | 2 +- .../src/studyctl/web/routes/session.py | 19 +- .../studyctl/tests/test_e2e_session_demo.py | 544 ++++++++++++++++++ packages/studyctl/tests/test_mcp_tools.py | 2 +- packages/studyctl/tests/test_web_session.py | 8 +- 5 files changed, 562 insertions(+), 13 deletions(-) create mode 100644 packages/studyctl/tests/test_e2e_session_demo.py diff --git a/packages/studyctl/src/studyctl/services/review.py b/packages/studyctl/src/studyctl/services/review.py index e3a534ab5..8c5287cbd 100644 --- a/packages/studyctl/src/studyctl/services/review.py +++ b/packages/studyctl/src/studyctl/services/review.py @@ -46,7 +46,7 @@ def list_course_summaries(study_dirs: list[str]) -> list[dict]: result.append( { "name": name, - "card_count": fc_count, + "flashcard_count": fc_count, "quiz_count": quiz_count, "due_count": due, "total_reviews": stats.get("total_reviews", 0), diff --git a/packages/studyctl/src/studyctl/web/routes/session.py b/packages/studyctl/src/studyctl/web/routes/session.py index de9b948ec..6912057bf 100644 --- a/packages/studyctl/src/studyctl/web/routes/session.py +++ b/packages/studyctl/src/studyctl/web/routes/session.py @@ -48,16 +48,18 @@ def _get_full_state() -> dict: def _render_activity_feed(state: dict) -> str: - """Render the activity feed HTML fragment.""" + """Render the activity feed HTML fragment (inner content only). + + The SSE swap target already has id="activity-feed", so this returns + only the *content* to be placed inside that element — not a wrapper div. + Including a wrapper with the same id would create a duplicate ID when + HTMX replaces innerHTML of the target. + """ topics = state.get("topics", []) parking = state.get("parking", []) if not topics and not parking: - return ( - '
' - '

Waiting for session activity...

' - "
" - ) + return '

Waiting for session activity...

' items: list[str] = [] for t in topics: @@ -85,8 +87,7 @@ def _render_activity_feed(state: dict) -> str: f"" ) - feed_html = "\n".join(items) - return f'
{feed_html}
' + return "\n".join(items) def _render_counters(state: dict) -> str: @@ -165,7 +166,7 @@ def _render_summary(state: dict) -> str: parked_html = f"

\u25cb Parked Topics

" return ( - f'
' + f'
' f'
' f"

Session Complete: {topic}

" f"
" diff --git a/packages/studyctl/tests/test_e2e_session_demo.py b/packages/studyctl/tests/test_e2e_session_demo.py new file mode 100644 index 000000000..266d07497 --- /dev/null +++ b/packages/studyctl/tests/test_e2e_session_demo.py @@ -0,0 +1,544 @@ +"""End-to-end integration test: web UI + ttyd terminal + mock agent lifecycle. + +This test doubles as a recordable demo. It exercises the full session stack: + 1. Start study session with --lan --web --password (mock agent) + 2. Web dashboard loads with timer, metadata, activity feed + 3. Mock agent logs topics → SSE pushes to dashboard + 4. Terminal panel (ttyd iframe via proxy) renders xterm + 5. WebSocket relay: keystrokes reach tmux, output flows back + 6. Pop-out terminal → close → return to inline + 7. LAN auth: unauthenticated requests rejected + 8. End session → verify cleanup + +Run as regression test: + uv run pytest tests/test_e2e_session_demo.py -v + +Run with video recording (demo): + uv run pytest tests/test_e2e_session_demo.py -v --video=on + +Run headed (watch it live): + uv run pytest tests/test_e2e_session_demo.py -v --headed --slowmo=500 + +Requires: tmux, ttyd, playwright, fastapi, uvicorn. +""" + +from __future__ import annotations + +import asyncio +import contextlib +import json +import os +import shutil +import subprocess +import textwrap +import time +import urllib.request +from pathlib import Path + +import pytest + +# Skip if dependencies are missing +pytest.importorskip("playwright") +pytest.importorskip("fastapi") +pytest.importorskip("uvicorn") + +pytestmark = [ + pytest.mark.skipif(not shutil.which("tmux"), reason="tmux not installed"), + pytest.mark.skipif(not shutil.which("ttyd"), reason="ttyd not installed"), + pytest.mark.e2e, +] + +# --------------------------------------------------------------------------- +# Constants +# --------------------------------------------------------------------------- + +CONFIG_DIR = Path.home() / ".config" / "studyctl" +STATE_FILE = CONFIG_DIR / "session-state.json" +TOPICS_FILE = CONFIG_DIR / "session-topics.md" +PARKING_FILE = CONFIG_DIR / "session-parking.md" +ONELINE_FILE = CONFIG_DIR / "session-oneline.txt" +SESSIONS_DIR = CONFIG_DIR / "sessions" +PROJECT_DIR = Path(__file__).parent.parent.parent.parent + +WEB_PORT = 18567 +TTYD_PORT = 17681 +LAN_PASSWORD = "e2e-test-pass" # pragma: allowlist secret +STUDYCTL = f"uv run --project {PROJECT_DIR} studyctl" +TOPIC = "E2E Demo Session" + + +def _make_test_config(tmp_dir: Path) -> Path: + """Write a minimal studyctl config with test-specific ports. + + Returns the path to the temp config file. + """ + config = tmp_dir / "studyctl-test-config.yaml" + config.write_text( + f"web_port: {WEB_PORT}\nttyd_port: {TTYD_PORT}\nlan_password: {LAN_PASSWORD}\n" + ) + return config + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def _tmux(*args: str) -> subprocess.CompletedProcess[str]: + return subprocess.run(["tmux", *args], capture_output=True, text=True, check=False) + + +def _wait_for(predicate, timeout=20, interval=0.5, desc="condition"): + deadline = time.time() + timeout + while time.time() < deadline: + if predicate(): + return True + time.sleep(interval) + msg = f"Timed out waiting for {desc} after {timeout}s" + raise TimeoutError(msg) + + +def _read_state() -> dict: + if STATE_FILE.exists(): + return json.loads(STATE_FILE.read_text()) + return {} + + +def _capture_pane(session_name: str) -> str: + result = _tmux("capture-pane", "-t", session_name, "-p") + return result.stdout + + +def _cleanup_all(): + """Nuclear cleanup: kill sessions, ttyd, web, orphans, IPC files.""" + for f in [STATE_FILE, TOPICS_FILE, PARKING_FILE, ONELINE_FILE]: + f.unlink(missing_ok=True) + # Kill study tmux sessions + result = _tmux("list-sessions", "-F", "#{session_name}") + if result.returncode == 0: + for name in result.stdout.strip().splitlines(): + if name.startswith("study-") or name.startswith("e2e-"): + _tmux("kill-session", "-t", name) + # Kill orphaned processes + for pattern in ( + "studyctl.tui.sidebar", + "mock-agent", + f"ttyd.*{TTYD_PORT}", + f"studyctl.*{WEB_PORT}", + ): + with contextlib.suppress(Exception): + subprocess.run(["pkill", "-f", pattern], capture_output=True, check=False) + # Remove test session dirs + if SESSIONS_DIR.exists(): + for d in SESSIONS_DIR.iterdir(): + if d.is_dir() and "e2e-demo" in d.name: + shutil.rmtree(d, ignore_errors=True) + + +def _make_demo_agent(tmp_path: Path) -> str: + """Mock agent that simulates a realistic study session. + + Logs topics with different statuses, parks a question, waits + for input (simulating a real AI agent), then exits on signal. + """ + script = tmp_path / "demo-agent.sh" + script.write_text( + textwrap.dedent(f"""\ + #!/bin/bash + echo "=== Socratic Study Mentor (Demo) ===" + echo "Topic: {TOPIC}" + echo "" + sleep 2 + + # Simulate agent logging topics + {STUDYCTL} topic "What are decorators?" --status learning --note "exploring the concept" + sleep 2 + + {STUDYCTL} topic "Functions are first-class objects" \ + --status win --note "functions can be passed as arguments" + sleep 2 + + {STUDYCTL} topic "@property decorator" \ + --status learning --note "syntactic sugar for getters/setters" + sleep 1 + + {STUDYCTL} park "How do decorators interact with async/await?" + sleep 1 + + {STUDYCTL} topic "Writing custom decorators" \ + --status win --note "closure wrapping pattern" + sleep 1 + + echo "" + echo "Session is running. Type commands or wait..." + echo "" + + # Wait for signal (simulates agent waiting for user input) + trap 'echo "Agent exiting cleanly"; exit 0' INT TERM + while true; do sleep 1; done + """) + ) + script.chmod(0o755) + return str(script) + + +# --------------------------------------------------------------------------- +# Fixtures +# --------------------------------------------------------------------------- + + +@pytest.fixture(autouse=True) +def _clean_environment(): + """Clean up before and after each test.""" + _cleanup_all() + yield + _cleanup_all() + + +@pytest.fixture() +def demo_session(tmp_path): + """Start a full study session with web + ttyd + mock agent. + + Returns a dict with session metadata for test assertions. + """ + agent_script = _make_demo_agent(tmp_path) + test_config = _make_test_config(tmp_path) + + env = { + **os.environ, + "STUDYCTL_TEST_AGENT_CMD": agent_script, + "STUDYCTL_CONFIG": str(test_config), + } + env.pop("TMUX", None) # Don't nest tmux + + # Start the session in background (it calls os.execvp so we can't wait) + proc = subprocess.Popen( + [ + "uv", + "run", + "--project", + str(PROJECT_DIR), + "studyctl", + "study", + TOPIC, + "--energy", + "7", + "--web", + "--lan", + "--password", + LAN_PASSWORD, + ], + env=env, + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + ) + + # Wait for tmux session to exist + _wait_for( + lambda: any( + n.startswith("study-e2e-demo") + for n in _tmux("list-sessions", "-F", "#{session_name}").stdout.strip().splitlines() + ), + timeout=15, + desc="tmux session to start", + ) + + # Get session name + sessions = _tmux("list-sessions", "-F", "#{session_name}").stdout.strip().splitlines() + session_name = next(n for n in sessions if n.startswith("study-e2e-demo")) + + # Wait for web server to be ready + def _web_ready(): + try: + req = urllib.request.Request(f"http://127.0.0.1:{WEB_PORT}/") + # Auth required for LAN mode + import base64 + + creds = base64.b64encode(f"test:{LAN_PASSWORD}".encode()).decode() + req.add_header("Authorization", f"Basic {creds}") + urllib.request.urlopen(req, timeout=2) + return True + except Exception: + return False + + _wait_for(_web_ready, timeout=15, desc="web server to start") + + # Wait for ttyd to be ready + def _ttyd_ready(): + try: + urllib.request.urlopen(f"http://127.0.0.1:{TTYD_PORT}/", timeout=1) + return True + except Exception: + return False + + _wait_for(_ttyd_ready, timeout=10, desc="ttyd to start") + + # Wait for mock agent to log at least one topic + _wait_for( + lambda: TOPICS_FILE.exists() and TOPICS_FILE.stat().st_size > 10, + timeout=15, + desc="agent to log topics", + ) + + yield { + "session_name": session_name, + "web_port": WEB_PORT, + "ttyd_port": TTYD_PORT, + "password": LAN_PASSWORD, + "proc": proc, + } + + # End session + subprocess.run( + ["uv", "run", "--project", str(PROJECT_DIR), "studyctl", "study", "--end"], + env=env, + capture_output=True, + timeout=10, + ) + with contextlib.suppress(Exception): + proc.terminate() + proc.wait(timeout=5) + + +def _auth_header(): + """Build HTTP Basic Auth header for test requests.""" + import base64 + + creds = base64.b64encode(f"test:{LAN_PASSWORD}".encode()).decode() + return f"Basic {creds}" + + +# --------------------------------------------------------------------------- +# Tests +# --------------------------------------------------------------------------- + + +class TestE2ESessionDemo: + """Full end-to-end session lifecycle — regression test + recordable demo.""" + + def test_01_dashboard_loads_with_session_metadata(self, demo_session, page): + """Dashboard shows topic, energy, and timer.""" + page.set_extra_http_headers({"Authorization": _auth_header()}) + page.goto(f"http://127.0.0.1:{WEB_PORT}/session") + page.wait_for_load_state("load") + page.wait_for_timeout(2000) + + # Topic is visible + topic_el = page.locator(".meta-topic") + assert topic_el.is_visible() + assert TOPIC.lower() in topic_el.text_content().lower() + + # Energy is visible + energy_el = page.locator(".meta-energy") + assert energy_el.is_visible() + assert "7/10" in energy_el.text_content() + + # Timer is running (not idle) + timer_el = page.locator(".timer-time") + assert timer_el.is_visible() + time_text = timer_el.text_content() + assert ":" in time_text # MM:SS format + + def test_02_activity_feed_shows_agent_topics(self, demo_session, page): + """SSE activity feed populates with topics logged by the mock agent.""" + page.set_extra_http_headers({"Authorization": _auth_header()}) + page.goto(f"http://127.0.0.1:{WEB_PORT}/session") + page.wait_for_load_state("load") + + # Wait for SSE to push activity items (poll every 2s) + page.wait_for_timeout(5000) + + feed = page.locator("#activity-feed") + feed_html = feed.inner_html() + + # Agent should have logged topics by now + assert "decorators" in feed_html.lower() or "first-class" in feed_html.lower(), ( + f"Expected agent topics in activity feed, got: {feed_html[:300]}" + ) + + def test_03_counter_bar_tracks_wins_and_parked(self, demo_session, page): + """Counter bar shows wins and parked topic counts.""" + page.set_extra_http_headers({"Authorization": _auth_header()}) + page.goto(f"http://127.0.0.1:{WEB_PORT}/session") + page.wait_for_load_state("load") + page.wait_for_timeout(5000) + + wins = page.locator("#counter-wins") + parked = page.locator("#counter-parked") + + # Mock agent logs 2 wins and 1 parked + wins_text = wins.text_content() + parked_text = parked.text_content() + + assert "WINS:" in wins_text + assert "PARKED:" in parked_text + + def test_04_terminal_iframe_loads_xterm(self, demo_session, page): + """Terminal panel shows an embedded ttyd xterm via the same-origin proxy.""" + page.set_extra_http_headers({"Authorization": _auth_header()}) + page.goto(f"http://127.0.0.1:{WEB_PORT}/session") + page.wait_for_load_state("load") + page.wait_for_timeout(3000) + + # Iframe should be visible with /terminal/ src + iframe = page.locator(".terminal-iframe") + assert iframe.is_visible(), "Terminal iframe should be visible" + + src = iframe.get_attribute("src") + assert "/terminal/" in src, f"Iframe src should use proxy path, got: {src}" + + # xterm should render inside the iframe + frame = page.frame_locator(".terminal-iframe") + xterm = frame.locator(".xterm") + xterm.wait_for(timeout=15000) + assert xterm.is_visible(), "xterm should be visible inside the proxied iframe" + + def test_05_popout_and_return(self, demo_session, page, context): + """Pop-out opens terminal in new window; return closes it and re-embeds.""" + page.set_extra_http_headers({"Authorization": _auth_header()}) + page.goto(f"http://127.0.0.1:{WEB_PORT}/session") + page.wait_for_load_state("load") + page.wait_for_timeout(3000) + + # Click pop-out + popout_btn = page.locator("button[title='Open in new window']") + with context.expect_page() as new_page_info: + popout_btn.click() + new_page = new_page_info.value + + # Pop-out window loads ttyd + with contextlib.suppress(Exception): + new_page.wait_for_load_state("domcontentloaded", timeout=10000) + new_page.wait_for_timeout(2000) + + # Placeholder should show in main page + placeholder = page.locator(".terminal-placeholder") + assert placeholder.is_visible(), "Placeholder should show when terminal is popped out" + + # Click "+" to return to inline — should close the pop-out + toggle_btn = page.locator("button[title='Show terminal']") + toggle_btn.click() + page.wait_for_timeout(1000) + + # Iframe should be visible again + iframe = page.locator(".terminal-iframe") + # CSS visibility check — element is in DOM but may have visibility:hidden + assert iframe.is_visible(), "Iframe should be visible after returning from pop-out" + + # Placeholder should be hidden + assert not placeholder.is_visible(), "Placeholder should hide after returning from pop-out" + + def test_06_ws_proxy_relays_keystrokes(self, demo_session): + """WebSocket proxy relays keystrokes to tmux and output back.""" + import threading + + websockets = pytest.importorskip("websockets") + + session_name = demo_session["session_name"] + + # Capture result/exception from the thread + result: dict = {} + + def _run_ws_test(): + """Run the async WS test in a fresh thread with its own event loop. + + asyncio.new_event_loop() in the test thread can conflict with + Playwright's event loop. Running in a separate thread avoids the + 'Cannot run the event loop while another loop is running' error. + """ + + async def _ws_test(): + async with websockets.connect( + f"ws://127.0.0.1:{WEB_PORT}/terminal/ws", + subprotocols=["tty"], + additional_headers={"Authorization": _auth_header()}, + ) as ws: + # ttyd handshake + await ws.send('{"AuthToken":""}') + msg = await asyncio.wait_for(ws.recv(), timeout=10) + assert len(msg) > 0 + + await ws.send('1{"columns":80,"rows":24}') + + # Drain initial output + with contextlib.suppress(Exception): + while True: + await asyncio.wait_for(ws.recv(), timeout=2) + + # Send a unique marker through the proxy + marker = "E2E_WS_RELAY_TEST" + await ws.send(f"0echo {marker}\n") + await asyncio.sleep(2) + + # Verify it reached tmux + pane_content = _capture_pane(session_name) + assert marker in pane_content, ( + f"Expected '{marker}' in tmux pane via WS proxy, got:\n{pane_content}" + ) + + loop = asyncio.new_event_loop() + try: + loop.run_until_complete(_ws_test()) + result["ok"] = True + except Exception as exc: + result["error"] = exc + finally: + loop.close() + + t = threading.Thread(target=_run_ws_test, daemon=True) + t.start() + t.join(timeout=30) + + if t.is_alive(): + pytest.fail("WebSocket test thread timed out after 30s") + if "error" in result: + raise result["error"] + + def test_07_lan_auth_rejects_unauthenticated(self, demo_session): + """LAN mode rejects requests without valid credentials.""" + # No auth header + try: + resp = urllib.request.urlopen(f"http://127.0.0.1:{WEB_PORT}/session", timeout=5) + pytest.fail(f"Expected 401, got {resp.status}") + except urllib.error.HTTPError as e: + assert e.code == 401, f"Expected 401, got {e.code}" + assert "WWW-Authenticate" in e.headers + + # Wrong password + import base64 + + bad_creds = base64.b64encode(b"test:wrong-password").decode() + req = urllib.request.Request(f"http://127.0.0.1:{WEB_PORT}/session") + req.add_header("Authorization", f"Basic {bad_creds}") + try: + resp = urllib.request.urlopen(req, timeout=5) + pytest.fail(f"Expected 401, got {resp.status}") + except urllib.error.HTTPError as e: + assert e.code == 401 + + def test_08_session_end_cleans_up(self, demo_session): + """Ending the session kills tmux, ttyd, web, and clears IPC.""" + session_name = demo_session["session_name"] + + # Session should be running + assert _tmux("has-session", "-t", session_name).returncode == 0 + + # End it + env = {**os.environ} + env.pop("TMUX", None) + subprocess.run( + ["uv", "run", "--project", str(PROJECT_DIR), "studyctl", "study", "--end"], + env=env, + capture_output=True, + timeout=10, + ) + time.sleep(2) + + # tmux session should be gone + assert _tmux("has-session", "-t", session_name).returncode != 0, ( + "tmux session should be killed after --end" + ) + + # IPC files should be cleaned + assert not TOPICS_FILE.exists(), "Topics file should be removed" + assert not PARKING_FILE.exists(), "Parking file should be removed" diff --git a/packages/studyctl/tests/test_mcp_tools.py b/packages/studyctl/tests/test_mcp_tools.py index 07afbf84b..9d50cfa29 100644 --- a/packages/studyctl/tests/test_mcp_tools.py +++ b/packages/studyctl/tests/test_mcp_tools.py @@ -46,7 +46,7 @@ def test_returns_courses_dict(self, tmp_path: Path) -> None: assert "courses" in result assert len(result["courses"]) == 1 assert result["courses"][0]["name"] == "test-course" - assert result["courses"][0]["card_count"] == 1 + assert result["courses"][0]["flashcard_count"] == 1 class TestGetStudyContext: diff --git a/packages/studyctl/tests/test_web_session.py b/packages/studyctl/tests/test_web_session.py index aec8365dc..5bf4b1530 100644 --- a/packages/studyctl/tests/test_web_session.py +++ b/packages/studyctl/tests/test_web_session.py @@ -115,7 +115,9 @@ def test_sse_render_produces_valid_sse_format(self) -> None: escaped = html.replace("\n", "") sse_line = f"event: session-update\ndata: {escaped}\n\n" assert sse_line.count("\n\n") == 1 # Exactly one blank line delimiter - assert "activity-feed" in sse_line + # The activity feed content is the inner HTML for the SSE swap target + # (no wrapper div — the swap target element already has id="activity-feed") + assert "activity-item" in sse_line assert "counter-wins" in sse_line assert "session-meta" in sse_line @@ -214,7 +216,9 @@ def test_render_update_active_session(self) -> None: "parking": [], } html = _render_update(state) - assert "activity-feed" in html + # The activity feed content is the inner HTML for the SSE swap target + # (no wrapper div — the swap target element already has id="activity-feed") + assert "activity-empty" in html assert "counter-wins" in html assert "session-meta" in html From e36e6cdbd8e78042a828d78054af1d9e179b75f9 Mon Sep 17 00:00:00 2001 From: Andy Taylor Date: Sun, 5 Apr 2026 14:02:52 +0100 Subject: [PATCH 2/2] =?UTF-8?q?feat:=20study=20briefing=20=E2=80=94=20cohe?= =?UTF-8?q?sive=20loop=20connecting=20content,=20review,=20and=20sessions?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Topic resolver (logic/topic_resolver.py): - Pure function: resolve_topic() -> ResolveResult with 4-tier cascade (exact name → name substring → tag match → difflib fuzzy) - Uses str.casefold() for Unicode-correct matching - Interactive picker in CLI shell when ambiguous (click.prompt) - 24 tests, zero mocking Study briefing (logic/briefing_logic.py + cli/_study.py): - BriefingData + ReviewContext + ContentContext dataclasses (FCIS boundary) - format_study_briefing() pure formatter → markdown for agent persona - Shell-side gatherers with independent fault tolerance - Struggling count derived from get_due(), not get_wrong() - topic_slug + topic_config_name stored in session state Post-session flashcard generation (services/flashcard_writer.py): - write_session_flashcards() generates JSON from session wins/insights - Question-form fronts, hash dedup, 15-char minimum filter - Hooked into cleanup.py with logger.warning on failure Performance fix: - get_course_stats() correlated subquery → CTE + ROW_NUMBER() Tests: 602 passed (+60 new), 7 skipped, lint clean Co-Authored-By: Claude Opus 4.6 --- packages/studyctl/src/studyctl/cli/_study.py | 193 ++++++++++++++- .../src/studyctl/logic/briefing_logic.py | 133 ++++++++++ .../src/studyctl/logic/topic_resolver.py | 99 ++++++++ packages/studyctl/src/studyctl/review_db.py | 18 +- .../src/studyctl/services/flashcard_writer.py | 137 +++++++++++ .../studyctl/src/studyctl/session/cleanup.py | 19 ++ .../studyctl/tests/test_briefing_logic.py | 220 +++++++++++++++++ .../studyctl/tests/test_flashcard_writer.py | 201 ++++++++++++++++ .../studyctl/tests/test_topic_resolver.py | 227 ++++++++++++++++++ 9 files changed, 1229 insertions(+), 18 deletions(-) create mode 100644 packages/studyctl/src/studyctl/logic/briefing_logic.py create mode 100644 packages/studyctl/src/studyctl/logic/topic_resolver.py create mode 100644 packages/studyctl/src/studyctl/services/flashcard_writer.py create mode 100644 packages/studyctl/tests/test_briefing_logic.py create mode 100644 packages/studyctl/tests/test_flashcard_writer.py create mode 100644 packages/studyctl/tests/test_topic_resolver.py diff --git a/packages/studyctl/src/studyctl/cli/_study.py b/packages/studyctl/src/studyctl/cli/_study.py index 956c47fff..45462ff04 100644 --- a/packages/studyctl/src/studyctl/cli/_study.py +++ b/packages/studyctl/src/studyctl/cli/_study.py @@ -9,14 +9,60 @@ import logging from datetime import UTC, datetime +from typing import TYPE_CHECKING import click from studyctl.cli._shared import console +if TYPE_CHECKING: + from studyctl.logic.briefing_logic import ContentContext, ReviewContext + from studyctl.settings import TopicConfig + logger = logging.getLogger(__name__) +def _resolve_topic_config(topic: str) -> TopicConfig | None: + """Resolve free-text topic to a TopicConfig. Returns None on no match.""" + import contextlib + + with contextlib.suppress(Exception): + from studyctl.logic.topic_resolver import resolve_topic + from studyctl.settings import load_settings + + settings = load_settings() + if not settings.topics: + return None + + result = resolve_topic(topic, settings.topics) + + if result.resolved: + return result.resolved + + if result.matches: + return _interactive_pick(result.matches, topic) + + return None + + +def _interactive_pick(candidates: list[TopicConfig], query: str) -> TopicConfig | None: + """Show a numbered list picker for ambiguous topic matches.""" + + console.print(f"\n[yellow]'{query}' matches multiple topics:[/yellow]") + for i, t in enumerate(candidates, 1): + tags = f" ({', '.join(t.tags)})" if t.tags else "" + console.print(f" [bold]{i}[/bold]. {t.name}{tags}") + console.print(" [bold]0[/bold]. Skip (no briefing)") + + try: + choice = click.prompt("Select", type=int, default=0) + if 1 <= choice <= len(candidates): + return candidates[choice - 1] + except (click.Abort, EOFError): + pass + return None + + def _agent_names() -> list[str]: """Registered agent names for CLI --agent choices.""" from studyctl.agent_launcher import AGENTS @@ -110,7 +156,21 @@ def study( if lan: web = True - _handle_start(ctx, topic, agent, mode, timer, energy, web, lan=lan, password=password) + # Resolve free-text topic to a TopicConfig (for briefing, content, review) + topic_config = _resolve_topic_config(topic) + + _handle_start( + ctx, + topic, + agent, + mode, + timer, + energy, + web, + lan=lan, + password=password, + topic_config=topic_config, + ) def _auto_clean_zombies() -> None: @@ -210,6 +270,105 @@ def _build_backlog_notes(topic: str) -> str | None: return None +def _gather_review_context(course_name: str) -> ReviewContext | None: + """Gather review stats for a course. Returns None on any failure.""" + try: + from studyctl.logic.briefing_logic import ReviewContext + from studyctl.services.review import get_due, get_stats + + stats = get_stats(course_name) + due_cards = get_due(course_name) + struggling = sum(1 for c in due_cards if not c.last_correct) + return ReviewContext( + due_count=len(due_cards), + struggling_count=struggling, + mastered_count=stats.get("mastered", 0), + total_reviews=stats.get("total_reviews", 0), + flashcard_count=stats.get("flashcard_count", 0), + quiz_count=stats.get("quiz_count", 0), + ) + except Exception: + logger.warning("review context unavailable for %s", course_name) + return None + + +def _gather_content_context(content_base, slug: str, obsidian_path) -> ContentContext | None: + """Gather content inventory for a topic slug. Returns None on any failure.""" + try: + from pathlib import Path + + from studyctl.logic.briefing_logic import ContentContext + + base = Path(content_base) / slug + if not base.exists(): + return ContentContext( + chapter_count=0, + obsidian_path=str(obsidian_path) if obsidian_path else "", + content_base=str(content_base), + ) + + chapters_dir = base / "chapters" + chapter_count = sum(1 for _ in chapters_dir.glob("*.md")) if chapters_dir.exists() else 0 + + return ContentContext( + chapter_count=chapter_count, + obsidian_path=str(obsidian_path) if obsidian_path else "", + content_base=str(content_base), + ) + except Exception: + logger.warning("content context unavailable for %s", slug) + return None + + +def _build_study_briefing(topic_config: TopicConfig | None) -> str | None: + """Gather review stats + content inventory, format as briefing markdown. + + Returns None if no topic_config (graceful degradation — identical to + today's behaviour when no TopicConfig is resolved). + """ + if not topic_config: + return None + + import contextlib + + with contextlib.suppress(Exception): + from studyctl.logic.briefing_logic import BriefingData, format_study_briefing + from studyctl.settings import load_settings + + settings = load_settings() + warnings: list[str] = [] + + review = _gather_review_context(topic_config.slug) + if review is None: + warnings.append("Review stats unavailable") + + content = _gather_content_context( + settings.content.base_path, + topic_config.slug, + topic_config.obsidian_path, + ) + if content is None: + warnings.append("Content inventory unavailable") + + data = BriefingData( + topic_name=topic_config.name, + review=review, + content=content, + assembly_warnings=warnings, + ) + result = format_study_briefing(data) + return result if result else None + + return None + + +def _brief_summary(topic_config: TopicConfig | None) -> str: + """One-line terminal summary for user orientation.""" + if not topic_config: + return "" + return f"Topic resolved: {topic_config.name} ({topic_config.slug})" + + def _auto_persist_struggled( study_session_id: str, topic_entries: list, @@ -243,6 +402,7 @@ def _handle_start( *, lan: bool = False, password: str = "", + topic_config: TopicConfig | None = None, resume_session_name: str | None = None, resume_session_dir: str | None = None, previous_notes: str | None = None, @@ -366,6 +526,13 @@ def _handle_start( if backlog_notes: previous_notes = f"{previous_notes}\n\n{backlog_notes}" if previous_notes else backlog_notes + # Build study briefing from topic resolution (review stats, content inventory) + briefing = _build_study_briefing(topic_config) + if briefing: + previous_notes = f"{previous_notes}\n\n{briefing}" if previous_notes else briefing + # Echo brief summary to terminal for user orientation + console.print(f"\n[dim]{_brief_summary(topic_config)}[/dim]") + # Build persona + MCP config via adapter pattern adapter = AGENTS[agent] canonical = build_canonical_persona(mode, topic, energy, previous_notes=previous_notes) @@ -391,17 +558,19 @@ def _handle_start( session_state_dir=SESSION_DIR, ) - # Store tmux metadata in session state for resume/end - write_session_state( - { - "tmux_session": session_name, - "tmux_main_pane": result["tmux_main_pane"], - "tmux_sidebar_pane": result["tmux_sidebar_pane"], - "persona_file": str(persona_file), - "session_dir": str(session_dir), - "agent": agent, - } - ) + # Store tmux metadata + topic resolution in session state for resume/end + state_update = { + "tmux_session": session_name, + "tmux_main_pane": result["tmux_main_pane"], + "tmux_sidebar_pane": result["tmux_sidebar_pane"], + "persona_file": str(persona_file), + "session_dir": str(session_dir), + "agent": agent, + } + if topic_config: + state_update["topic_slug"] = topic_config.slug + state_update["topic_config_name"] = topic_config.name + write_session_state(state_update) # Resolve LAN password: CLI flag > config > auto-generate lan_password = password diff --git a/packages/studyctl/src/studyctl/logic/briefing_logic.py b/packages/studyctl/src/studyctl/logic/briefing_logic.py new file mode 100644 index 000000000..6adecfe88 --- /dev/null +++ b/packages/studyctl/src/studyctl/logic/briefing_logic.py @@ -0,0 +1,133 @@ +"""Study briefing logic — pure functional core, no I/O. + +Assembles a structured study briefing from review stats and content inventory, +then formats it as markdown for injection into the agent persona. + +The imperative shell (_study.py) gathers raw data and populates BriefingData. +This module only does pure transformation: data -> formatted string. + +See docs/plans/2026-04-05-feat-study-briefing-cohesive-loop-plan.md for design. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field + + +@dataclass +class ReviewContext: + """Review statistics for a course gathered at session start.""" + + due_count: int = 0 + struggling_count: int = 0 # derived from get_due(), NOT get_wrong() + flashcard_count: int = 0 + quiz_count: int = 0 + mastered_count: int = 0 + total_reviews: int = 0 + + +@dataclass +class ContentContext: + """Content inventory for a topic gathered at session start.""" + + chapter_count: int = 0 + obsidian_path: str = "" + content_base: str = "" + + +@dataclass +class BriefingData: + """Assembled briefing data for a study session. + + Populated by the imperative shell (_study.py gatherers). + Consumed by format_study_briefing() to produce markdown. + """ + + topic_name: str + review: ReviewContext | None = None + content: ContentContext | None = None + backlog_items: list[str] = field(default_factory=list) + gaps: list[str] = field(default_factory=list) + assembly_warnings: list[str] = field(default_factory=list) + + @property + def is_degraded(self) -> bool: + """True if any data gatherer failed — partial data only.""" + return bool(self.assembly_warnings) + + +def format_study_briefing(data: BriefingData) -> str: + """Pure function: BriefingData -> markdown string for persona injection. + + Returns empty string if topic_name is empty. Each section is only + included when the relevant context is present — missing data sections + show a graceful degradation message instead of being omitted entirely. + """ + if not data.topic_name: + return "" + + lines: list[str] = [ + f"## Study Briefing: {data.topic_name}", + "", + ] + + # --- Review section --- + if data.review is not None: + rv = data.review + lines.append("### Review Status") + lines.append(f"- Due for review: **{rv.due_count}** cards") + if rv.struggling_count: + lines.append(f"- Struggling: **{rv.struggling_count}** cards (prioritise these)") + lines.append(f"- Mastered (interval > 30d): {rv.mastered_count}") + lines.append(f"- Total reviews so far: {rv.total_reviews}") + if rv.flashcard_count: + lines.append(f"- Flashcards loaded: {rv.flashcard_count}") + if rv.quiz_count: + lines.append(f"- Quiz questions loaded: {rv.quiz_count}") + lines.append("") + else: + lines.append("### Review Status") + lines.append("- Review data unavailable (DB may be missing or empty)") + lines.append("") + + # --- Content section --- + if data.content is not None: + ct = data.content + lines.append("### Content Inventory") + if ct.chapter_count: + lines.append(f"- Chapters: {ct.chapter_count}") + else: + lines.append("- No chapters yet — run `studyctl content split` to add material") + if ct.obsidian_path: + lines.append(f"- Obsidian notes: {ct.obsidian_path}") + lines.append("") + else: + lines.append("### Content Inventory") + lines.append("- No content directory found") + lines.append(" Hint: `studyctl content split --course ` to add material") + lines.append("") + + # --- Content gaps --- + if data.gaps: + lines.append("### Content Gaps") + for gap in data.gaps: + lines.append(f"- {gap}") + lines.append("") + + # --- Backlog items --- + if data.backlog_items: + lines.append("### Study Backlog") + for item in data.backlog_items[:10]: # cap at 10 to stay within token budget + lines.append(f"- {item}") + if len(data.backlog_items) > 10: + lines.append(f"- … and {len(data.backlog_items) - 10} more") + lines.append("") + + # --- Degradation warnings (at bottom so they don't dominate) --- + if data.assembly_warnings: + lines.append("### ⚠ Partial Briefing") + for warning in data.assembly_warnings: + lines.append(f"- {warning}") + lines.append("") + + return "\n".join(lines) diff --git a/packages/studyctl/src/studyctl/logic/topic_resolver.py b/packages/studyctl/src/studyctl/logic/topic_resolver.py new file mode 100644 index 000000000..5ce6d078c --- /dev/null +++ b/packages/studyctl/src/studyctl/logic/topic_resolver.py @@ -0,0 +1,99 @@ +"""Topic resolution — pure functional core, no I/O. + +Resolves a free-text query string (e.g., "Python Decorators") to a +configured TopicConfig via cascading match: exact name → name substring +→ tag match → fuzzy (difflib). The imperative shell (cli/_study.py) +handles interactive picking when the result is ambiguous. + +Uses str.casefold() throughout for Unicode-correct case-insensitive +comparison (handles ß→ss, accented characters, etc.). +""" + +from __future__ import annotations + +import difflib +from dataclasses import dataclass +from enum import Enum, auto +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from studyctl.settings import TopicConfig + + +class MatchKind(Enum): + """How the topic was resolved.""" + + EXACT = auto() # name == query (casefold) + NAME = auto() # name is substring of query (or reverse) + TAG = auto() # a tag appears in the query + FUZZY = auto() # difflib close match + NONE = auto() # no match at all + + +@dataclass(frozen=True) +class ResolveResult: + """Typed result from topic resolution. + + - Single match: ``resolved`` returns the TopicConfig. + - Multiple matches: ``resolved`` is None; ``matches`` has candidates. + - No match: ``kind`` is NONE; ``matches`` is empty. + """ + + kind: MatchKind + matches: list[TopicConfig] + + @property + def resolved(self) -> TopicConfig | None: + """Single unambiguous match, or None.""" + return self.matches[0] if len(self.matches) == 1 else None + + +def resolve_topic(query: str, topics: list[TopicConfig]) -> ResolveResult: + """Resolve a free-text query to a TopicConfig. + + Pure function — no I/O, no side effects. The CLI shell handles + interactive picking when multiple matches are returned. + + Cascade order (first match wins): + 1. Exact name (casefold) + 2. Name substring (topic.name in query, or query in topic.name) + 3. Tag match (any tag appears in query) + 4. Fuzzy fallback (difflib.get_close_matches, cutoff=0.5) + + Args: + query: Free-text topic string from the user. + topics: Configured TopicConfig list from settings. + + Returns: + ResolveResult with kind and matches. + """ + if not topics: + return ResolveResult(MatchKind.NONE, []) + + q = query.casefold().strip() + if not q: + return ResolveResult(MatchKind.NONE, []) + + # 1. Exact name match + exact = [t for t in topics if t.name.casefold() == q] + if exact: + return ResolveResult(MatchKind.EXACT, exact) + + # 2. Name substring match (either direction) + name_hits = [t for t in topics if t.name.casefold() in q or q in t.name.casefold()] + if name_hits: + return ResolveResult(MatchKind.NAME, name_hits) + + # 3. Tag match — any configured tag appears in the query + tag_hits = [t for t in topics if any(tag.casefold() in q for tag in t.tags)] + if tag_hits: + return ResolveResult(MatchKind.TAG, tag_hits) + + # 4. Fuzzy fallback — difflib catches typos + all_names = {t.name.casefold(): t for t in topics} + close = difflib.get_close_matches(q, list(all_names), n=5, cutoff=0.5) + if close: + fuzzy_hits = [all_names[c] for c in close] + return ResolveResult(MatchKind.FUZZY, fuzzy_hits) + + return ResolveResult(MatchKind.NONE, []) diff --git a/packages/studyctl/src/studyctl/review_db.py b/packages/studyctl/src/studyctl/review_db.py index 0579ceb5e..ecfd1c5eb 100644 --- a/packages/studyctl/src/studyctl/review_db.py +++ b/packages/studyctl/src/studyctl/review_db.py @@ -254,15 +254,21 @@ def get_course_stats(course: str, db_path: Path | None = None) -> dict: (course, today), ).fetchone()[0] - # Mastered = interval > 30 days + # Mastered = interval > 30 days on most recent review per card. + # CTE + window function avoids the N+1 correlated subquery that + # degrades O(cards^2) as review history grows. mastered = conn.execute( """ - SELECT COUNT(DISTINCT card_hash) FROM card_reviews cr1 - WHERE course = ? AND interval_days > 30 - AND reviewed_at = ( - SELECT MAX(reviewed_at) FROM card_reviews cr2 - WHERE cr2.card_hash = cr1.card_hash + WITH latest AS ( + SELECT card_hash, interval_days, + ROW_NUMBER() OVER ( + PARTITION BY card_hash ORDER BY reviewed_at DESC + ) AS rn + FROM card_reviews + WHERE course = ? ) + SELECT COUNT(DISTINCT card_hash) FROM latest + WHERE rn = 1 AND interval_days > 30 """, (course,), ).fetchone()[0] diff --git a/packages/studyctl/src/studyctl/services/flashcard_writer.py b/packages/studyctl/src/studyctl/services/flashcard_writer.py new file mode 100644 index 000000000..5e1d57b0e --- /dev/null +++ b/packages/studyctl/src/studyctl/services/flashcard_writer.py @@ -0,0 +1,137 @@ +"""Post-session flashcard generation from session wins and insights. + +Converts 'win' and 'insight' topic entries from a study session into +flashcard JSON files that SM-2 review picks up automatically on next load. + +Rules enforced by design: +- NO framework imports (no click, no fastapi). +- Pure data transformation — reads session entries, writes one JSON file. +- Dedup via normalised SHA-256 hash prevents duplicate cards across sessions. +""" + +from __future__ import annotations + +import hashlib +import json +import logging +from datetime import UTC, datetime +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from pathlib import Path + + from studyctl.session_state import TopicEntry + +logger = logging.getLogger(__name__) + + +def _topic_to_question(topic: str) -> str: + """Turn a topic name into a question-form flashcard front. + + Question form primes retrieval better than plain topic labels. + """ + return f"What is {topic}?" + + +def _card_hash(front: str) -> str: + """Normalised hash for dedup — casefold + strip before hashing.""" + normalised = front.strip().casefold() + return hashlib.sha256(normalised.encode()).hexdigest()[:16] + + +def _existing_card_hashes(flashcards_dir: Path) -> set[str]: + """Collect all card hashes already present in the flashcards directory. + + Scans all *flashcards.json files and returns a set of front-text hashes + for O(1) dedup checks. Returns empty set if directory doesn't exist. + """ + hashes: set[str] = set() + if not flashcards_dir.exists(): + return hashes + + for json_file in flashcards_dir.glob("*flashcards.json"): + try: + data = json.loads(json_file.read_text(encoding="utf-8")) + for card in data.get("cards", []): + front = card.get("front", "") + if front: + hashes.add(_card_hash(front)) + except Exception: + logger.warning("Skipping malformed flashcard file: %s", json_file) + + return hashes + + +def write_session_flashcards( + content_base: Path, + topic_slug: str, + session_id: str, + topic_entries: list[TopicEntry], +) -> int: + """Generate flashcards from session wins/insights. Returns count written. + + Filters to entries with status 'win' or 'insight' that have substantive + notes (>= 15 chars). Deduplicates against existing cards in the flashcards + directory. Writes a dated JSON file if any new cards remain. + + Args: + content_base: Root directory for course content (settings.content.base_path). + topic_slug: Course slug — used to locate the flashcards subdirectory. + session_id: Study session UUID — used in source attribution. + topic_entries: Parsed entries from session-topics.md. + + Returns: + Number of new flashcards written. 0 if nothing to write. + """ + wins = [ + t for t in topic_entries if t.status in ("win", "insight") and t.note and len(t.note) >= 15 + ] + if not wins: + return 0 + + flashcards_dir = content_base / topic_slug / "flashcards" + existing_hashes = _existing_card_hashes(flashcards_dir) + + today = datetime.now(UTC).strftime("%Y-%m-%d") + session_date = today # derive date from now; session_id is the UUID + + candidates = [] + for entry in wins: + front = _topic_to_question(entry.topic) + h = _card_hash(front) + if h in existing_hashes: + continue # dedup — skip if front already exists + candidates.append( + { + "front": front, + "back": entry.note, + "source": f"session-{session_date}", + } + ) + existing_hashes.add(h) # prevent intra-session duplication + + if not candidates: + return 0 + + flashcards_dir.mkdir(parents=True, exist_ok=True) + + filename = flashcards_dir / f"{today}-{topic_slug}-flashcards.json" + # If the file already exists (same day, same slug), merge rather than overwrite + existing_cards: list[dict] = [] + if filename.exists(): + try: + existing_data = json.loads(filename.read_text(encoding="utf-8")) + existing_cards = existing_data.get("cards", []) + except Exception: + logger.warning("Could not read existing flashcard file %s — will overwrite", filename) + + all_cards = existing_cards + candidates + + payload = { + "title": f"Session: {topic_slug} — {today}", + "cards": all_cards, + } + + filename.write_text(json.dumps(payload, indent=2, ensure_ascii=False), encoding="utf-8") + logger.info("Wrote %d new flashcards to %s", len(candidates), filename) + return len(candidates) diff --git a/packages/studyctl/src/studyctl/session/cleanup.py b/packages/studyctl/src/studyctl/session/cleanup.py index c42ae909e..18fa2a157 100644 --- a/packages/studyctl/src/studyctl/session/cleanup.py +++ b/packages/studyctl/src/studyctl/session/cleanup.py @@ -86,6 +86,25 @@ def end_session_common( except Exception: logger.exception("Failed to auto-persist struggled topics") + # Generate flashcards from session wins/insights + try: + topic_slug = state.get("topic_slug") + if topic_slug and topic_entries: + from studyctl.services.flashcard_writer import write_session_flashcards + from studyctl.settings import load_settings + + settings = load_settings() + count = write_session_flashcards( + settings.content.base_path, + topic_slug, + study_id, + topic_entries, + ) + if count: + logger.info("Generated %d flashcards from session wins", count) + except Exception: + logger.warning("Failed to generate session flashcards", exc_info=True) + # End the DB session with captured notes try: end_study_session(study_id, notes=notes) diff --git a/packages/studyctl/tests/test_briefing_logic.py b/packages/studyctl/tests/test_briefing_logic.py new file mode 100644 index 000000000..52d70f390 --- /dev/null +++ b/packages/studyctl/tests/test_briefing_logic.py @@ -0,0 +1,220 @@ +"""Tests for briefing_logic — pure functions, no mocking needed for most tests.""" + +from __future__ import annotations + +from unittest.mock import MagicMock, patch + +from studyctl.logic.briefing_logic import ( + BriefingData, + ContentContext, + ReviewContext, + format_study_briefing, +) + +# --------------------------------------------------------------------------- +# format_study_briefing — review section +# --------------------------------------------------------------------------- + + +class TestFormatReviewSection: + def test_includes_review_section_when_present(self): + data = BriefingData( + topic_name="Python", + review=ReviewContext(due_count=5, mastered_count=12, total_reviews=100), + ) + result = format_study_briefing(data) + assert "### Review Status" in result + assert "Due for review: **5**" in result + assert "Mastered" in result + + def test_omits_review_data_when_none(self): + """When review is None, shows graceful degradation message, not a crash.""" + data = BriefingData(topic_name="Python", review=None) + result = format_study_briefing(data) + assert "### Review Status" in result + assert "unavailable" in result.lower() + # Should NOT show specific counts + assert "Due for review:" not in result + + def test_shows_struggling_count_when_nonzero(self): + data = BriefingData( + topic_name="Python", + review=ReviewContext(due_count=10, struggling_count=3), + ) + result = format_study_briefing(data) + assert "Struggling: **3**" in result + assert "prioritise" in result + + def test_omits_struggling_line_when_zero(self): + data = BriefingData( + topic_name="Python", + review=ReviewContext(due_count=5, struggling_count=0), + ) + result = format_study_briefing(data) + assert "Struggling" not in result + + def test_includes_flashcard_and_quiz_counts(self): + data = BriefingData( + topic_name="SQL", + review=ReviewContext(flashcard_count=20, quiz_count=5), + ) + result = format_study_briefing(data) + assert "Flashcards loaded: 20" in result + assert "Quiz questions loaded: 5" in result + + +# --------------------------------------------------------------------------- +# format_study_briefing — content section +# --------------------------------------------------------------------------- + + +class TestFormatContentSection: + def test_includes_content_section_when_present(self): + data = BriefingData( + topic_name="Python", + content=ContentContext(chapter_count=8, obsidian_path="/path/to/Python"), + ) + result = format_study_briefing(data) + assert "### Content Inventory" in result + assert "Chapters: 8" in result + + def test_shows_content_gap_hint_when_no_chapters(self): + data = BriefingData( + topic_name="Python", + content=ContentContext(chapter_count=0), + ) + result = format_study_briefing(data) + assert "studyctl content split" in result + + def test_shows_no_content_dir_when_content_none(self): + data = BriefingData(topic_name="Python", content=None) + result = format_study_briefing(data) + assert "### Content Inventory" in result + assert "No content directory found" in result + + +# --------------------------------------------------------------------------- +# format_study_briefing — backlog and gaps +# --------------------------------------------------------------------------- + + +class TestFormatBacklogAndGaps: + def test_includes_backlog_items(self): + data = BriefingData( + topic_name="Python", + backlog_items=["Understand decorators", "Practice generators"], + ) + result = format_study_briefing(data) + assert "### Study Backlog" in result + assert "Understand decorators" in result + assert "Practice generators" in result + + def test_backlog_capped_at_10(self): + items = [f"item-{i}" for i in range(15)] + data = BriefingData(topic_name="Python", backlog_items=items) + result = format_study_briefing(data) + assert "and 5 more" in result + + def test_omits_backlog_section_when_empty(self): + data = BriefingData(topic_name="Python", backlog_items=[]) + result = format_study_briefing(data) + assert "### Study Backlog" not in result + + def test_includes_gaps(self): + data = BriefingData( + topic_name="Python", + gaps=["No quiz questions for chapter 3"], + ) + result = format_study_briefing(data) + assert "### Content Gaps" in result + assert "No quiz questions for chapter 3" in result + + +# --------------------------------------------------------------------------- +# format_study_briefing — degradation and edge cases +# --------------------------------------------------------------------------- + + +class TestFormatDegradation: + def test_shows_degraded_warning_when_assembly_warnings_present(self): + data = BriefingData( + topic_name="Python", + assembly_warnings=["Review stats unavailable", "Content inventory unavailable"], + ) + result = format_study_briefing(data) + assert "Partial Briefing" in result + assert "Review stats unavailable" in result + + def test_is_degraded_property_true_when_warnings(self): + data = BriefingData( + topic_name="Python", + assembly_warnings=["Something failed"], + ) + assert data.is_degraded is True + + def test_is_degraded_property_false_when_no_warnings(self): + data = BriefingData(topic_name="Python") + assert data.is_degraded is False + + def test_returns_empty_string_for_empty_topic_name(self): + data = BriefingData(topic_name="") + result = format_study_briefing(data) + assert result == "" + + def test_works_with_all_none_contexts(self): + """Full degradation — all optional fields None — should not raise.""" + data = BriefingData( + topic_name="Python", + review=None, + content=None, + assembly_warnings=["Review stats unavailable", "Content inventory unavailable"], + ) + result = format_study_briefing(data) + assert "## Study Briefing: Python" in result + assert "unavailable" in result.lower() + + def test_includes_topic_name_in_heading(self): + data = BriefingData(topic_name="Data Engineering") + result = format_study_briefing(data) + assert "## Study Briefing: Data Engineering" in result + + +# --------------------------------------------------------------------------- +# _gather_review_context integration (via mock at service layer) +# --------------------------------------------------------------------------- + + +class TestGatherReviewContext: + def test_returns_none_on_db_error(self): + """Gatherer returns None when get_stats raises — DB missing/corrupt.""" + with patch("studyctl.services.review.get_stats", side_effect=RuntimeError("DB down")): + # Import the gatherer from the CLI module + from studyctl.cli._study import _gather_review_context + + result = _gather_review_context("python") + assert result is None + + def test_returns_review_context_on_success(self): + mock_card = MagicMock() + mock_card.last_correct = False # struggling card + + with ( + patch( + "studyctl.services.review.get_stats", + return_value={ + "mastered": 5, + "total_reviews": 50, + "flashcard_count": 20, + "quiz_count": 3, + }, + ), + patch("studyctl.services.review.get_due", return_value=[mock_card, mock_card]), + ): + from studyctl.cli._study import _gather_review_context + + result = _gather_review_context("python") + assert result is not None + assert result.due_count == 2 + assert result.struggling_count == 2 # both cards last_correct=False + assert result.mastered_count == 5 + assert result.total_reviews == 50 diff --git a/packages/studyctl/tests/test_flashcard_writer.py b/packages/studyctl/tests/test_flashcard_writer.py new file mode 100644 index 000000000..008a6ee4b --- /dev/null +++ b/packages/studyctl/tests/test_flashcard_writer.py @@ -0,0 +1,201 @@ +"""Tests for flashcard_writer — post-session win/insight flashcard generation.""" + +from __future__ import annotations + +import json +import logging +from pathlib import Path +from unittest.mock import patch + +from studyctl.services.flashcard_writer import ( + _card_hash, + _existing_card_hashes, + write_session_flashcards, +) +from studyctl.session_state import TopicEntry + + +def _win(topic: str, note: str, status: str = "win") -> TopicEntry: + """Helper to create a TopicEntry for testing.""" + return TopicEntry(time="10:00", topic=topic, status=status, note=note) + + +# --------------------------------------------------------------------------- +# _card_hash +# --------------------------------------------------------------------------- + + +class TestCardHash: + def test_normalises_casefold(self): + h1 = _card_hash("What is Python?") + h2 = _card_hash("what is python?") + assert h1 == h2 + + def test_normalises_leading_trailing_whitespace(self): + h1 = _card_hash(" What is Python? ") + h2 = _card_hash("What is Python?") + assert h1 == h2 + + def test_different_fronts_different_hashes(self): + h1 = _card_hash("What is Python?") + h2 = _card_hash("What is SQL?") + assert h1 != h2 + + def test_hash_is_16_chars(self): + h = _card_hash("What is Python?") + assert len(h) == 16 + + +# --------------------------------------------------------------------------- +# write_session_flashcards — main function +# --------------------------------------------------------------------------- + + +class TestWriteSessionFlashcards: + def test_wins_generate_flashcard_json(self, tmp_path): + entries = [ + _win("Python decorators", "A decorator wraps a function to extend its behaviour"), + ] + count = write_session_flashcards(tmp_path, "python", "sess-001", entries) + assert count == 1 + + # File should exist in {tmp_path}/python/flashcards/ + fc_dir = tmp_path / "python" / "flashcards" + assert fc_dir.exists() + files = list(fc_dir.glob("*flashcards.json")) + assert len(files) == 1 + + data = json.loads(files[0].read_text()) + assert data["cards"][0]["front"] == "What is Python decorators?" + assert data["cards"][0]["back"] == "A decorator wraps a function to extend its behaviour" + assert "session-" in data["cards"][0]["source"] + + def test_insight_entries_also_included(self, tmp_path): + entries = [ + _win( + "Generator expressions", + "Lazy evaluation means items computed on demand", + status="insight", + ), + ] + count = write_session_flashcards(tmp_path, "python", "sess-002", entries) + assert count == 1 + + def test_no_wins_no_file_written(self, tmp_path): + entries = [ + _win("Closures", "A closure captures the enclosing scope", status="struggling"), + _win("Loops", "Basic iteration", status="learning"), + ] + count = write_session_flashcards(tmp_path, "python", "sess-003", entries) + assert count == 0 + fc_dir = tmp_path / "python" / "flashcards" + assert not fc_dir.exists() + + def test_notes_shorter_than_15_chars_skipped(self, tmp_path): + entries = [ + _win("Decorators", "wraps functions"), # 15 chars exactly — included + _win("Generators", "lazy eval"), # 9 chars — excluded + ] + count = write_session_flashcards(tmp_path, "python", "sess-004", entries) + # "wraps functions" is exactly 15 chars → included + assert count == 1 + + def test_dedup_skips_existing_cards(self, tmp_path): + # Pre-create a flashcard file with the same front + fc_dir = tmp_path / "python" / "flashcards" + fc_dir.mkdir(parents=True) + existing = { + "title": "Previous session", + "cards": [ + { + "front": "What is Python decorators?", + "back": "old note", + "source": "session-2026-01-01", + }, + ], + } + (fc_dir / "2026-01-01-python-flashcards.json").write_text(json.dumps(existing)) + + entries = [ + _win("Python decorators", "A decorator wraps a function to extend its behaviour"), + ] + count = write_session_flashcards(tmp_path, "python", "sess-005", entries) + assert count == 0 # deduped — card already exists + + def test_intra_session_dedup(self, tmp_path): + """Two entries with same topic in same session → only one card.""" + entries = [ + _win("Python decorators", "A decorator wraps a function to extend its behaviour"), + _win("Python decorators", "Decorators use @ syntax and wrap the function"), + ] + count = write_session_flashcards(tmp_path, "python", "sess-006", entries) + assert count == 1 + + def test_write_failure_propagates_to_caller(self, tmp_path, caplog): + """OSError from write_text propagates; cleanup.py wraps it with logger.warning.""" + import contextlib + + entries = [ + _win("Python decorators", "A decorator wraps a function to extend its behaviour"), + ] + with ( + patch.object(Path, "write_text", side_effect=OSError("disk full")), + contextlib.suppress(OSError), + ): + write_session_flashcards(tmp_path, "python", "sess-007", entries) + + def test_empty_entries_returns_zero(self, tmp_path): + count = write_session_flashcards(tmp_path, "python", "sess-008", []) + assert count == 0 + + def test_json_structure_has_required_fields(self, tmp_path): + entries = [ + _win("Context managers", "with statement ensures __exit__ is called on block exit"), + ] + write_session_flashcards(tmp_path, "python", "sess-009", entries) + + fc_dir = tmp_path / "python" / "flashcards" + files = list(fc_dir.glob("*flashcards.json")) + data = json.loads(files[0].read_text()) + + assert "title" in data + assert "cards" in data + card = data["cards"][0] + assert "front" in card + assert "back" in card + assert "source" in card + + +# --------------------------------------------------------------------------- +# _existing_card_hashes +# --------------------------------------------------------------------------- + + +class TestExistingCardHashes: + def test_returns_empty_set_when_dir_missing(self, tmp_path): + hashes = _existing_card_hashes(tmp_path / "nonexistent") + assert hashes == set() + + def test_collects_hashes_from_existing_files(self, tmp_path): + fc_dir = tmp_path / "flashcards" + fc_dir.mkdir() + payload = { + "title": "Test", + "cards": [{"front": "What is Python?", "back": "A language"}], + } + (fc_dir / "test-flashcards.json").write_text(json.dumps(payload)) + + hashes = _existing_card_hashes(fc_dir) + expected = _card_hash("What is Python?") + assert expected in hashes + + def test_skips_malformed_json_files(self, tmp_path, caplog): + fc_dir = tmp_path / "flashcards" + fc_dir.mkdir() + (fc_dir / "bad-flashcards.json").write_text("not valid json{{{") + + with caplog.at_level(logging.WARNING, logger="studyctl.services.flashcard_writer"): + hashes = _existing_card_hashes(fc_dir) + + assert hashes == set() + assert "malformed" in caplog.text.lower() or "Skipping" in caplog.text diff --git a/packages/studyctl/tests/test_topic_resolver.py b/packages/studyctl/tests/test_topic_resolver.py new file mode 100644 index 000000000..8858d4849 --- /dev/null +++ b/packages/studyctl/tests/test_topic_resolver.py @@ -0,0 +1,227 @@ +"""Tests for the topic resolver — pure function, no mocking needed.""" + +from __future__ import annotations + +from pathlib import Path + +from studyctl.logic.topic_resolver import MatchKind, ResolveResult, resolve_topic +from studyctl.settings import TopicConfig + + +def _topic(name: str, tags: list[str] | None = None) -> TopicConfig: + return TopicConfig( + name=name, + slug=name.lower().replace(" ", "-"), + obsidian_path=Path("/fake"), + tags=tags or [], + ) + + +TOPICS = [ + _topic("Python", tags=["python", "programming"]), + _topic("SQL", tags=["sql", "databases"]), + _topic("Data Engineering", tags=["data-engineering", "spark", "glue"]), +] + + +# --------------------------------------------------------------------------- +# Exact match +# --------------------------------------------------------------------------- + + +class TestExactMatch: + def test_exact_name(self): + result = resolve_topic("Python", TOPICS) + assert result.kind == MatchKind.EXACT + assert result.resolved is not None + assert result.resolved.name == "Python" + + def test_exact_name_casefold(self): + result = resolve_topic("python", TOPICS) + assert result.kind == MatchKind.EXACT + assert result.resolved.name == "Python" + + def test_exact_name_uppercase(self): + result = resolve_topic("SQL", TOPICS) + assert result.kind == MatchKind.EXACT + assert result.resolved.name == "SQL" + + def test_exact_multi_word(self): + result = resolve_topic("data engineering", TOPICS) + assert result.kind == MatchKind.EXACT + assert result.resolved.name == "Data Engineering" + + +# --------------------------------------------------------------------------- +# Name substring +# --------------------------------------------------------------------------- + + +class TestNameSubstring: + def test_query_contains_name(self): + """'Python Decorators' contains 'Python'.""" + result = resolve_topic("Python Decorators", TOPICS) + assert result.kind == MatchKind.NAME + assert result.resolved is not None + assert result.resolved.name == "Python" + + def test_name_contains_query(self): + """'Data' is contained in 'Data Engineering'.""" + result = resolve_topic("Data", TOPICS) + assert result.kind == MatchKind.NAME + assert result.resolved is not None + assert result.resolved.name == "Data Engineering" + + def test_multiple_substring_matches(self): + """Query matching multiple topics returns all candidates.""" + topics = [ + _topic("Python Basics"), + _topic("Python Advanced"), + _topic("SQL"), + ] + result = resolve_topic("Python", topics) + assert result.kind == MatchKind.NAME + assert result.resolved is None # ambiguous + assert len(result.matches) == 2 + + def test_substring_case_insensitive(self): + result = resolve_topic("python decorators", TOPICS) + assert result.kind == MatchKind.NAME + assert result.resolved.name == "Python" + + +# --------------------------------------------------------------------------- +# Tag match +# --------------------------------------------------------------------------- + + +class TestTagMatch: + def test_single_tag_match(self): + result = resolve_topic("Spark Joins", TOPICS) + assert result.kind == MatchKind.TAG + assert result.resolved is not None + assert result.resolved.name == "Data Engineering" + + def test_tag_match_partial(self): + """Tag 'databases' matches query containing 'databases'.""" + result = resolve_topic("databases fundamentals", TOPICS) + assert result.kind == MatchKind.TAG + assert result.resolved.name == "SQL" + + def test_tag_match_multiple(self): + """Query matching tags on multiple topics returns all.""" + topics = [ + _topic("Python", tags=["coding"]), + _topic("SQL", tags=["coding"]), + ] + result = resolve_topic("coding", topics) + assert result.kind == MatchKind.TAG + assert result.resolved is None + assert len(result.matches) == 2 + + +# --------------------------------------------------------------------------- +# Fuzzy match +# --------------------------------------------------------------------------- + + +class TestFuzzyMatch: + def test_typo_correction(self): + result = resolve_topic("Pyhton", TOPICS) + assert result.kind == MatchKind.FUZZY + assert any(t.name == "Python" for t in result.matches) + + def test_close_match(self): + result = resolve_topic("Sequel", TOPICS) + assert result.kind == MatchKind.FUZZY + assert any(t.name == "SQL" for t in result.matches) + + +# --------------------------------------------------------------------------- +# No match +# --------------------------------------------------------------------------- + + +class TestNoMatch: + def test_no_match_garbage(self): + result = resolve_topic("xyzzy plugh", TOPICS) + assert result.kind == MatchKind.NONE + assert result.matches == [] + assert result.resolved is None + + def test_empty_query(self): + result = resolve_topic("", TOPICS) + assert result.kind == MatchKind.NONE + + def test_whitespace_only(self): + result = resolve_topic(" ", TOPICS) + assert result.kind == MatchKind.NONE + + def test_empty_topics_list(self): + result = resolve_topic("Python", []) + assert result.kind == MatchKind.NONE + + +# --------------------------------------------------------------------------- +# Unicode / casefold +# --------------------------------------------------------------------------- + + +class TestUnicode: + def test_casefold_accent(self): + topics = [_topic("Café Networking")] + result = resolve_topic("café networking", topics) + assert result.kind == MatchKind.EXACT + assert result.resolved.name == "Café Networking" + + def test_casefold_german_eszett(self): + topics = [_topic("Straße")] + result = resolve_topic("strasse", topics) + assert result.kind == MatchKind.EXACT + + +# --------------------------------------------------------------------------- +# Precedence +# --------------------------------------------------------------------------- + + +class TestPrecedence: + def test_exact_beats_substring(self): + """If exact match exists, don't fall through to substring.""" + topics = [ + _topic("Python"), + _topic("Python Advanced"), + ] + result = resolve_topic("Python", topics) + assert result.kind == MatchKind.EXACT + assert result.resolved.name == "Python" + + def test_name_beats_tag(self): + """Substring match on name wins over tag match.""" + topics = [ + _topic("Spark", tags=["data"]), + _topic("Data Science", tags=["spark"]), + ] + result = resolve_topic("Spark Joins", topics) + assert result.kind == MatchKind.NAME + assert result.resolved.name == "Spark" + + +# --------------------------------------------------------------------------- +# ResolveResult properties +# --------------------------------------------------------------------------- + + +class TestResolveResult: + def test_resolved_single(self): + r = ResolveResult(MatchKind.EXACT, [_topic("Python")]) + assert r.resolved is not None + assert r.resolved.name == "Python" + + def test_resolved_none_on_multiple(self): + r = ResolveResult(MatchKind.NAME, [_topic("A"), _topic("B")]) + assert r.resolved is None + + def test_resolved_none_on_empty(self): + r = ResolveResult(MatchKind.NONE, []) + assert r.resolved is None