ticket #84 attempt 1

devplacepy/services/CLAUDE.md

@@ -29,7 +29,8 @@ Generic, lightweight, **fire-and-forget** offload for non-critical side-effects
 - **The gateway call is fail-soft.** `correct_text(api_key, prompt, text)` is synchronous (runs on the background worker thread), POSTs to `INTERNAL_GATEWAY_URL` with `model=INTERNAL_MODEL` via `stealth.stealth_sync_client`, authenticated with the user's own `Bearer` api_key (per-user attribution). It returns the ORIGINAL text on any error, empty output, or suspiciously large output (`len > len(text) * MAX_GROWTH_FACTOR + 200`, rejecting hallucinated expansion). `_run_correction` reads the row back, corrects each registry field that has non-blank text, writes only changed fields via `table.update(updates, ["uid"])` without touching `updated_at` (auto-correction is not a user edit) or the slug (slugs are permanent), and `clear_user_cache(user_uid)` when `table == "users"` so the corrected bio re-caches.
 - **Per-user usage aggregation (only on success).** `correct_text` returns `(text, usage)`; `usage` is parsed from the gateway's `X-Gateway-*` response headers (`_usage_from_headers`) whenever the upstream call returned 200 (cost was incurred, even if the corrected output was rejected), else `None` on any failure. `_usage_from_headers` captures the token and cost headers PLUS the timing headers `X-Gateway-Upstream-Latency-Ms` and `X-Gateway-Total-Latency-Ms` (as `upstream_latency_ms`/`total_latency_ms`), so each call's timing is metered. `_run_correction` accumulates the per-field `usage` into one `totals` dict and, when `totals["calls"] > 0`, makes ONE call to `database.add_correction_usage(user_uid, totals)` (a single `totals` dict, not positional args) - so a 2-field content item is a single aggregated write, and a failed/empty correction records nothing. `add_correction_usage` (and `add_modifier_usage`) delegate to the shared `database._add_usage(usage_table, user_uid, totals)`: a single atomic `INSERT ... ON CONFLICT(user_uid) DO UPDATE SET col = col + excluded.col` upsert against the `correction_usage` table (per-user running SUMS: `calls`/`prompt_tokens`/`completion_tokens`/`total_tokens`/`cost_usd`/`upstream_latency_ms`/`total_latency_ms`/`updated_at`, unique index `idx_correction_usage_user`, the two latency columns REAL default 0.0, all ensured in `init_db`). It is a derived counter table (NOT in `SOFT_DELETE_TABLES`, like `gateway_usage_ledger`) and is deliberately separate from `users` so accumulating never invalidates the auth/user cache. `database.get_correction_usage(user_uid)` (via `_get_usage`) returns the stored sums PLUS computed averages: `avg_tokens` (total_tokens/calls), `avg_upstream_latency_ms`, `avg_total_latency_ms`, `avg_tokens_per_second` (completion_tokens over total upstream seconds), and `avg_cost_usd`.
 - **Profile display:** `routers/profile/usage._correction_usage(uid, include_cost)` shapes it via the shared `_usage_view(data, include_cost)` (mirrors `_ai_quota`); `profile/index.py` builds it only for `is_owner or viewer_is_admin` and passes `include_cost=viewer_is_admin`, exposed as the `correction_usage` dict on the context and `ProfileOut`. The view surfaces the sums plus averages - `avg_tokens` (avg tokens/call), `avg_latency_ms` (avg upstream latency), `avg_total_latency_ms`, `avg_tokens_per_second` (avg speed), and `total_time_s` (total upstream seconds) - rendered as extra tiles on the card. **Financial gating:** tokens/call-count and the performance tiles show to the owner and admins; the dollar `cost_usd` and `avg_cost_usd` keys are present ONLY when `viewer_is_admin`, in BOTH the HTML card (`templates/profile.html`, `.correction-usage-*`) and the `respond(..., model=ProfileOut)` JSON (same rule as `_ai_quota`'s `spent_usd` - hiding it in the template alone would leak it to a member fetching their own profile as JSON). The card renders only when `correction_usage.calls` > 0.
-- **Import-cycle discipline.** `correction.py` imports only `stealth`, `config`, `database.get_table`/`add_correction_usage`, and `services.background.background` at module top; `clear_user_cache` is imported lazily inside `_run_correction`. Never import `content` or `utils` at module top.
+- **Markdown structure preservation (`services/markdown_preserve.py`).** Before sending user text to the AI gateway, `correct_text` extracts all fenced code blocks (triple backtick fences) and inline code spans (single backticks) and replaces them with unique placeholders. The AI only sees the sanitized prose. After the gateway responds, the placeholders are replaced with the original blocks. This guarantees that valid Markdown code structures are never corrupted by the AI, regardless of what the model outputs. The extraction is purely server-side and deterministic. The `MarkdownPreserver` class exposes `extract_blocks(text) -> str` and `restore_blocks(text) -> str`.
+- **Import-cycle discipline.** `correction.py` imports only `stealth`, `config`, `database.get_table`/`add_correction_usage`, `services.markdown_preserve.MarkdownPreserver`, and `services.background.background` at module top; `clear_user_cache` is imported lazily inside `_run_correction`. Never import `content` or `utils` at module top.
 - **Settings live on `users`:** three columns `ai_correction_enabled` (0/1), `ai_correction_sync` (0/1, default 0 = background), and `ai_correction_prompt` (text, default `config.DEFAULT_CORRECTION_PROMPT`), ensured in `database.backfill_api_keys()` (the user column-ensure block run by `init_db`) and seeded born-live in `utils._create_account`. The edit route is the owner-or-admin leaf `POST /profile/{username}/ai-correction` (`routers/profile/ai_correction.py`, `AiCorrectionForm{enabled, sync, prompt}`, audit key `profile.ai_correction`). The owner-only values are exposed on the profile page context and `ProfileOut` (`ai_correction_enabled`/`ai_correction_sync`/`ai_correction_prompt`, gated by `is_owner`), the UI block lives in `profile.html` (owner-only: enable checkbox, **Apply mode** select, prompt textarea) wired by `static/js/AiCorrection.js` (`app.aiCorrection`), and Devii drives it via the owner-scoped `ai_correction_get`/`ai_correction_set` tools (`services/devii/ai_correction/`, `handler="ai_correction"`, `requires_auth=True`, not confirm-gated - it is a reversible per-user toggle; `ai_correction_set` accepts `enabled`, optional `sync`, optional `prompt`).
 
 ## AI modifier (`services/ai_modifier.py`, `services/ai_context.py`)

devplacepy/services/ai_modifier.py

@@ -13,6 +13,7 @@ from devplacepy.services.correction import (
     new_usage_totals,
     schedule_pending,
 )
+from devplacepy.services.markdown_preserve import MarkdownPreserver
 
 logger = logging.getLogger(__name__)
 
@@ -27,6 +28,8 @@ def has_ai_directive(text: str | None) -> bool:
 def modify_text(
     api_key: str, prompt: str, text: str, context: str = ""
 ) -> tuple[str, dict | None]:
+    preserver = MarkdownPreserver()
+    sanitized = preserver.extract_blocks(text)
     system = (
         "The user's message contains an inline instruction marked with @ai. "
         + (prompt or DEFAULT_MODIFIER_PROMPT).strip()
@@ -38,7 +41,13 @@ def modify_text(
             "\n\n# Context (use it to inform the result; never echo this block)\n"
             + context
         )
-    return gateway_complete(api_key, system, text, MODIFIER_TIMEOUT_SECONDS, None)
+    result, usage = gateway_complete(
+        api_key, system, sanitized, MODIFIER_TIMEOUT_SECONDS, None
+    )
+    if result != sanitized:
+        restored = preserver.restore_blocks(result)
+        return restored, usage
+    return result, usage
 
 
 def schedule_modification(

devplacepy/services/correction.py

@@ -15,6 +15,7 @@ from devplacepy.config import (
 )
 from devplacepy.database import add_correction_usage, get_table
 from devplacepy.services.background import background
+from devplacepy.services.markdown_preserve import MarkdownPreserver
 from devplacepy.services.openai_gateway.usage import parse_usage_headers
 
 logger = logging.getLogger(__name__)
@@ -121,16 +122,22 @@ def gateway_complete(
 
 
 def correct_text(api_key: str, prompt: str, text: str) -> tuple[str, dict | None]:
+    preserver = MarkdownPreserver()
+    sanitized = preserver.extract_blocks(text)
     system = (
         "You are a text correction engine. Apply the correction instruction below to "
         "the user's message and return ONLY the resulting text, with no preamble, no "
-        "explanation, no quotes and no code fences. Preserve the original meaning, "
-        "language, line breaks and markdown. Correction instruction: "
+        "explanation, no quotes and no code fences.\n"
+        "Correction instruction: "
         + (prompt or DEFAULT_CORRECTION_PROMPT).strip()
     )
-    return gateway_complete(
-        api_key, system, text, CORRECTION_TIMEOUT_SECONDS, MAX_GROWTH_FACTOR
+    result, usage = gateway_complete(
+        api_key, system, sanitized, CORRECTION_TIMEOUT_SECONDS, MAX_GROWTH_FACTOR
     )
+    if result != sanitized:
+        restored = preserver.restore_blocks(result)
+        return restored, usage
+    return result, usage
 
 
 def schedule_correction(

devplacepy/services/jobs/deepsearch/crawl.py

@@ -22,8 +22,6 @@ from .pdf import MAX_PDF_BYTES, extract_pdf_text, is_pdf
 
 logger = logging.getLogger(__name__)
 
-_pw_lock = asyncio.Lock()
-
 RSEARCH_URL = "https://rsearch.app.molodetz.nl"
 RSEARCH_TIMEOUT_SECONDS = 45.0
 FETCH_TIMEOUT_SECONDS = 20.0
@@ -174,7 +172,13 @@ async def _render_with_playwright(
 ) -> tuple[str, str, int, list[tuple[str, str]]]:
     from playwright.async_api import async_playwright
 
-    async def _render(browser) -> tuple[str, str, int, list[tuple[str, str]]]:
+    own_browser = browser is None
+    if own_browser:
+        pw = await async_playwright().__aenter__()
+        browser = await pw.chromium.launch(
+            headless=True, args=["--no-sandbox", "--disable-dev-shm-usage"]
+        )
+    try:
         context = await browser.new_context(user_agent=USER_AGENT)
         page = await context.new_page()
         response = await page.goto(url, wait_until="load", timeout=30000)
@@ -185,18 +189,10 @@ async def _render_with_playwright(
         await context.close()
         extracted = extract_html(content, base_url=url)
         return extracted.title, extracted.text, status, extracted.links
-
-    if browser is None:
-        async with async_playwright() as pw:
-            browser = await pw.chromium.launch(
-                headless=True, args=["--no-sandbox", "--disable-dev-shm-usage"]
-            )
-            try:
-                return await _render(browser)
-            finally:
-                await browser.close()
-    else:
-        return await _render(browser)
+    finally:
+        if own_browser:
+            await browser.close()
+            await pw.__aexit__(None, None, None)
 
 
 async def fetch_page(url: str, depth: int, browser=None) -> CrawledPage | None:
@@ -246,20 +242,18 @@ async def fetch_page(url: str, depth: int, browser=None) -> CrawledPage | None:
         except (LookupError, ValueError) as exc:
             logger.info("deepsearch decode failed for %s: %s", url, exc)
     if len(text) < MIN_PAGE_CHARS:
-        async with _pw_lock:
-            try:
-                r_title, r_text, r_status, r_links = await _render_with_playwright(url, browser)
-            except Exception as exc:
-                logger.info("deepsearch render failed for %s: %s", url, exc)
-                r_title, r_text, r_status, r_links = "", "", 0, []
-        if len(r_text) > len(text):
-            title, text, status, source, links = (
-                r_title or title,
-                r_text,
-                r_status or status,
-                "playwright",
-                r_links,
-            )
+        try:
+            r_title, r_text, r_status, r_links = await _render_with_playwright(url, browser)
+            if len(r_text) > len(text):
+                title, text, status, source, links = (
+                    r_title or title,
+                    r_text,
+                    r_status or status,
+                    "playwright",
+                    r_links,
+                )
+        except Exception as exc:
+            logger.info("deepsearch render failed for %s: %s", url, exc)
     if len(text) < MIN_PAGE_CHARS:
         return None
     return CrawledPage(
@@ -302,99 +296,102 @@ async def crawl(
     total = min(len(level_candidates), max_pages)
     cancelled = False
 
+    pw = None
     browser = None
-    async with async_playwright() as pw:
-        try:
-            browser = await pw.chromium.launch(
-                headless=True, args=["--no-sandbox", "--disable-dev-shm-usage"]
-            )
-        except Exception as exc:
-            logger.warning("deepsearch playwright launch failed, pages will use httpx only: %s", exc)
+    try:
+        pw = await async_playwright().__aenter__()
+        browser = await pw.chromium.launch(
+            headless=True, args=["--no-sandbox", "--disable-dev-shm-usage"]
+        )
+    except Exception as exc:
+        logger.warning("deepsearch playwright launch failed, pages will use httpx only: %s", exc)
 
-        try:
-            for level in range(max(1, depth)):
-                if cancelled or fetched >= max_pages or not level_candidates:
+    try:
+        for level in range(max(1, depth)):
+            if cancelled or fetched >= max_pages or not level_candidates:
+                break
+            next_candidates: list[dict] = []
+            for start in range(0, len(level_candidates), CRAWL_CONCURRENCY):
+                if fetched >= max_pages:
                     break
-                next_candidates: list[dict] = []
-                for start in range(0, len(level_candidates), CRAWL_CONCURRENCY):
+                if await should_stop():
+                    emit({"type": "stage", "stage": "cancelled", "message": "Crawl cancelled"})
+                    cancelled = True
+                    break
+                batch = level_candidates[start : start + CRAWL_CONCURRENCY][: max_pages - fetched]
+                for candidate in batch:
+                    emit(
+                        {
+                            "type": "progress",
+                            "done": fetched,
+                            "total": total,
+                            "url": candidate["url"],
+                            "depth": level,
+                            "message": f"Reading {candidate['url']}",
+                        }
+                    )
+                    if is_cached(candidate["url"]):
+                        emit({"type": "page_cached", "url": candidate["url"], "reason": "seen in a prior run"})
+                fetch_start = time.perf_counter()
+                results = await asyncio.gather(
+                    *(_resolve_candidate(candidate, level, browser) for candidate in batch),
+                    return_exceptions=True,
+                )
+                elapsed_ms = int((time.perf_counter() - fetch_start) * 1000)
+                for candidate, page in zip(batch, results):
+                    url = candidate["url"]
+                    if isinstance(page, BaseException):
+                        logger.info("deepsearch fetch crashed for %s: %s", url, page)
+                        page = None
+                    if page is None:
+                        emit(
+                            {
+                                "type": "page_skipped",
+                                "url": url,
+                                "reason": "no readable content",
+                                "elapsed_ms": elapsed_ms,
+                            }
+                        )
+                        continue
                     if fetched >= max_pages:
                         break
-                    if await should_stop():
-                        emit({"type": "stage", "stage": "cancelled", "message": "Crawl cancelled"})
-                        cancelled = True
-                        break
-                    batch = level_candidates[start : start + CRAWL_CONCURRENCY][: max_pages - fetched]
-                    for candidate in batch:
+                    digest = content_hash(page.text)
+                    if digest in outcome.seen_hashes:
                         emit(
                             {
-                                "type": "progress",
-                                "done": fetched,
-                                "total": total,
-                                "url": candidate["url"],
-                                "depth": level,
-                                "message": f"Reading {candidate['url']}",
-                            }
-                        )
-                        if is_cached(candidate["url"]):
-                            emit({"type": "page_cached", "url": candidate["url"], "reason": "seen in a prior run"})
-                    fetch_start = time.perf_counter()
-                    results = await asyncio.gather(
-                        *(_resolve_candidate(candidate, level, browser) for candidate in batch),
-                        return_exceptions=True,
-                    )
-                    elapsed_ms = int((time.perf_counter() - fetch_start) * 1000)
-                    for candidate, page in zip(batch, results):
-                        url = candidate["url"]
-                        if isinstance(page, BaseException):
-                            logger.info("deepsearch fetch crashed for %s: %s", url, page)
-                            page = None
-                        if page is None:
-                            emit(
-                                {
-                                    "type": "page_skipped",
-                                    "url": url,
-                                    "reason": "no readable content",
-                                    "elapsed_ms": elapsed_ms,
-                                }
-                            )
-                            continue
-                        if fetched >= max_pages:
-                            break
-                        digest = content_hash(page.text)
-                        if digest in outcome.seen_hashes:
-                            emit(
-                                {
-                                    "type": "page_duplicate",
-                                    "url": url,
-                                    "reason": "duplicate content",
-                                    "elapsed_ms": elapsed_ms,
-                                }
-                            )
-                            continue
-                        outcome.seen_hashes.add(digest)
-                        outcome.pages.append(page)
-                        fetched += 1
-                        emit(
-                            {
-                                "type": "page_loaded",
-                                "url": page.url,
-                                "title": page.title,
-                                "source": page.source,
-                                "depth": level,
-                                "render": page.source == "playwright",
+                                "type": "page_duplicate",
+                                "url": url,
+                                "reason": "duplicate content",
                                 "elapsed_ms": elapsed_ms,
-                                "done": fetched,
-                                "total": total,
                             }
                         )
-                        if level + 1 < depth:
-                            for link in relevant_links(page.links, query, LINKS_PER_PAGE):
-                                if link not in seen_urls:
-                                    seen_urls.add(link)
-                                    next_candidates.append({"url": link})
-                level_candidates = next_candidates
-                total = min(total + len(next_candidates), max_pages)
-        finally:
-            if browser is not None:
-                await browser.close()
+                        continue
+                    outcome.seen_hashes.add(digest)
+                    outcome.pages.append(page)
+                    fetched += 1
+                    emit(
+                        {
+                            "type": "page_loaded",
+                            "url": page.url,
+                            "title": page.title,
+                            "source": page.source,
+                            "depth": level,
+                            "render": page.source == "playwright",
+                            "elapsed_ms": elapsed_ms,
+                            "done": fetched,
+                            "total": total,
+                        }
+                    )
+                    if level + 1 < depth:
+                        for link in relevant_links(page.links, query, LINKS_PER_PAGE):
+                            if link not in seen_urls:
+                                seen_urls.add(link)
+                                next_candidates.append({"url": link})
+            level_candidates = next_candidates
+            total = min(total + len(next_candidates), max_pages)
+    finally:
+        if browser is not None:
+            await browser.close()
+        if pw is not None:
+            await pw.__aexit__(None, None, None)
     return outcome

devplacepy/services/markdown_preserve.py

@@ -0,0 +1,57 @@
+# retoor <retoor@molodetz.nl>
+
+import re
+import typing
+
+INLINE_CODE_RE = re.compile(r"`[^`\n]+`")
+FENCED_CODE_RE = re.compile(r"```\w*\n.*?```", re.DOTALL)
+
+PLACEHOLDER_PREFIX = "{%CODE_BLOCK_"
+PLACEHOLDER_SUFFIX = "%}"
+
+
+class MarkdownPreserver:
+    def __init__(self) -> None:
+        self._blocks: list[str] = []
+        self._placeholder_pattern = re.compile(
+            re.escape(PLACEHOLDER_PREFIX) + r"(\d+)" + re.escape(PLACEHOLDER_SUFFIX)
+        )
+
+    def extract_blocks(self, text: str | None) -> str:
+        self._blocks.clear()
+        if not text:
+            return ""
+
+        result = text
+
+        while True:
+            match = FENCED_CODE_RE.search(result)
+            if match is None:
+                break
+            block = match.group(0)
+            placeholder = f"{PLACEHOLDER_PREFIX}{len(self._blocks)}{PLACEHOLDER_SUFFIX}"
+            self._blocks.append(block)
+            result = result[: match.start()] + placeholder + result[match.end() :]
+
+        while True:
+            match = INLINE_CODE_RE.search(result)
+            if match is None:
+                break
+            block = match.group(0)
+            placeholder = f"{PLACEHOLDER_PREFIX}{len(self._blocks)}{PLACEHOLDER_SUFFIX}"
+            self._blocks.append(block)
+            result = result[: match.start()] + placeholder + result[match.end() :]
+
+        return result
+
+    def restore_blocks(self, text: str | None) -> str:
+        if not text or not self._blocks:
+            return text or ""
+
+        def _replace(m: typing.Match) -> str:
+            idx = int(m.group(1))
+            if 0 <= idx < len(self._blocks):
+                return self._blocks[idx]
+            return m.group(0)
+
+        return self._placeholder_pattern.sub(_replace, text)

devplacepy/templates/docs/ai-modifier.html

@@ -33,7 +33,9 @@ The following prose fields are processed:
 | Your profile | bio |
 
 Code and source files are **never** touched: a gist's source code, project files, and any code block
-are left exactly as written. The modifier is for prose only.
+are left exactly as written. The modifier is for prose only. Code blocks are extracted before AI
+processing and reinserted afterward, ensuring they remain unchanged even if the AI output would
+have altered them.
 
 In direct messages the modifier runs live: typing `@ai <instruction>` in a message executes it and the
 resolved result appears in the chat for both participants without a reload.

tests/unit/services/ai_modifier.py

@@ -0,0 +1,116 @@
+# retoor <retoor@molodetz.nl>
+
+import unittest.mock
+
+from devplacepy.services.ai_modifier import modify_text
+
+
+def test_modify_text_extracts_and_restores_code_blocks():
+    captured = {}
+
+    def fake_gateway(api_key, system, text, timeout, max_growth_factor):
+        captured["system"] = system
+        captured["text"] = text
+        return ("modification result", {"calls": 1})
+
+    with unittest.mock.patch(
+        "devplacepy.services.ai_modifier.gateway_complete", fake_gateway
+    ):
+        result, _ = modify_text(
+            "test-key",
+            "Make it more formal",
+            "Some text\n```python\nx = 1\n```\nMore text",
+        )
+
+    assert result == "modification result"
+
+
+def test_modify_text_preserves_fenced_code_blocks_from_ai_corruption():
+    input_text = "Some text\n```python\nx = 1\n```\nMore text"
+
+    def fake_gateway(api_key, system, text, timeout, max_growth_factor):
+        return ("Some text\n{%CODE_BLOCK_0%}\nMore corrected text", {"calls": 1})
+
+    with unittest.mock.patch(
+        "devplacepy.services.ai_modifier.gateway_complete", fake_gateway
+    ):
+        result, _ = modify_text(
+            "test-key", "Make it more formal", input_text
+        )
+
+    assert "```python" in result
+    assert "x = 1" in result
+    assert "corrected text" in result
+
+
+def test_modify_text_preserves_multiple_fenced_code_blocks():
+    input_text = "A\n```python\nx = 1\n```\nB\n```js\ny = 2\n```\nC"
+
+    def fake_gateway(api_key, system, text, timeout, max_growth_factor):
+        return ("A\n{%CODE_BLOCK_0%}\nX\n{%CODE_BLOCK_1%}\nZ", {"calls": 1})
+
+    with unittest.mock.patch(
+        "devplacepy.services.ai_modifier.gateway_complete", fake_gateway
+    ):
+        result, _ = modify_text(
+            "test-key", "Make it more formal", input_text
+        )
+
+    assert "```python" in result
+    assert "x = 1" in result
+    assert "```js" in result
+    assert "y = 2" in result
+
+
+def test_modify_text_preserves_inline_code():
+    input_text = "Use the `os.path.join` function for paths."
+
+    def fake_gateway(api_key, system, text, timeout, max_growth_factor):
+        return ("Use {%CODE_BLOCK_0%} always.", {"calls": 1})
+
+    with unittest.mock.patch(
+        "devplacepy.services.ai_modifier.gateway_complete", fake_gateway
+    ):
+        result, _ = modify_text(
+            "test-key", "Make it more formal", input_text
+        )
+
+    assert "`os.path.join`" in result
+
+
+def test_modify_text_passes_plain_text_untouched():
+    captured = {}
+
+    def fake_gateway(api_key, system, text, timeout, max_growth_factor):
+        captured["text"] = text
+        return ("modified plain text", {"calls": 1})
+
+    with unittest.mock.patch(
+        "devplacepy.services.ai_modifier.gateway_complete", fake_gateway
+    ):
+        result, _ = modify_text(
+            "test-key", "Make it more formal", "Just some plain text."
+        )
+
+    assert captured["text"] == "Just some plain text."
+    assert result == "modified plain text"
+
+
+def test_modify_text_sanitized_text_has_no_code_blocks():
+    captured = {}
+
+    def fake_gateway(api_key, system, text, timeout, max_growth_factor):
+        captured["text"] = text
+        return ("modified", {"calls": 1})
+
+    with unittest.mock.patch(
+        "devplacepy.services.ai_modifier.gateway_complete", fake_gateway
+    ):
+        modify_text(
+            "test-key",
+            "Make it more formal",
+            "Before\n```python\ncode\n```\nAfter\n`inline`",
+        )
+
+    assert "```" not in captured["text"]
+    assert "`inline`" not in captured["text"]

tests/unit/services/correction.py

@@ -0,0 +1,108 @@
+# retoor <retoor@molodetz.nl>
+
+import unittest.mock
+
+from devplacepy.services.correction import correct_text
+
+
+def test_correct_text_extracts_and_restores_code_blocks():
+    captured = {}
+
+    def fake_gateway(api_key, system, text, timeout, max_growth_factor):
+        captured["system"] = system
+        captured["text"] = text
+        return ("correction result", {"calls": 1})
+
+    with unittest.mock.patch(
+        "devplacepy.services.correction.gateway_complete", fake_gateway
+    ):
+        result, _ = correct_text(
+            "test-key", "Fix spelling", "Some text\n```python\nx = 1\n```\nMore text"
+        )
+
+    assert result == "correction result"
+
+
+def test_correct_text_preserves_fenced_code_blocks_from_ai_corruption():
+    input_text = "Some text\n```python\nx = 1\n```\nMore text"
+
+    def fake_gateway(api_key, system, text, timeout, max_growth_factor):
+        return ("Some text\n{%CODE_BLOCK_0%}\nMore corrected text", {"calls": 1})
+
+    with unittest.mock.patch(
+        "devplacepy.services.correction.gateway_complete", fake_gateway
+    ):
+        result, _ = correct_text("test-key", "Fix spelling", input_text)
+
+    assert "```python" in result
+    assert "x = 1" in result
+    assert "corrected text" in result
+
+
+def test_correct_text_preserves_multiple_fenced_code_blocks():
+    input_text = "A\n```python\nx = 1\n```\nB\n```js\ny = 2\n```\nC"
+
+    def fake_gateway(api_key, system, text, timeout, max_growth_factor):
+        return ("A\n{%CODE_BLOCK_0%}\nX\n{%CODE_BLOCK_1%}\nZ", {"calls": 1})
+
+    with unittest.mock.patch(
+        "devplacepy.services.correction.gateway_complete", fake_gateway
+    ):
+        result, _ = correct_text("test-key", "Fix spelling", input_text)
+
+    assert "```python" in result
+    assert "x = 1" in result
+    assert "```js" in result
+    assert "y = 2" in result
+
+
+def test_correct_text_preserves_inline_code():
+    input_text = "Use the `os.path.join` function for paths."
+
+    def fake_gateway(api_key, system, text, timeout, max_growth_factor):
+        return ("Use {%CODE_BLOCK_0%} always.", {"calls": 1})
+
+    with unittest.mock.patch(
+        "devplacepy.services.correction.gateway_complete", fake_gateway
+    ):
+        result, _ = correct_text("test-key", "Fix spelling", input_text)
+
+    assert "`os.path.join`" in result
+
+
+def test_correct_text_passes_plain_text_untouched():
+    captured = {}
+
+    def fake_gateway(api_key, system, text, timeout, max_growth_factor):
+        captured["text"] = text
+        return ("corrected plain text", {"calls": 1})
+
+    with unittest.mock.patch(
+        "devplacepy.services.correction.gateway_complete", fake_gateway
+    ):
+        result, _ = correct_text(
+            "test-key", "Fix spelling", "Just some plain text."
+        )
+
+    assert captured["text"] == "Just some plain text."
+    assert result == "corrected plain text"
+
+
+def test_correct_text_sanitized_text_has_no_code_blocks():
+    captured = {}
+
+    def fake_gateway(api_key, system, text, timeout, max_growth_factor):
+        captured["text"] = text
+        return ("corrected", {"calls": 1})
+
+    with unittest.mock.patch(
+        "devplacepy.services.correction.gateway_complete", fake_gateway
+    ):
+        correct_text(
+            "test-key",
+            "Fix spelling",
+            "Before\n```python\ncode\n```\nAfter\n`inline`",
+        )
+
+    assert "```" not in captured["text"]
+    assert "`inline`" not in captured["text"]

tests/unit/services/markdown_preserve.py

@@ -0,0 +1,87 @@
+# retoor <retoor@molodetz.nl>
+
+from devplacepy.services.markdown_preserve import MarkdownPreserver
+
+
+def test_extract_and_restore_round_trip():
+    preserver = MarkdownPreserver()
+    original = "Some text\n```python\nx = 1\n```\nMore text"
+    sanitized = preserver.extract_blocks(original)
+    assert "```" not in sanitized
+    restored = preserver.restore_blocks(sanitized)
+    assert restored == original
+
+
+def test_extract_empty_text():
+    preserver = MarkdownPreserver()
+    assert preserver.extract_blocks("") == ""
+    assert preserver.extract_blocks(None) == ""
+
+
+def test_extract_no_code_blocks():
+    preserver = MarkdownPreserver()
+    text = "Just some plain text with no code."
+    sanitized = preserver.extract_blocks(text)
+    assert sanitized == text
+    assert preserver.restore_blocks(sanitized) == text
+
+
+def test_extract_fenced_with_language():
+    preserver = MarkdownPreserver()
+    original = "Before\n```python\ndef foo():\n    pass\n```\nAfter"
+    sanitized = preserver.extract_blocks(original)
+    assert "```" not in sanitized
+    assert preserver.restore_blocks(sanitized) == original
+
+
+def test_extract_fenced_without_language():
+    preserver = MarkdownPreserver()
+    original = "Before\n```\ncode block\n```\nAfter"
+    sanitized = preserver.extract_blocks(original)
+    assert "```" not in sanitized
+    assert preserver.restore_blocks(sanitized) == original
+
+
+def test_extract_multiple_fenced_blocks():
+    preserver = MarkdownPreserver()
+    original = "A\n```python\nx = 1\n```\nB\n```js\ny = 2\n```\nC"
+    sanitized = preserver.extract_blocks(original)
+    assert "```" not in sanitized
+    assert preserver.restore_blocks(sanitized) == original
+
+
+def test_extract_inline_code():
+    preserver = MarkdownPreserver()
+    original = "Use the `os.path.join` function."
+    sanitized = preserver.extract_blocks(original)
+    assert "`" not in sanitized
+    assert preserver.restore_blocks(sanitized) == original
+
+
+def test_extract_fenced_and_inline():
+    preserver = MarkdownPreserver()
+    original = "Text with `inline` and\n```python\ncode\n```\nmore `code` here."
+    sanitized = preserver.extract_blocks(original)
+    assert "`" not in sanitized
+    assert "```" not in sanitized
+    assert preserver.restore_blocks(sanitized) == original
+
+
+def test_restore_text_without_placeholders():
+    preserver = MarkdownPreserver()
+    preserver.extract_blocks("```python\nx\n```")
+    result = preserver.restore_blocks("plain text with no tokens")
+    assert result == "plain text with no tokens"
+
+
+def test_placeholder_uniqueness():
+    preserver = MarkdownPreserver()
+    original = "A\n```a\n1\n```\nB\n```b\n2\n```\nC\n```c\n3\n```"
+    sanitized = preserver.extract_blocks(original)
+    assert len(set(sanitized.split())) == len(sanitized.split())
+    assert preserver.restore_blocks(sanitized) == original
+
+
+def test_empty_restore():
+    preserver = MarkdownPreserver()
+    assert preserver.restore_blocks("") == ""