fix: normalize escaped Work artifacts

agent_platform/runtime/loop.py

@@ -7,6 +7,7 @@ import re
 import uuid
 from collections.abc import Awaitable, Callable
 from dataclasses import dataclass
+from pathlib import PurePosixPath
 from typing import Any
 
 from agent_platform.auth import UserIdentity
@@ -72,6 +73,46 @@ VERIFICATION_TOOLS = {"read_file", "git_status", "git_diff", "exec", "poll_proce
 EXECUTION_TOOLS = {"exec", "start_process", "poll_process"}
 REPEAT_GUARDED_EXECUTION_TOOLS = {"exec", "start_process"}
 MAX_CONSECUTIVE_CHECKPOINT_REJECTIONS = 3
+CODE_FILE_SUFFIXES = {
+    ".c",
+    ".cc",
+    ".cpp",
+    ".css",
+    ".go",
+    ".h",
+    ".hpp",
+    ".html",
+    ".java",
+    ".js",
+    ".jsx",
+    ".php",
+    ".py",
+    ".rb",
+    ".rs",
+    ".sh",
+    ".sql",
+    ".svelte",
+    ".ts",
+    ".tsx",
+    ".vue",
+}
+PLAIN_TEXT_FILE_SUFFIXES = {
+    ".csv",
+    ".json",
+    ".jsonl",
+    ".md",
+    ".rst",
+    ".toml",
+    ".tsv",
+    ".txt",
+    ".xml",
+    ".yaml",
+    ".yml",
+}
+READ_ONLY_EXEC_COMMAND = re.compile(
+    r"^(?:cat|cut|diff|find|git\s+(?:diff|show|status)|grep|head|jq|ls|pwd|rg|sed\s+-n|stat|tail|wc)\b",
+    re.IGNORECASE,
+)
 
 
 def latest_user_text(messages: list[dict[str, Any]]) -> str:
@@ -156,6 +197,109 @@ def execution_call_key(name: str, arguments: dict[str, Any]) -> str | None:
     return f"{name}\0{cwd}\0{command}"
 
 
+def decode_code_layout_escapes(content: str) -> str:
+    output: list[str] = []
+    index = 0
+    quote = ""
+    triple = False
+    escaped = False
+    line_comment = False
+    while index < len(content):
+        if quote:
+            if escaped:
+                output.append(content[index])
+                escaped = False
+                index += 1
+                continue
+            if content[index] == "\\":
+                output.append(content[index])
+                escaped = True
+                index += 1
+                continue
+            delimiter = quote * (3 if triple else 1)
+            if content.startswith(delimiter, index):
+                output.append(delimiter)
+                index += len(delimiter)
+                quote = ""
+                triple = False
+                continue
+            output.append(content[index])
+            index += 1
+            continue
+
+        if content.startswith("\\r\\n", index):
+            output.append("\n")
+            index += 4
+            line_comment = False
+            continue
+        if content.startswith("\\n", index):
+            output.append("\n")
+            index += 2
+            line_comment = False
+            continue
+        if content.startswith("\\t", index):
+            output.append("\t")
+            index += 2
+            continue
+        if line_comment:
+            if content[index] == "\n":
+                line_comment = False
+            output.append(content[index])
+            index += 1
+            continue
+        if content.startswith("//", index) or content[index] == "#":
+            line_comment = True
+            delimiter = "//" if content.startswith("//", index) else "#"
+            output.append(delimiter)
+            index += len(delimiter)
+            continue
+        if content[index] in {'"', "'"}:
+            quote = content[index]
+            triple = content.startswith(quote * 3, index)
+            delimiter = quote * (3 if triple else 1)
+            output.append(delimiter)
+            index += len(delimiter)
+            continue
+        output.append(content[index])
+        index += 1
+    return "".join(output)
+
+
+def normalize_write_file_content(path: str, content: str) -> str:
+    literal_newlines = content.count("\\n") + content.count("\\r\\n")
+    actual_newlines = content.count("\n")
+    if literal_newlines < 2 or literal_newlines <= max(2, actual_newlines * 2):
+        return content
+    suffix = PurePosixPath(path).suffix.lower()
+    if suffix in CODE_FILE_SUFFIXES:
+        return decode_code_layout_escapes(content)
+    if suffix in PLAIN_TEXT_FILE_SUFFIXES:
+        return content.replace("\\r\\n", "\n").replace("\\n", "\n").replace("\\t", "\t")
+    return content
+
+
+def normalize_tool_arguments(name: str, arguments: dict[str, Any]) -> dict[str, Any]:
+    if name != "write_file":
+        return arguments
+    path = arguments.get("path")
+    content = arguments.get("content")
+    if not isinstance(path, str) or not isinstance(content, str):
+        return arguments
+    normalized = normalize_write_file_content(path, content)
+    if normalized == content:
+        return arguments
+    return {**arguments, "content": normalized}
+
+
+def is_substantive_execution(name: str, arguments: dict[str, Any]) -> bool:
+    if name == "start_process":
+        return True
+    if name != "exec":
+        return False
+    command = str(arguments.get("command", "")).strip()
+    return bool(command and not READ_ONLY_EXEC_COMMAND.match(command))
+
+
 @dataclass(slots=True)
 class RunRecorder:
     store: RuntimeStore
@@ -382,10 +526,10 @@ class AgentLoop:
                         mutation_count += 1
                     elif name in VERIFICATION_TOOLS and mutation_count:
                         verified_mutation = mutation_count
-                    if name in EXECUTION_TOOLS:
+                    if is_substantive_execution(name, arguments):
                         successful_executions += 1
                         unresolved_execution_failure = None
-                elif name in EXECUTION_TOOLS:
+                elif is_substantive_execution(name, arguments):
                     unresolved_execution_failure = f"{name}: {public_tool_summary(result, 1000)}"
                     execution_failures.append(unresolved_execution_failure)
                     key = execution_call_key(name, arguments) if isinstance(arguments, dict) else None
@@ -458,6 +602,7 @@ class AgentLoop:
             except (json.JSONDecodeError, ValueError) as exc:
                 parsed.append((index, call, name, {"__parse_error__": str(exc)}, False))
                 continue
+            arguments = normalize_tool_arguments(name, arguments)
             metadata = TOOL_METADATA.get(name)
             parallel = bool(metadata and metadata.parallel_safe)
             if name == "delegate_task":