[FEATURE] GraphQL Escape Sequence Lint Gap and Untested Mutation Syntax

pyproject.toml

@@ -110,6 +110,11 @@ markers = [
     "large: full integration — everything allowed (default for unannotated tests)",
     "feature(name): marks tests requiring a specific feature gate to be enabled",
 ]
+filterwarnings = [
+    "error::SyntaxWarning",
+    "error::DeprecationWarning:autoskillit",
+    "default:AUTOSKILLIT_HEADLESS=1 without:DeprecationWarning",
+]
 
 [tool.coverage.run]
 parallel = true
@@ -120,7 +125,7 @@ target-version = "py311"
 line-length = 99
 
 [tool.ruff.lint]
-select = ["E", "F", "I", "UP", "TID251"]
+select = ["E", "F", "I", "UP", "TID251", "W"]
 
 [tool.ruff.lint.per-file-ignores]
 "tests/execution/test_session_classification_e2e.py" = ["E402"]

src/autoskillit/recipe/_cmd_rpc_issues.py

@@ -115,6 +115,23 @@ def export_local_bundle(
 # ─── batch_create_issues helpers ────────────────────────────────────────────
 
 
+def _validate_mutation_variables(mutation: str, variables: dict[str, object]) -> None:
+    r"""Assert that mutation variable declarations match the variables dict.
+
+    Guards against: (1) variable-name drift between the mutation template and
+    the variables dict, and (2) the ruff W605 raw-string auto-fix trap — if
+    ruff converts f"$i{k}" to rf"\$i{k}", the runtime string contains literal
+    backslash-dollar and this check fails because "$i0:" is absent.
+    """
+    for key in variables:
+        decl = f"${key}:"
+        ref = f"${key})"
+        if decl not in mutation:
+            raise ValueError(f"Variable ${key} in variables dict but not declared in mutation")
+        if ref not in mutation:
+            raise ValueError(f"Variable ${key} declared but not referenced in mutation body")
+
+
 def _extract_title(raw: str) -> str:
     """Return the text following '# ' from the first H1 line, or a fallback."""
     m = re.search(r"^#\s+(.+)$", raw, re.MULTILINE)
@@ -290,7 +307,7 @@ def batch_create_issues(
         for idx, (title, body, _) in enumerate(chunk):
             alias = f"issue{idx}"
             mutation_parts.append(
-                f"{alias}: createIssue(input: \$i{idx}) {{ issue {{ number url }} }}"
+                f"{alias}: createIssue(input: $i{idx}) {{ issue {{ number url }} }}"
             )
             variables[f"i{idx}"] = {
                 "repositoryId": repo_id,
@@ -300,11 +317,12 @@ def batch_create_issues(
             }
         mutation = (
             "mutation("
-            + ",".join(f"\$i{k}: CreateIssueInput!" for k in range(len(chunk)))
+            + ",".join(f"$i{k}: CreateIssueInput!" for k in range(len(chunk)))
             + ") {"
             + " ".join(mutation_parts)
             + "}"
         )
+        _validate_mutation_variables(mutation, variables)
         payload = json.dumps({"query": mutation, "variables": variables})
         result = run_gh(["api", "graphql", "--input", "-"], cwd=workspace, input_data=payload)
         if result.returncode != 0:

tests/recipe/test_cmd_rpc.py

@@ -5,6 +5,7 @@
 import json
 import os
 import subprocess
+import sys
 from pathlib import Path
 from unittest.mock import patch
 
@@ -442,6 +443,8 @@ def test_refetch_issues_builds_query():
     call_args = mock_run_gh.call_args[0][0]
     assert "graphql" in call_args
     query_arg = next(a for a in call_args if a.startswith("query="))
+    assert query_arg.startswith("query={")
+    assert "repository(owner:" in query_arg
     assert "org" in query_arg
     assert "repo" in query_arg
     assert "issue(number: 1)" in query_arg
@@ -597,8 +600,12 @@ def test_batch_create_issues_constructs_graphql_mutation(tmp_path):
             mutation_call = json.loads(kwargs["input_data"])
             query = mutation_call["query"]
             variables = mutation_call["variables"]
+            assert query.startswith("mutation(")
             assert "issue0: createIssue" in query
             assert "issue1: createIssue" in query
+            for var_name in variables:
+                assert f"${var_name}:" in query, f"Variable ${var_name} not declared in mutation"
+                assert f"${var_name})" in query, f"Variable ${var_name} not referenced in mutation"
             assert variables["i0"]["repositoryId"] == "R_123"
             assert variables["i1"]["repositoryId"] == "R_123"
             assert variables["i0"]["labelIds"] == ["L_1", "L_2"]
@@ -608,6 +615,19 @@ def test_batch_create_issues_constructs_graphql_mutation(tmp_path):
     assert found, "no createIssue mutation call found in mock_run_gh calls"
 
 
+def test_validate_mutation_variables_catches_mismatch():
+    from autoskillit.recipe._cmd_rpc_issues import _validate_mutation_variables
+
+    with pytest.raises(ValueError, match="not declared"):
+        _validate_mutation_variables("mutation() { ... }", {"i0": {}})
+    with pytest.raises(ValueError, match="not referenced"):
+        _validate_mutation_variables("mutation($i0: CreateIssueInput!) { }", {"i0": {}})
+    _validate_mutation_variables(
+        "mutation($i0: CreateIssueInput!) { issue0: createIssue(input: $i0) { ... } }",
+        {"i0": {}},
+    )
+
+
 def test_batch_create_issues_chunks_large_batches(tmp_path):
     va_dir = tmp_path / ".autoskillit" / "temp" / "validate-audit"
     va_dir.mkdir(parents=True)
@@ -693,12 +713,19 @@ def side_effect(_args, **_kwargs):
 
         mock_run_gh.side_effect = side_effect_factory()
         result = batch_create_issues(workspace=str(tmp_path), chunk_size="10")
-    mutation_calls = sum(
-        1
-        for call in mock_run_gh.call_args_list
-        if call[1].get("input_data") and "createIssue" in call[1]["input_data"]
-    )
-    assert mutation_calls == 3
+    mutation_calls = []
+    for call in mock_run_gh.call_args_list:
+        kwargs = call[1]
+        if kwargs.get("input_data") and "createIssue" in kwargs["input_data"]:
+            mutation_call = json.loads(kwargs["input_data"])
+            query = mutation_call["query"]
+            variables = mutation_call["variables"]
+            assert query.startswith("mutation(")
+            for var_name in variables:
+                assert f"${var_name}:" in query, f"Variable ${var_name} not declared in mutation"
+                assert f"${var_name})" in query, f"Variable ${var_name} not referenced in mutation"
+            mutation_calls.append(mutation_call)
+    assert len(mutation_calls) == 3
     assert result["issue_count"] == "25"
 
 
@@ -1326,3 +1353,37 @@ def _small_per_file(args: list[str], **kwargs):
         ):
             result = _check_regression(str(tmp_path), files_small, "main", file_status_small)
     assert result is None, f"Expected None (aggregate > 10 but per-file all < 5) but got: {result}"
+
+
+def test_cmd_rpc_issues_no_invalid_escape_sequences():
+    """Source-level guard: invalid escapes must not appear in GraphQL builders."""
+    ruff = Path(sys.executable).resolve().parent / "ruff"
+    if not ruff.is_file():
+        pytest.skip("ruff not found in venv bin directory")
+    target = (
+        Path(__file__).resolve().parent.parent.parent
+        / "src"
+        / "autoskillit"
+        / "recipe"
+        / "_cmd_rpc_issues.py"
+    )
+    assert target.is_file(), f"Target file not found: {target}"
+    result = subprocess.run(
+        [
+            str(ruff),
+            "check",
+            "--select",
+            "W605",
+            "--output-format",
+            "json",
+            str(target),
+        ],
+        capture_output=True,
+        text=True,
+    )
+    assert result.returncode in (0, 1), f"ruff failed (exit {result.returncode}): {result.stderr}"
+    findings = json.loads(result.stdout) if result.stdout.strip() else []
+    assert findings == [], f"Invalid escape sequences found: {findings}"
+    assert result.returncode == 0, (
+        f"ruff exited {result.returncode} but produced no JSON findings: {result.stderr}"
+    )