feat(api): add completion request cancellation and mermaid rendering

Add support for cancelling in-progress LLM completion requests via new /v1/completions/cancel endpoint with task tracking. Implement mermaid diagram rendering in the Milkdown editor with a new mermaidPlugin. Update copilotPlugin to properly abort requests with descriptive reasons. Refactor settings panel to handle system theme changes reactively. Add camera capture support for image uploads.

backend/llm.py

@@ -1,6 +1,7 @@
 import os
 import time
 import logging
+import asyncio
 from datetime import datetime
 import ollama
 from dotenv import load_dotenv
@@ -97,6 +98,17 @@ async def call_ollama(
             kwargs["think"] = thinking
 
         response = await client.chat(**kwargs)
+    except asyncio.CancelledError:
+        elapsed_ms = (time.perf_counter() - start) * 1000
+        end_dt = datetime.now()
+        logger.info(
+            "[LLM][%s] call_time [%s --> %s]",
+            tag,
+            start_dt.strftime("%H:%M:%S"),
+            end_dt.strftime("%H:%M:%S"),
+        )
+        logger.warning("[LLM][%s] request cancelled after %.1fms", tag, elapsed_ms)
+        raise
     except Exception:
         elapsed_ms = (time.perf_counter() - start) * 1000
         end_dt = datetime.now()

backend/main.py

@@ -1,16 +1,19 @@
-from fastapi import FastAPI, Request, HTTPException, Security
-from fastapi.security import APIKeyHeader
-from fastapi.middleware.cors import CORSMiddleware
-from fastapi.responses import StreamingResponse, JSONResponse
-from pydantic import BaseModel
-import json
+import asyncio
 import base64
-import uuid
+import json
 import logging
+import uuid
+from typing import Optional
+
+from fastapi import FastAPI, HTTPException, Request, Security
+from fastapi.middleware.cors import CORSMiddleware
+from fastapi.responses import JSONResponse, StreamingResponse
+from fastapi.security import APIKeyHeader
+from pydantic import BaseModel
 
-from prompt import build_completion_prompts, prepare_prompt_context
-from llm import call_ollama, call_vlm_ocr
 from geoip import get_ip_location_text
+from llm import call_ollama, call_vlm_ocr
+from prompt import build_completion_prompts, prepare_prompt_context
 
 logging.basicConfig(
     level=logging.INFO,
@@ -20,44 +23,54 @@ logger = logging.getLogger("api")
 
 app = FastAPI()
 
+ACTIVE_COMPLETIONS: dict[str, asyncio.Task] = {}
+ACTIVE_COMPLETIONS_LOCK = asyncio.Lock()
+
 app.add_middleware(
     CORSMiddleware,
     allow_origins=["*"],
     allow_credentials=True,
     allow_methods=["*"],
-    allow_headers=["*", "X-API-Key", "X-Client-IP"],
+    allow_headers=["*", "X-API-Key", "X-Client-IP", "X-Request-Id"],
 )
 
-API_KEY = "your-secret-key-here"  # 建议从环境变量读取
+API_KEY = "your-secret-key-here"
 api_key_header = APIKeyHeader(name="X-API-Key")
 
+
 async def get_api_key(api_key: str = Security(api_key_header)):
     if api_key != API_KEY:
         raise HTTPException(
             status_code=403,
-            detail="Could not validate credentials"
+            detail="Could not validate credentials",
         )
     return api_key
 
-from typing import Optional
 
 class UserPreferences(BaseModel):
-    language: str = 'auto'
-    currency: str = 'auto'
-    timezone: str = 'auto'
+    language: str = "auto"
+    currency: str = "auto"
+    timezone: str = "auto"
+
 
 class CompletionRequest(BaseModel):
     prefix: str
     suffix: str
-    languageId: str = 'markdown'
-    model_thinking: str = 'low'
+    languageId: str = "markdown"
+    model_thinking: str = "low"
     privacy_mode: bool = False
     user_preferences: Optional[UserPreferences] = None
 
+
+class CancelCompletionRequest(BaseModel):
+    request_id: str
+    reason: str = "abort"
+
+
 class OCRRequest(BaseModel):
     image: str
     filename: str = "image.jpg"
-    language: str = 'auto'
+    language: str = "auto"
 
 
 def _preview(text: str, limit: int = 80) -> str:
@@ -66,73 +79,143 @@ def _preview(text: str, limit: int = 80) -> str:
         return value
     return value[:limit] + "..."
 
+
+def _sse_payload(payload: dict) -> str:
+    return f"data: {json.dumps(payload)}\n\n"
+
+
 def get_client_ip(request: Request) -> str:
-    return request.headers.get("X-Client-IP") or request.client.host if request.client else "unknown"
+    if request.client:
+        return request.headers.get("X-Client-IP") or request.client.host
+    return request.headers.get("X-Client-IP") or "unknown"
+
 
 @app.post("/v1/completions")
 async def create_completion(request: Request, req: CompletionRequest, api_key: str = Security(get_api_key)):
-    request_id = str(uuid.uuid4())[:8]
-    
+    request_id = request.headers.get("X-Request-Id") or str(uuid.uuid4())
+    request_tag = request_id[:8]
+    inference_task: Optional[asyncio.Task] = None
+
     client_ip = "hidden"
     location = ""
-    
+
     if not req.privacy_mode:
         client_ip = get_client_ip(request)
-        # 查询 IP 归属地
         location = get_ip_location_text(client_ip)
         if location:
-            logger.info("[%s] client_location=%s", request_id, location)
-    
+            logger.info("[%s] client_location=%s", request_tag, location)
+
     try:
         logger.info(
-            "[%s] /v1/completions client_ip=%s prefix_chars=%d suffix_chars=%d lang=%s thinking=%s privacy=%s",
+            "[%s] /v1/completions request_id=%s client_ip=%s prefix_chars=%d suffix_chars=%d lang=%s thinking=%s privacy=%s",
+            request_tag,
             request_id,
             client_ip,
             len(req.prefix or ""),
             len(req.suffix or ""),
             req.languageId,
             req.model_thinking,
-            req.privacy_mode
-        )
-        llm_prefix, llm_suffix = prepare_prompt_context(req.prefix or "", req.suffix or "")
-        logger.info("[%s] llm_input_prefix=%r", request_id, llm_prefix)
-        logger.info("[%s] llm_input_suffix=%r", request_id, llm_suffix)
-        
-        system_prompt, user_prompt = build_completion_prompts(
-            req.prefix, 
-            req.suffix, 
-            req.languageId, 
-            location=location,
-            thinking_level=req.model_thinking,
-            preferences=req.user_preferences
-        )
-        result = await call_ollama(
-            user_prompt,
-            system_prompt=system_prompt,
-            tag=f"{request_id}-primary", 
-            temperature=0.7,
-            thinking=req.model_thinking if req.model_thinking != "none" else None
+            req.privacy_mode,
         )
 
+        llm_prefix, llm_suffix = prepare_prompt_context(req.prefix or "", req.suffix or "")
+        logger.info("[%s] llm_input_prefix=%r", request_tag, llm_prefix)
+        logger.info("[%s] llm_input_suffix=%r", request_tag, llm_suffix)
+
+        system_prompt, user_prompt = build_completion_prompts(
+            req.prefix,
+            req.suffix,
+            req.languageId,
+            location=location,
+            thinking_level=req.model_thinking,
+            preferences=req.user_preferences,
+        )
+
+        inference_task = asyncio.create_task(
+            call_ollama(
+                user_prompt,
+                system_prompt=system_prompt,
+                tag=f"{request_tag}-primary",
+                temperature=0.7,
+                thinking=req.model_thinking if req.model_thinking != "none" else None,
+            )
+        )
+
+        async with ACTIVE_COMPLETIONS_LOCK:
+            existing = ACTIVE_COMPLETIONS.get(request_id)
+            if existing and not existing.done():
+                existing.cancel()
+            ACTIVE_COMPLETIONS[request_id] = inference_task
+
+        result = await inference_task
         content = result["content"] or ""
         if not content.strip():
-            logger.warning("[%s] primary returned empty content, returning empty result", request_id)
+            logger.warning("[%s] primary returned empty content, returning empty result", request_tag)
         logger.info(
-            "[%s] completion resolved source=primary content_chars=%d content_preview='%s'",
+            "[%s] completion resolved source=primary request_id=%s content_chars=%d content_preview='%s'",
+            request_tag,
             request_id,
             len(content),
             _preview(content, 120),
         )
 
         async def generate():
-            yield f"data: {json.dumps({'content': content})}\n\n"
-            yield f"data: {json.dumps({'done': True})}\n\n"
+            yield _sse_payload({"content": content})
+            yield _sse_payload({"done": True})
 
         return StreamingResponse(generate(), media_type="text/event-stream")
+    except asyncio.CancelledError:
+        logger.info("[%s] /v1/completions cancelled request_id=%s", request_tag, request_id)
 
+        async def cancelled():
+            yield _sse_payload({"cancelled": True, "request_id": request_id, "done": True})
+
+        return StreamingResponse(cancelled(), media_type="text/event-stream")
     except Exception as e:
-        logger.exception("[%s] /v1/completions failed: %s", request_id, e)
+        logger.exception("[%s] /v1/completions failed request_id=%s: %s", request_tag, request_id, e)
         return JSONResponse(content={"error": str(e)}, status_code=500)
+    finally:
+        async with ACTIVE_COMPLETIONS_LOCK:
+            active = ACTIVE_COMPLETIONS.get(request_id)
+            if active is not None and active is inference_task:
+                ACTIVE_COMPLETIONS.pop(request_id, None)
+
+
+@app.post("/v1/completions/cancel")
+async def cancel_completion(req: CancelCompletionRequest, api_key: str = Security(get_api_key)):
+    request_tag = str(uuid.uuid4())[:8]
+    request_id = req.request_id or ""
+
+    async with ACTIVE_COMPLETIONS_LOCK:
+        task = ACTIVE_COMPLETIONS.get(request_id)
+        if task is None:
+            logger.info(
+                "[%s] /v1/completions/cancel request_id=%s status=not_found reason=%s",
+                request_tag,
+                request_id,
+                req.reason,
+            )
+            return {"cancelled": False, "status": "not_found"}
+
+        if task.done():
+            logger.info(
+                "[%s] /v1/completions/cancel request_id=%s status=already_done reason=%s",
+                request_tag,
+                request_id,
+                req.reason,
+            )
+            return {"cancelled": False, "status": "already_done"}
+
+        task.cancel()
+
+    logger.info(
+        "[%s] /v1/completions/cancel request_id=%s status=ok reason=%s",
+        request_tag,
+        request_id,
+        req.reason,
+    )
+    return {"cancelled": True, "status": "ok"}
+
 
 @app.post("/v1/ocr")
 async def ocr_image(request: OCRRequest, api_key: str = Security(get_api_key)):
@@ -159,6 +242,8 @@ async def ocr_image(request: OCRRequest, api_key: str = Security(get_api_key)):
         logger.exception("[%s] /v1/ocr failed: %s", request_id, e)
         return JSONResponse(content={"error": str(e)}, status_code=500)
 
+
 if __name__ == "__main__":
     import uvicorn
+
     uvicorn.run(app, host="0.0.0.0", port=8001)

backend/prompt.py

@@ -63,6 +63,14 @@ def _prepare_context(prefix: str, suffix: str) -> Tuple[str, str]:
 
 
 FENCE_LINE_RE = re.compile(r"^[ \t]*```.*$")
+FENCE_INFO_RE = re.compile(r"^[ \t]*```[ \t]*(.*)$")
+MERMAID_CONTEXT_RE = re.compile(
+    r"```[ \t]*mermaid\b|"
+    r"\b(flowchart|sequencediagram|classdiagram|statediagram(?:-v2)?|"
+    r"erdiagram|journey|gantt|pie|mindmap|timeline|gitgraph|quadrantchart|xychart-beta)\b|"
+    r"\bgraph[ \t]+(TD|TB|BT|RL|LR)\b",
+    re.IGNORECASE,
+)
 
 
 def _cursor_in_fenced_code_block(prefix: str) -> bool:
@@ -71,12 +79,48 @@ def _cursor_in_fenced_code_block(prefix: str) -> bool:
     The state is computed by toggling on each markdown fence line that matches:
     ^[ \t]*```.*$
     """
+    return _active_fence_language(prefix) != "none"
+
+
+def _active_fence_language(prefix: str) -> str:
+    """
+    Return active fence language at cursor based on prefix.
+    - "none": cursor is outside fenced code block
+    - "unknown": cursor is inside a fence without language tag
+    - "<language>": cursor is inside a fenced block with language tag
+    """
     normalized = _normalize_newlines(prefix)
     in_fence = False
+    active_language = "none"
     for line in normalized.split("\n"):
         if FENCE_LINE_RE.match(line):
-            in_fence = not in_fence
-    return in_fence
+            if in_fence:
+                in_fence = False
+                active_language = "none"
+            else:
+                info_match = FENCE_INFO_RE.match(line)
+                info = info_match.group(1).strip() if info_match else ""
+                if not info:
+                    active_language = "unknown"
+                else:
+                    first_token = info.split()[0]
+                    lang_chars = []
+                    for ch in first_token.strip():
+                        if ch.isalnum() or ch in "-_+.":
+                            lang_chars.append(ch)
+                    active_language = "".join(lang_chars)[:32].lower() or "unknown"
+                in_fence = True
+    return active_language if in_fence else "none"
+
+
+def _is_mermaid_context(prefix: str, suffix: str, cursor_fence_language: str) -> bool:
+    if cursor_fence_language == "mermaid":
+        return True
+
+    prefix_tail = (prefix or "")[-1200:]
+    suffix_head = (suffix or "")[:400]
+    combined = f"{prefix_tail}\n{suffix_head}"
+    return MERMAID_CONTEXT_RE.search(combined) is not None
 
 
 def prepare_prompt_context(prefix: str, suffix: str) -> Tuple[str, str]:
@@ -113,22 +157,36 @@ Hard constraints you must follow:
   - Do not output code snippets as inline backticks.
   - Choose the language tag from context (no default fallback tag instruction).
 
-4) Boundary newline repair:
+4) Mermaid-specific completion rules:
+- Read CURSOR_FENCE_LANGUAGE and MERMAID_CONTEXT from the user prompt.
+- If CURSOR_FENCE_LANGUAGE=mermaid:
+  - Output Mermaid statements only.
+  - Never output triple backticks.
+  - Never output prose explanations.
+- If CURSOR_IN_FENCED_CODE_BLOCK=false and MERMAID_CONTEXT=true:
+  - Output a complete Mermaid fenced block:
+    ```mermaid
+    ...
+    ```
+  - Keep Mermaid syntax valid and concise.
+- Never mix Mermaid code and explanatory narration in one output.
+
+5) Boundary newline repair:
 - Read PREFIX_ENDS_WITH_NEWLINE and SUFFIX_STARTS_WITH_NEWLINE from the user prompt.
 - Carefully reason about whether OUTPUT should start or end with a newline.
 - If PREFIX lacks a required boundary newline, add it at OUTPUT start.
 - If SUFFIX lacks a required boundary newline, add it at OUTPUT end.
 - Ensure PREFIX + OUTPUT + SUFFIX is structurally natural.
 
-5) Context stitching:
+6) Context stitching:
 - Do not repeat text that already appears at the start of SUFFIX.
 - Preserve nearby language, tone, punctuation, indentation, and markdown structure.
 - Continue existing structures naturally (lists, tables, block quotes, headings).
 
-6) OCR safety:
+7) OCR safety:
 - PREFIX may include hidden OCR metadata tags like <OCR:...>.
 - Never output any OCR tag.
-- Never output strings containing <OCR: or > as OCR artifacts."""
+- Never output OCR tag fragments such as <OCR:...>."""
     return system_prompt.strip()
 
 
@@ -227,6 +285,29 @@ The area is $A = \\pi r^2$.
 
 ```javascript
 const area = (r) => Math.PI * r * r;
+```
+
+[EX13] Cursor inside mermaid fence: no backticks, mermaid lines only
+CURSOR_IN_FENCED_CODE_BLOCK=true
+CURSOR_FENCE_LANGUAGE=mermaid
+<PREFIX>```mermaid
+flowchart TD
+A[Start] --> </PREFIX>
+<SUFFIX>
+```</SUFFIX>
+Expected OUTPUT:
+B{Valid?}
+B -->|Yes| C[Done]
+
+[EX14] Mermaid context outside fence: return full mermaid block
+CURSOR_IN_FENCED_CODE_BLOCK=false
+MERMAID_CONTEXT=true
+<PREFIX>Please provide a simple release pipeline diagram.</PREFIX>
+<SUFFIX></SUFFIX>
+Expected OUTPUT:
+```mermaid
+flowchart LR
+Build --> Test --> Deploy
 ```"""
 
 
@@ -243,7 +324,11 @@ def build_completion_prompts(
     recent_prefix = _normalize_newlines(recent_prefix)
     recent_suffix = _normalize_newlines(recent_suffix)
 
-    cursor_in_fenced_code_block = _cursor_in_fenced_code_block(recent_prefix)
+    cursor_fence_language = _active_fence_language(recent_prefix)
+    cursor_in_fenced_code_block = cursor_fence_language != "none"
+    mermaid_context = _is_mermaid_context(
+        recent_prefix, recent_suffix, cursor_fence_language
+    )
     prefix_ends_with_newline = recent_prefix.endswith("\n")
     suffix_starts_with_newline = recent_suffix.startswith("\n")
 
@@ -268,6 +353,8 @@ Editor language id: {safe_language_id}
 
 Completion state flags:
 - CURSOR_IN_FENCED_CODE_BLOCK: {"true" if cursor_in_fenced_code_block else "false"}
+- CURSOR_FENCE_LANGUAGE: {cursor_fence_language}
+- MERMAID_CONTEXT: {"true" if mermaid_context else "false"}
 - PREFIX_ENDS_WITH_NEWLINE: {"true" if prefix_ends_with_newline else "false"}
 - SUFFIX_STARTS_WITH_NEWLINE: {"true" if suffix_starts_with_newline else "false"}
 

backend/tests/test_main_cancel.py

@@ -0,0 +1,118 @@
+import asyncio
+import importlib
+import sys
+import threading
+from pathlib import Path
+
+import pytest
+from fastapi.testclient import TestClient
+
+
+BACKEND_DIR = Path(__file__).resolve().parents[1]
+if str(BACKEND_DIR) not in sys.path:
+    sys.path.insert(0, str(BACKEND_DIR))
+
+try:
+    main = importlib.import_module("main")
+except ModuleNotFoundError:
+    pytest.skip("main module dependencies are not available", allow_module_level=True)
+
+
+API_KEY_HEADERS = {"X-API-Key": "your-secret-key-here"}
+
+
+def _completion_payload():
+    return {
+        "prefix": "hello",
+        "suffix": "",
+        "languageId": "markdown",
+        "model_thinking": "low",
+        "privacy_mode": True,
+    }
+
+
+def test_cancel_endpoint_cancels_running_task(monkeypatch):
+    main.ACTIVE_COMPLETIONS.clear()
+    started = threading.Event()
+    cancelled = threading.Event()
+
+    async def fake_call_ollama(*args, **kwargs):
+        started.set()
+        try:
+            while True:
+                await asyncio.sleep(0.05)
+        except asyncio.CancelledError:
+            cancelled.set()
+            raise
+
+    monkeypatch.setattr(main, "call_ollama", fake_call_ollama)
+    monkeypatch.setattr(main, "build_completion_prompts", lambda *a, **k: ("system", "user"))
+    monkeypatch.setattr(main, "prepare_prompt_context", lambda *a, **k: ("prefix", "suffix"))
+
+    with TestClient(main.app) as client:
+        request_id = "req-cancel-1"
+        completion_headers = {**API_KEY_HEADERS, "X-Request-Id": request_id}
+        response_box = {}
+
+        def send_completion():
+            response_box["response"] = client.post(
+                "/v1/completions",
+                headers=completion_headers,
+                json=_completion_payload(),
+            )
+
+        completion_thread = threading.Thread(target=send_completion, daemon=True)
+        completion_thread.start()
+
+        assert started.wait(timeout=2.0)
+
+        cancel_response = client.post(
+            "/v1/completions/cancel",
+            headers=API_KEY_HEADERS,
+            json={"request_id": request_id, "reason": "superseded"},
+        )
+        assert cancel_response.status_code == 200
+        assert cancel_response.json() == {"cancelled": True, "status": "ok"}
+
+        completion_thread.join(timeout=5.0)
+        assert not completion_thread.is_alive()
+        assert cancelled.wait(timeout=2.0)
+
+        completion_response = response_box["response"]
+        assert completion_response.status_code == 200
+        assert '"cancelled": true' in completion_response.text
+
+
+def test_cancel_not_found():
+    main.ACTIVE_COMPLETIONS.clear()
+    with TestClient(main.app) as client:
+        response = client.post(
+            "/v1/completions/cancel",
+            headers=API_KEY_HEADERS,
+            json={"request_id": "missing", "reason": "abort"},
+        )
+    assert response.status_code == 200
+    assert response.json() == {"cancelled": False, "status": "not_found"}
+
+
+def test_completion_normal_flow(monkeypatch):
+    main.ACTIVE_COMPLETIONS.clear()
+
+    async def fake_call_ollama(*args, **kwargs):
+        return {"content": "completion text", "think": ""}
+
+    monkeypatch.setattr(main, "call_ollama", fake_call_ollama)
+    monkeypatch.setattr(main, "build_completion_prompts", lambda *a, **k: ("system", "user"))
+    monkeypatch.setattr(main, "prepare_prompt_context", lambda *a, **k: ("prefix", "suffix"))
+
+    with TestClient(main.app) as client:
+        response = client.post(
+            "/v1/completions",
+            headers=API_KEY_HEADERS,
+            json=_completion_payload(),
+        )
+
+    assert response.status_code == 200
+    assert '"content": "completion text"' in response.text
+    assert '"done": true' in response.text
+    assert main.ACTIVE_COMPLETIONS == {}

backend/tests/test_prompt.py

@@ -21,7 +21,13 @@ def test_prompt_builds_system_and_user():
     assert "$...$" in system_prompt
     assert "$$...$$" in system_prompt
     assert "```{language}" in system_prompt
+    assert "Mermaid-specific completion rules" in system_prompt
+    assert "CURSOR_FENCE_LANGUAGE" in system_prompt
+    assert "MERMAID_CONTEXT" in system_prompt
+    assert "Output Mermaid statements only." in system_prompt
     assert "CURSOR_IN_FENCED_CODE_BLOCK" in user_prompt
+    assert "CURSOR_FENCE_LANGUAGE" in user_prompt
+    assert "MERMAID_CONTEXT" in user_prompt
     assert "PREFIX_ENDS_WITH_NEWLINE" in user_prompt
     assert "SUFFIX_STARTS_WITH_NEWLINE" in user_prompt
 
@@ -33,12 +39,22 @@ def test_cursor_in_fence_detection():
     assert prompt._cursor_in_fenced_code_block("text ```not-a-fence``` tail") is False
 
 
+def test_active_fence_language_detection():
+    assert prompt._active_fence_language("") == "none"
+    assert prompt._active_fence_language("```mermaid\nflowchart TD\nA-->B\n") == "mermaid"
+    assert prompt._active_fence_language("```python\nprint('x')\n") == "python"
+    assert prompt._active_fence_language("```\nline\n") == "unknown"
+    assert prompt._active_fence_language("```mermaid\nA-->B\n```\n") == "none"
+
+
 def test_newline_flags():
     _, user_prompt_a = prompt.build_completion_prompts(
         prefix="Hello",
         suffix="World",
     )
     assert "CURSOR_IN_FENCED_CODE_BLOCK: false" in user_prompt_a
+    assert "CURSOR_FENCE_LANGUAGE: none" in user_prompt_a
+    assert "MERMAID_CONTEXT: false" in user_prompt_a
     assert "PREFIX_ENDS_WITH_NEWLINE: false" in user_prompt_a
     assert "SUFFIX_STARTS_WITH_NEWLINE: false" in user_prompt_a
 
@@ -46,11 +62,30 @@ def test_newline_flags():
         prefix="Hello\n",
         suffix="\nWorld",
     )
+    assert "CURSOR_FENCE_LANGUAGE: none" in user_prompt_b
     assert "PREFIX_ENDS_WITH_NEWLINE: true" in user_prompt_b
     assert "SUFFIX_STARTS_WITH_NEWLINE: true" in user_prompt_b
 
 
+def test_mermaid_context_flags():
+    _, prompt_in_mermaid = prompt.build_completion_prompts(
+        prefix="```mermaid\nflowchart TD\nA --> ",
+        suffix="\n```",
+    )
+    assert "CURSOR_IN_FENCED_CODE_BLOCK: true" in prompt_in_mermaid
+    assert "CURSOR_FENCE_LANGUAGE: mermaid" in prompt_in_mermaid
+    assert "MERMAID_CONTEXT: true" in prompt_in_mermaid
+
+    _, prompt_mermaid_keyword = prompt.build_completion_prompts(
+        prefix="Please draw a mermaid flowchart for deploy pipeline.",
+        suffix="",
+    )
+    assert "CURSOR_IN_FENCED_CODE_BLOCK: false" in prompt_mermaid_keyword
+    assert "CURSOR_FENCE_LANGUAGE: none" in prompt_mermaid_keyword
+    assert "MERMAID_CONTEXT: true" in prompt_mermaid_keyword
+
+
 def test_examples_coverage():
     _, user_prompt = prompt.build_completion_prompts(prefix="", suffix="")
-    for ex in range(1, 13):
+    for ex in range(1, 15):
         assert f"[EX{ex:02d}]" in user_prompt