feat(job_writer): reject code that is not valid UTF-8

write_job_file now validates that the incoming code string is encodable
as UTF-8 BEFORE doing any file I/O. A lone surrogate (U+D800..U+DFFF) —
typically the result of a broken decode somewhere upstream in the
agent's pipeline — cannot be represented in UTF-8, and would otherwise
either silently produce mojibake on disk or crash with an opaque
UnicodeEncodeError from open().

Behavior:
  - try: code.encode('utf-8')
  - on UnicodeEncodeError: log a warning and raise ValueError with a
    message that points the agent at 're-generate the code as plain
    Unicode'. ValueError -> HTTP 400 via the existing server.py
    exception handler, so the agent gets a clean error and no file is
    written.
  - normal CJK / emoji / accented Latin still work — the encode check
    is essentially free.

Tests:
  - test_write_accepts_valid_utf8_including_cjk_and_emoji: positive
    case for the common LLM output patterns.
  - test_write_rejects_lone_surrogate: negative case, also asserts
    that tmp_path is empty (no half-written .py left behind).

Full suite 246 passed (+2 from 244).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

spark_executor/core/job_writer.py

@@ -33,12 +33,34 @@ def resolve_jobs_dir(jobs_dir: str | None = None) -> str:
     return settings.jobs_dir
 
 
+def _not_utf8_error(exc: UnicodeEncodeError) -> ValueError:
+    """Standard 400 error when the LLM-supplied code string cannot be
+    encoded as UTF-8 (typically because it contains a lone surrogate
+    from a broken decode earlier in the pipeline)."""
+    return ValueError(
+        f"code is not valid UTF-8 and cannot be written as a .py file. "
+        f"This usually means the LLM produced a string with a lone "
+        f"surrogate codepoint (U+D800..U+DFFF) — re-generate the code "
+        f"as plain Unicode. Underlying error: {exc}"
+    )
+
+
 def write_job_file(code: str, jobs_dir: str | None = None) -> str:
     """Write `code` to <jobs_dir>/job_<timestamp>_<rand>.py; return abs path.
 
     Creates the directory if missing. Filename is unique per call (timestamp
     down to the second + 3 random bytes) so concurrent agents cannot collide.
+
+    Raises ValueError (-> 400) if `code` cannot be encoded as UTF-8 (e.g.
+    contains a lone surrogate). The check is cheap and runs before any
+    file I/O, so a bad payload never produces a half-written .py.
     """
+    try:
+        code.encode("utf-8")
+    except UnicodeEncodeError as exc:
+        logger.warning(f"write_job_file rejected: code is not valid UTF-8: {exc}")
+        raise _not_utf8_error(exc) from exc
+
     effective_dir = resolve_jobs_dir(jobs_dir)
     os.makedirs(effective_dir, exist_ok=True)
     stamp = datetime.utcnow().strftime("%Y%m%d%H%M%S")

tests/unit/test_job_writer.py

@@ -96,3 +96,22 @@ def test_write_returns_unique_paths_for_concurrent_calls(tmp_path: Path):
     out1 = write_job_file("a", jobs_dir=str(tmp_path))
     out2 = write_job_file("b", jobs_dir=str(tmp_path))
     assert out1 != out2
+
+
+def test_write_accepts_valid_utf8_including_cjk_and_emoji(tmp_path: Path):
+    """Common LLM output — CJK + emoji — must encode as UTF-8 cleanly."""
+    code = "# 中文注释 ✨\nspark.read.parquet('/data/中文.parquet')\n"
+    out = write_job_file(code, jobs_dir=str(tmp_path))
+    with open(out, encoding="utf-8") as f:
+        assert f.read() == code
+
+
+def test_write_rejects_lone_surrogate(tmp_path: Path):
+    """A lone surrogate (U+D800..U+DFFF) is not a valid Unicode scalar value
+    and cannot be encoded as UTF-8. write_job_file must reject it BEFORE
+    writing — never leave a half-written .py on disk."""
+    bad = "print('ok')\n\udcff\noops"
+    with pytest.raises(ValueError, match="not valid UTF-8"):
+        write_job_file(bad, jobs_dir=str(tmp_path))
+    # No file was written.
+    assert list(tmp_path.iterdir()) == []