fix(goal): base no-progress counting on successful operations (#7271)

* fix(goal): track successful operations for no-progress checks

* fix(goal): reject failed exits and narration-only progress

---------

Co-authored-by: Totoro-qaq <279883115+Totoro-qaq@users.noreply.github.com>

README.md

@@ -2246,7 +2246,7 @@ Supported commands:
 
 After each Gateway-backed run, DeerFlow evaluates the visible conversation, including the assistant's tool calls and shortened tool results, against the active goal with a non-thinking evaluator model. A successful tool result alone does not satisfy a goal, and when the assistant had to guess missing or ambiguous information the evaluator reports `needs_user_input`. The evaluator must return a typed blocker (`missing_evidence`, `needs_user_input`, `run_failed`, `external_wait`, or `goal_not_met_yet`) plus visible evidence. DeerFlow only injects a hidden continuation when the latest assistant turn is durably checkpointed, the blocker is `goal_not_met_yet`, the thread did not change during evaluation, and the no-progress breaker has not fired. The safety cap defaults to 8 hidden continuations, and repeated identical non-progress evaluations stop after 2 attempts. `/goal clear` and any user-authored new input win over queued continuations. When the evaluator considers the goal satisfied, DeerFlow clears it after the run successfully finalizes and publishes the updated thread state. The same checkpoint records the completion as `goal_outcome`, which stays until the next goal change. If required artifact delivery or its receipt persistence fails, the run reports an error and the goal remains active for a later retry; this does not start another hidden continuation.
 
-An unmet goal also stops automatic continuation after a token or loop-detection hard stop; it stays active for a later user turn.
+For tool-using turns, successful operation results drive the no-progress count: changing only reply wording or a sandbox tool's UI explanation does not reset it, while newly observed work can keep it going. Text-only turns keep their reply-based check. An unmet goal still stops automatic continuation after a token or loop-detection hard stop and stays active for a later user turn.
 
 Regenerating an earlier response keeps the goal currently set for the chat and its used continuation budget. It does not restore a goal that was cleared or achieved. The earlier answer's completion record is removed when its historical checkpoint is replayed.
 

backend/docs/runtime-guidance-details.md

@@ -2,6 +2,28 @@
 
 The runtime AGENTS.md indexes these detailed contracts. Preserve their behavior when changing the corresponding code.
 
+**Goal no-progress counting** (`runtime/goal_progress.py`): pair actual tool calls
+with their results in the current human/hidden-continuation interval. Normalize
+success/error metadata without mutating checkpoint messages. Retain only exact
+successful-operation hashes in the active Goal's `last_evaluation`; no global
+cache, arbitrary operation-count cutoff, file inspection or extra model call.
+Repeated results, UI-only sandbox descriptions, failed commands, receipt IDs and
+timestamps alone do not reset the counter. Validate receipts against the original
+arguments before excluding narration from operation identity. Shared shell-result
+classification recognizes nonzero exit markers, correcting older generic success
+metadata while preserving explicit producer stamps. A successful new operation is
+activity, not proof of usefulness or completion. Pure-text turns retain the
+existing reply comparison.
+
+Already evaluated hashes survive checkpoint restart and compaction. Results
+removed before observation cannot be reconstructed; do not claim complete tool
+history capture. Goal replacement/clearing drops the active set, while historical
+checkpoints retain their existing lifecycle. Storage grows with observed work;
+measure it rather than reintroducing a silent fallback to text after a fixed
+number of operations. Preserve old counters on first migration, merge against
+freshly read Goal state when persisting, and leave all existing stop, durability,
+cancellation and racing-user checks unchanged.
+
 **JSONL tail recovery** (`runtime/events/store/jsonl.py`, #6520): every append
 path goes through `_append_records`. A nonempty file whose last byte is not LF
 gets a separator written first, so a crash that left either valid JSON without

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

@@ -94,7 +94,7 @@ No match: skip inspection, log ID. Tests: `test_read_mark_tool_call_correlation.
 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. Known gaps: a process substitution handed straight to an interpreter (`bash <(curl u)`), or hidden behind a flag that takes its value as the next word (`bash -o pipefail < <(curl u)`, `bash --rcfile f < <(curl u)`), is undetected, and two-step forms (`x=$(curl u); eval "$x"`) need dataflow analysis. 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.
 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.
+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. Shell exit classification shares `utils/bash_exit.py` with subagent evidence; nonzero markers correct old generic-success metadata, preserving explicit producer stamps. 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.
 16. **ArtifactCaptureMiddleware** - See [tool artifacts](TOOL_ARTIFACTS.md).
 
 Authorization identity is independent of enforcement. Gateway strips client identity overrides: only the server auth source sets `is_internal`, and only authenticated IM `body.context` supplies `channel_user_id` (never `body.config`). `build_principal_from_context` applies role defaults, strict provenance, and copied attributes; RBAC rejects unknown defaults. Delegation and `GuardrailMiddleware` share this identity. Layer 1 precedes deferred assembly across agent paths and its provider is reused for Layer 2; framework skill/memory ordering stays stable. Trusted `DeerFlowClient.stream()` accepts identity overrides. Its graph key always includes effective storage `user_id` and, when enforced, the full Principal; nested attributes are copied so mutation cannot hide stale cache state.

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

@@ -15,6 +15,8 @@ from typing import Literal, get_args
 from langchain_core.messages import ToolMessage
 from langgraph.types import Command
 
+from deerflow.utils.bash_exit import BASH_TOOL_NAMES, bash_exit_status
+
 TOOL_META_KEY = "deerflow_tool_meta"
 PROGRESS_GUARD_ERROR_TYPE = "blocked_by_progress_guard"
 
@@ -259,10 +261,16 @@ _SUBAGENT_FAILURE_STATUSES = frozenset({"failed", "cancelled", "timed_out", "pol
 
 
 def normalize_tool_message(msg: ToolMessage) -> ToolMessage:
-    """Attach deerflow_tool_meta to a ToolMessage if not already present."""
+    """Attach tool status, correcting legacy bash analysis that missed an exit."""
+    content = msg.content if isinstance(msg.content, str) else ""
+    bash_status = bash_exit_status(content) if msg.name in BASH_TOOL_NAMES else None
+    failed_bash = bash_status is not None and bash_status[0] == "error"
     existing = (msg.additional_kwargs or {}).get(TOOL_META_KEY)
     if existing is not None:
-        return msg
+        # Explicit producer stamps remain authoritative. Older checkpoints can
+        # contain generic analysis success for a provider-reported failed exit.
+        if not (failed_bash and isinstance(existing, dict) and existing.get("source") == "content_analysis" and existing.get("status") == "success"):
+            return msg
 
     kwargs = msg.additional_kwargs or {}
     subagent_status = kwargs.get("subagent_status")
@@ -277,7 +285,6 @@ def normalize_tool_message(msg: ToolMessage) -> ToolMessage:
         msg.additional_kwargs = updated_kwargs
         return msg
 
-    content = msg.content if isinstance(msg.content, str) else ""
     # Pre-compute once; reused by the partial-success marker check below to avoid calling
     # content.lower() once per _PARTIAL_MARKERS entry inside the generator.
     content_lower = content.lower()
@@ -308,6 +315,11 @@ def normalize_tool_message(msg: ToolMessage) -> ToolMessage:
     elif (json_error := _extract_json_error_text(content)) is not None:
         attrs = _classify_error_text(json_error)
         meta = _make_meta(status="error", source="tool_return", **attrs)
+    elif failed_bash:
+        # A nonzero shell exit does not raise a Python exception. Keep it
+        # recoverable: grep/diff and diagnostic probes can legitimately use 1.
+        # Do not derive auth/stop categories from arbitrary stdout numbers.
+        meta = _make_meta(status="error", source="content_analysis", **_UNKNOWN_ERROR)
     elif (shell_attrs := _classify_error_shell(msg, content)) is not None:
         meta = _make_meta(status="error", source="content_analysis", **shell_attrs)
     elif any(m in content_lower for m in _PARTIAL_MARKERS):

backend/packages/harness/deerflow/runtime/AGENTS.md

@@ -116,7 +116,7 @@ answers for paginated history. See `docs/skill-usage-ui.md`.
 
 **Workspace change detection:** Truncated snapshots cannot prove absence. Mark skipped delivery verification in the receipt. Details: `backend/docs/runtime-guidance-details.md`.
 
-Unmet goals stand down on run-level token/loop hard stops and retain the original stop reason; hidden continuations must not re-enter that run's shared safety state.
+Unmet goals stand down on run-level token/loop hard stops and retain the original stop reason; hidden continuations must not re-enter that run's shared safety state. Operation-based no-progress counting: `backend/docs/runtime-guidance-details.md`.
 
 **Deferred terminal commit:** With an event store, the worker stages its terminal
 status locally and commits it only after finalization's receipt and duration

backend/packages/harness/deerflow/runtime/goal.py

@@ -28,6 +28,7 @@ from deerflow.agents.human_input import read_human_input_response
 from deerflow.agents.interaction_policy import RunInteractionPolicy
 from deerflow.config.pii_redaction_config import PiiRedactionConfig
 from deerflow.models import create_chat_model
+from deerflow.runtime.goal_progress import previous_goal_operations
 from deerflow.runtime.keyed_lock import AsyncKeyedLockTable
 from deerflow.tracing import inject_langfuse_metadata
 from deerflow.utils.file_io import await_drained
@@ -644,15 +645,21 @@ def compute_goal_progress_key(evaluation: GoalEvaluation, *, evidence_signature:
     )
 
 
-def compute_no_progress_count(goal: GoalState, evaluation: GoalEvaluation, *, evidence_signature: str = "") -> int:
-    """Increment repeated-progress count when visible evidence has not advanced."""
+def compute_no_progress_count(goal: GoalState, evaluation: GoalEvaluation, *, evidence_signature: str = "", operation_progress: dict[str, Any] | None = None) -> int:
+    """Count repeated tool work, preserving the original pure-text behavior."""
     if evaluation["satisfied"]:
         return 0
     progress_key = compute_goal_progress_key(evaluation, evidence_signature=evidence_signature)
     previous = goal.get("last_evaluation", {})
-    if isinstance(previous, dict) and previous.get("progress_key") == progress_key:
-        return int(goal.get("no_progress_count", 0)) + 1
-    return 0
+    old_count = int(goal.get("no_progress_count", 0))
+    text_count = old_count + 1 if isinstance(previous, dict) and previous.get("progress_key") == progress_key else 0
+    if operation_progress is None or previous_goal_operations(goal) is None:
+        # The existing text progress key is unchanged. Preserve its exact count
+        # on the first observation, rather than inventing a migration stop rule.
+        return text_count
+    if operation_progress.get("has_tool_activity"):
+        return 0 if operation_progress.get("new_keys") else old_count + 1
+    return text_count
 
 
 def make_goal_continuation_message(goal: GoalState, evaluation: GoalEvaluation, *, pii_redaction: PiiRedactionConfig | None = None) -> HumanMessage:
@@ -846,6 +853,7 @@ def attach_goal_evaluation(
     no_progress_count: int | None = None,
     stand_down_reason: str | None = None,
     evidence_signature: str = "",
+    operation_progress: dict[str, Any] | None = None,
 ) -> GoalState:
     """Return a goal copy with the latest evaluator result attached."""
     next_goal = copy.deepcopy(goal)
@@ -864,6 +872,12 @@ def attach_goal_evaluation(
         "evaluated_at": next_goal["updated_at"],
         "progress_key": compute_goal_progress_key(evaluation, evidence_signature=evidence_signature),
     }
+    if operation_progress is not None:
+        current_operations = previous_goal_operations(goal)
+        next_goal["last_evaluation"]["operation_progress"] = {
+            "version": operation_progress["version"],
+            "seen_keys": sorted(set(operation_progress["seen_keys"]) | set(current_operations["seen_keys"] if current_operations is not None else [])),
+        }
     if stand_down_reason:
         next_goal["last_evaluation"]["stand_down_reason"] = stand_down_reason
     return next_goal

backend/packages/harness/deerflow/runtime/goal_progress.py

@@ -0,0 +1,147 @@
+"""Successful-operation observations for Goal's no-progress counter.
+
+These records measure newly observed operations, not correctness or usefulness.
+Only hashes are retained, in the current Goal's existing evaluation record.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+from collections.abc import Mapping
+from typing import Any
+
+from langchain_core.messages import HumanMessage, ToolMessage
+
+from deerflow.agents.middlewares.message_utils import is_genuine_user_message
+from deerflow.agents.middlewares.tool_receipt import TOOL_RECEIPT_KEY, is_valid_receipt, make_tool_receipt
+from deerflow.agents.middlewares.tool_result_meta import TOOL_META_KEY, normalize_tool_message
+from deerflow.agents.middlewares.tool_transform_meta import TOOL_TRANSFORMS_KEY
+
+# Native sandbox tools use this argument only to explain the call in the UI.
+# Other tools may persist or send a description as their actual payload.
+_UI_NARRATION_TOOLS = frozenset({"bash", "read_file", "write_file", "str_replace", "ls", "glob", "grep"})
+
+
+def _field(message: Any, name: str, default: Any = None) -> Any:
+    return message.get(name, default) if isinstance(message, dict) else getattr(message, name, default)
+
+
+def _kind(message: Any) -> str | None:
+    kind = _field(message, "type") or _field(message, "role")
+    return {"user": "human", "assistant": "ai"}.get(kind, kind)
+
+
+def _starts_goal_turn(message: Any) -> bool:
+    if _kind(message) != "human":
+        return False
+    kwargs = _field(message, "additional_kwargs", {}) or {}
+    kwargs = dict(kwargs) if isinstance(kwargs, Mapping) else {}
+    if kwargs.get("deerflow_goal_continuation") is True:
+        return True
+    # Share the established card-answer/summary/framework distinction. Only
+    # boundary metadata is needed to adapt serialized checkpoint messages.
+    candidate = message if isinstance(message, HumanMessage) else HumanMessage(content="", name=_field(message, "name"), additional_kwargs=kwargs)
+    return is_genuine_user_message(candidate)
+
+
+def previous_goal_operations(goal: Mapping[str, Any]) -> dict[str, Any] | None:
+    previous = goal.get("last_evaluation")
+    record = previous.get("operation_progress") if isinstance(previous, Mapping) else None
+    if not isinstance(record, dict) or record.get("version") != 1:
+        return None
+    keys = record.get("seen_keys")
+    if not isinstance(keys, list) or any(not isinstance(key, str) or len(key) != 64 or any(char not in "0123456789abcdef" for char in key) for key in keys):
+        return None
+    return record
+
+
+def _successful_operation_key(call: dict[str, Any], message: Any) -> str | None:
+    name = call.get("name")
+    args = call.get("args")
+    if not isinstance(name, str) or not name or not isinstance(args, dict):
+        return None
+    if _field(message, "status") == "error":
+        return None
+    kwargs = _field(message, "additional_kwargs", {}) or {}
+    if not isinstance(kwargs, dict):
+        return None
+    receipt = kwargs.get(TOOL_RECEIPT_KEY)
+    if is_valid_receipt(receipt) and receipt["tool_call_id"] == call["id"] and receipt["tool_name"] == name and receipt["status"] != "success":
+        return None
+    # Reuse the established classification for old messages without structured
+    # status. Work on a copy: observing a checkpoint must never rewrite it.
+    status_message = normalize_tool_message(ToolMessage(content=_field(message, "content", ""), tool_call_id=str(call["id"]), name=name, additional_kwargs=dict(kwargs)))
+    meta = status_message.additional_kwargs.get(TOOL_META_KEY)
+    if not isinstance(meta, Mapping) or meta.get("status") != "success":
+        return None
+    # Use the native canonical format on the fallback path too, so enabling
+    # receipts does not turn an old operation into a new one. The generated
+    # timestamp is discarded; only the existing fingerprint fields are retained.
+    canonical = make_tool_receipt(call, status_message)
+    matching = (
+        is_valid_receipt(receipt)
+        and receipt["tool_call_id"] == call["id"]
+        and receipt["tool_name"] == name
+        and receipt["args_sha256"] == canonical["args_sha256"]
+        and receipt["output_bytes"] >= 0
+        and len(receipt["output_sha256"]) == len(canonical["output_sha256"])
+        and all(char in "0123456789abcdef" for char in receipt["output_sha256"])
+    )
+    if matching:
+        fingerprint = receipt
+    elif kwargs.get(TOOL_TRANSFORMS_KEY):
+        # Externalized previews contain per-call paths. Without a matching raw
+        # receipt their changing wrappers cannot establish a new result; do not
+        # read saved files or grant progress from a partial transformed view.
+        return None
+    else:
+        fingerprint = canonical
+    args_key = fingerprint["args_sha256"]
+    if name in _UI_NARRATION_TOOLS and "description" in args:
+        # Validate the receipt against raw arguments above. Only repetition
+        # identity excludes narration; native receipts remain unchanged.
+        operation_args = {key: value for key, value in args.items() if key != "description"}
+        args_key = hashlib.sha256(json.dumps(operation_args, sort_keys=True, default=str).encode("utf-8")).hexdigest()[: len(canonical["args_sha256"])]
+    body = json.dumps([name, args_key, fingerprint["output_sha256"], fingerprint["output_bytes"]])
+    return hashlib.sha256(body.encode("utf-8")).hexdigest()
+
+
+def observe_goal_operations(goal: Mapping[str, Any], messages: list[Any]) -> dict[str, Any]:
+    """Observe paired successful results since this turn's initiating input.
+
+    Genuine user/card replies and explicit hidden Goal continuations start turns;
+    framework reminders do not. Already evaluated operations survive compaction
+    and restart in ``last_evaluation``. There is no global cache or arbitrary
+    operation cutoff; retained storage follows actual work.
+    """
+    previous = previous_goal_operations(goal)
+    seen = set(previous["seen_keys"]) if previous is not None else set()
+    start = 0
+    for index in range(len(messages) - 1, -1, -1):
+        if _starts_goal_turn(messages[index]):
+            start = index + 1
+            break
+    calls: dict[str, dict[str, Any]] = {}
+    observed: set[str] = set()
+    has_tool_activity = False
+    for message in messages[start:]:
+        kind = _kind(message)
+        if kind == "ai":
+            for call in _field(message, "tool_calls", []) or []:
+                if isinstance(call, dict) and isinstance(call.get("id"), str) and call["id"]:
+                    calls[call["id"]] = call
+                    has_tool_activity = True
+        elif kind == "tool":
+            has_tool_activity = True
+            call = calls.pop(_field(message, "tool_call_id"), None)
+            if call is not None:
+                key = _successful_operation_key(call, message)
+                if key is not None:
+                    observed.add(key)
+    return {
+        "version": 1,
+        "seen_keys": sorted(seen | observed),
+        "has_tool_activity": has_tool_activity,
+        "new_keys": len(observed - seen),
+    }

backend/packages/harness/deerflow/runtime/runs/worker.py

@@ -90,6 +90,7 @@ from deerflow.runtime.goal import (
     visible_conversation_signature,
     write_thread_goal,
 )
+from deerflow.runtime.goal_progress import observe_goal_operations
 from deerflow.runtime.keyed_lock import AsyncKeyedLockTable
 from deerflow.runtime.runs.stream_cleanup import close_agent_stream
 from deerflow.runtime.serialization import serialize
@@ -2483,6 +2484,7 @@ async def _persist_goal_evaluation(
     continuation_count: int | None = None,
     stand_down_reason: str | None = None,
     evidence_signature: str = "",
+    operation_progress: dict[str, Any] | None = None,
 ) -> GoalState | None:
     try:
         async with goal_thread_lock(thread_id):
@@ -2512,6 +2514,7 @@ async def _persist_goal_evaluation(
                 no_progress_count=no_progress_count,
                 stand_down_reason=stand_down_reason,
                 evidence_signature=evidence_signature,
+                operation_progress=operation_progress,
             )
             values = await write_thread_goal(
                 checkpointer,
@@ -2604,6 +2607,7 @@ async def _prepare_goal_continuation_input(
             continuation_count=continuation_count,
             stand_down_reason=stand_down_reason,
             evidence_signature=evidence_signature,
+            operation_progress=operation_progress,
         )
         if updated is not None and verdict_callback is not None:
             verdict_callback(updated, evaluation, stand_down_reason)
@@ -2622,6 +2626,7 @@ async def _prepare_goal_continuation_input(
         messages = await _materialized_checkpoint_messages(accessor, thread_id)
         conversation_signature_before = visible_conversation_signature(messages)
         evidence_signature = latest_visible_assistant_signature(messages)
+        operation_progress = observe_goal_operations(goal, messages)
 
         if _ends_on_human_input_request(messages):
             # The agent asked the user something. Continuing would tell it to keep
@@ -2632,7 +2637,7 @@ async def _prepare_goal_continuation_input(
                 reason="The turn ended on a question to the user that has not been answered.",
                 evidence_summary="",
             )
-            no_progress_count = compute_no_progress_count(goal, evaluation, evidence_signature=evidence_signature)
+            no_progress_count = compute_no_progress_count(goal, evaluation, evidence_signature=evidence_signature, operation_progress=operation_progress)
             await _persist(goal, evaluation, no_progress_count, stand_down_reason=_stand_down_reason(goal, evaluation, no_progress_count))
             return None
 
@@ -2643,7 +2648,7 @@ async def _prepare_goal_continuation_input(
                 reason="No durable assistant end-of-turn receipt was available.",
                 evidence_summary="",
             )
-            no_progress_count = compute_no_progress_count(goal, evaluation, evidence_signature=evidence_signature)
+            no_progress_count = compute_no_progress_count(goal, evaluation, evidence_signature=evidence_signature, operation_progress=operation_progress)
             await _persist(goal, evaluation, no_progress_count, stand_down_reason="no_durable_end_of_turn")
             return None
 
@@ -2682,13 +2687,13 @@ async def _prepare_goal_continuation_input(
             reason=f"The goal evaluator did not return a verdict ({type(exc).__name__}).",
             evidence_summary="",
         )
-        no_progress_count = compute_no_progress_count(goal, evaluation, evidence_signature=evidence_signature)
+        no_progress_count = compute_no_progress_count(goal, evaluation, evidence_signature=evidence_signature, operation_progress=operation_progress)
         await _persist(goal, evaluation, no_progress_count, stand_down_reason="evaluator_failed")
         return None
     if abort_event is not None and abort_event.is_set():
         return None
 
-    no_progress_count = compute_no_progress_count(goal, evaluation, evidence_signature=evidence_signature)
+    no_progress_count = compute_no_progress_count(goal, evaluation, evidence_signature=evidence_signature, operation_progress=operation_progress)
 
     # Re-check that neither the goal nor the visible conversation changed while the
     # evaluator ran — a user message or /goal clear racing the evaluation must win.
@@ -2735,7 +2740,7 @@ async def _prepare_goal_continuation_input(
             reason=f"The goal continuation could not be redacted ({type(exc).__name__}).",
             evidence_summary="",
         )
-        await _persist(goal, evaluation, compute_no_progress_count(goal, evaluation, evidence_signature=evidence_signature), stand_down_reason="evaluator_failed")
+        await _persist(goal, evaluation, compute_no_progress_count(goal, evaluation, evidence_signature=evidence_signature, operation_progress=operation_progress), stand_down_reason="evaluator_failed")
         return None
 
     next_count = int(goal.get("continuation_count", 0)) + 1

backend/packages/harness/deerflow/subagents/executor.py

@@ -5,7 +5,6 @@ import atexit
 import json
 import logging
 import os
-import re
 import sys
 import threading
 import uuid
@@ -62,6 +61,7 @@ from deerflow.tracing import (
     build_tracing_callbacks,
     inject_langfuse_metadata,
 )
+from deerflow.utils.bash_exit import BASH_TOOL_NAMES, bash_exit_status
 from deerflow.utils.messages import message_content_to_text
 
 if TYPE_CHECKING:
@@ -471,23 +471,13 @@ def _harvest_tool_receipts(
 
 
 #: Bash-family tool names whose calls count as recorded command executions.
-_BASH_EVIDENCE_TOOL_NAMES = frozenset({"bash", "bash_tool"})
+_BASH_EVIDENCE_TOOL_NAMES = BASH_TOOL_NAMES
 #: Bounds for the harvested evidence: only the newest few executions travel,
 #: with command/output text capped (test summaries print at the tail).
 _BASH_EVIDENCE_MAX_ENTRIES = 20
 _BASH_EVIDENCE_COMMAND_CHARS = 500
 _BASH_EVIDENCE_OUTPUT_TAIL_CHARS = 1000
 
-#: Exit-status markers in bash *output text*: a nonzero exit does not raise —
-#: local sandboxes append ``Exit Code: N``; e2b/opensandbox emit
-#: ``Command exited with code N`` when the command produced no output.
-_BASH_EXIT_CODE_MARKER_RE = re.compile(r"Exit Code: (-?\d+)\s*$")
-#: Remote providers emit ``Command exited with code N`` ONLY as the complete
-#: output of a silent command — anchor it to the whole (trimmed) content so
-#: a successful command that merely prints the phrase while exercising an
-#: error path is not misrecorded as failed.
-_BASH_EXITED_WITH_CODE_RE = re.compile(r"Command exited with code (-?\d+)")
-
 
 def _bash_evidence_status(content: str, meta_status: str) -> tuple[str, str | None]:
     """Derive the recorded status from the shell exit marker when present.
@@ -495,16 +485,10 @@ def _bash_evidence_status(content: str, meta_status: str) -> tuple[str, str | No
     Returns ``(status, marker)``: the marker text actually seen (e.g. ``Exit
     Code: 5``), so consumers can report it instead of asserting a failure the
     harness cannot distinguish from the command's own trailing text. The
-    explicit marker is authoritative: ``deerflow_tool_meta`` reports the
-    generic ToolMessage status, which stays ``success`` for a nonzero exit
-    rendered as ordinary output text.
+    explicit marker takes precedence over generic metadata in old checkpoints
+    or provider stamps that report ``success`` for a nonzero shell exit.
     """
-    match = _BASH_EXIT_CODE_MARKER_RE.search(content) or _BASH_EXITED_WITH_CODE_RE.fullmatch(content.strip())
-    if match is None:
-        return meta_status, None
-    # Signal-killed local subprocesses report signed codes (Exit Code: -9);
-    # only an exact zero is a success.
-    return ("success" if int(match.group(1)) == 0 else "error"), " ".join(match.group(0).split())
+    return bash_exit_status(content) or (meta_status, None)
 
 
 def _harvest_shell_persistence(final_state: Any) -> bool | None:

backend/packages/harness/deerflow/utils/bash_exit.py

@@ -0,0 +1,26 @@
+"""Exit markers in the sandbox bash result protocol, without executing commands."""
+
+import re
+from typing import Literal
+
+BASH_TOOL_NAMES = frozenset({"bash", "bash_tool"})
+
+# Local/providers append a final line; silent remote failures use the whole body.
+# Do not treat a phrase embedded in ordinary command output as an exit marker.
+_EXIT_CODE_TAIL = re.compile(r"(?:^|\n)(Exit Code: (-?[0-9]+))\s*\Z")
+_SILENT_EXIT = re.compile(r"(Command exited with code (-?[0-9]+))")
+
+
+def bash_exit_status(content: str) -> tuple[Literal["success", "error"], str] | None:
+    """Return reported execution status and the exact marker, or no marker.
+
+    This is execution evidence, not task acceptance. A command's own trailing
+    text can mimic the protocol, so retain the marker for evidence consumers.
+    """
+    match = _EXIT_CODE_TAIL.search(content) or _SILENT_EXIT.fullmatch(content.strip())
+    if match is None:
+        return None
+    # String comparison also handles signed zero without converting unbounded
+    # command-supplied digits through int(). Signals and timeout 124 are errors.
+    status = "error" if match.group(2).lstrip("-0") else "success"
+    return status, match.group(1)

backend/tests/test_goal_operation_progress.py

@@ -0,0 +1,281 @@
+import copy
+import json
+
+import pytest
+from langchain_core.messages import AIMessage, HumanMessage, ToolMessage
+
+from deerflow.runtime.goal import GoalEvaluation, attach_goal_evaluation, build_goal_state, compute_no_progress_count, latest_visible_assistant_signature
+from deerflow.runtime.goal_progress import observe_goal_operations
+
+EVALUATION = GoalEvaluation(satisfied=False, blocker="goal_not_met_yet", reason="Work remains.", evidence_summary="")
+
+
+def operation(index=1, *, call_id=None, status="success", content="OK", receipt=False, name="write_file", args=None):
+    call_id = call_id or f"call-{index}"
+    call = {"name": name, "args": args if args is not None else {"path": f"/tmp/part-{index}.txt", "content": f"Part {index}"}, "id": call_id}
+    result = ToolMessage(content=content, tool_call_id=call_id, status=status)
+    if receipt:
+        from deerflow.agents.middlewares.tool_receipt import make_tool_receipt
+
+        result.additional_kwargs["deerflow_tool_receipt"] = make_tool_receipt(call, result)
+    return [AIMessage(content="", tool_calls=[call]), result]
+
+
+def turn(*operations, text="More work remains."):
+    return [HumanMessage(content="Continue", additional_kwargs={"hide_from_ui": True, "deerflow_goal_continuation": True}), *[message for pair in operations for message in pair], AIMessage(content=text)]
+
+
+def evaluate(goal, messages):
+    signature = latest_visible_assistant_signature(messages)
+    observation = observe_goal_operations(goal, messages)
+    count = compute_no_progress_count(goal, EVALUATION, evidence_signature=signature, operation_progress=observation)
+    return attach_goal_evaluation(goal, EVALUATION, run_id="offline", no_progress_count=count, evidence_signature=signature, operation_progress=observation)
+
+
+def test_same_reply_with_new_success_resets_count():
+    goal = evaluate(build_goal_state("Write the parts"), turn(operation()))
+    goal = evaluate(goal, turn(operation(call_id="repeat")))
+    assert goal["no_progress_count"] == 1
+    assert evaluate(goal, turn(operation(2)))["no_progress_count"] == 0
+
+
+def test_rephrased_reply_repeated_operation_does_not_reset():
+    goal = evaluate(build_goal_state("Write the parts"), turn(operation()))
+    assert evaluate(goal, turn(operation(call_id="new-id"), text="Making steady progress"))["no_progress_count"] == 1
+
+
+def test_alternating_subsets_cannot_reuse_old_work():
+    goal = evaluate(build_goal_state("Write the parts"), turn(operation()))
+    goal = evaluate(goal, turn(operation(), operation(2)))
+    assert goal["no_progress_count"] == 0
+    goal = evaluate(goal, turn(operation()))
+    assert goal["no_progress_count"] == 1
+    assert evaluate(goal, turn(operation(), operation(2)))["no_progress_count"] == 2
+
+
+@pytest.mark.parametrize("count", [65, 1000])
+def test_exact_history_has_no_64_operation_cutoff(count):
+    goal = evaluate(build_goal_state("Write the parts"), turn(*(operation(index) for index in range(count))))
+    assert len(goal["last_evaluation"]["operation_progress"]["seen_keys"]) == count
+    assert "text" not in goal["last_evaluation"]["operation_progress"]
+    # The previous PR revision wrote this unused field; it remains readable.
+    goal["last_evaluation"]["operation_progress"]["text"] = "old-format-field"
+    goal = json.loads(json.dumps(goal))
+    assert evaluate(goal, turn(operation(count - 1, call_id="restart")))["no_progress_count"] == 1
+    assert evaluate(goal, turn(operation(count)))["no_progress_count"] == 0
+
+
+@pytest.mark.parametrize("body,status,kwargs", [("Error: missing file", "success", {}), ("varied error", "error", {}), ("partial result", "success", {"deerflow_tool_meta": {"status": "partial_success"}})])
+def test_failures_do_not_gain_progress_from_new_arguments(body, status, kwargs):
+    goal = evaluate(build_goal_state("Write the parts"), turn(operation()))
+    failed = operation(2, status=status, content=body)
+    failed[1].additional_kwargs.update(kwargs)
+    assert evaluate(goal, turn(failed, text="Trying another way"))["no_progress_count"] == 1
+
+
+def test_unpaired_tool_result_cannot_reset_count():
+    goal = evaluate(build_goal_state("Write the parts"), turn(operation()))
+    assert evaluate(goal, turn([ToolMessage(content="OK", tool_call_id="orphan")], text="Trying again"))["no_progress_count"] == 1
+
+
+def test_latest_hidden_human_message_bounds_current_interval():
+    goal = evaluate(build_goal_state("Write the parts"), turn(operation()))
+    messages = turn(operation(2)) + turn(operation())
+    assert evaluate(goal, messages)["no_progress_count"] == 1
+    assert len(evaluate(goal, messages)["last_evaluation"]["operation_progress"]["seen_keys"]) == 1
+
+
+@pytest.mark.parametrize("receipt", [False, True])
+def test_receipt_setting_and_ids_do_not_change_identity_or_mutate_messages(receipt):
+    goal = evaluate(build_goal_state("Write the parts"), turn(operation(receipt=receipt)))
+    messages = turn(operation(call_id="different", receipt=not receipt))
+    original = copy.deepcopy(messages)
+    assert evaluate(goal, messages)["no_progress_count"] == 1
+    assert messages == original
+
+
+def test_pure_text_preserves_native_behavior_after_tool_turn():
+    goal = evaluate(build_goal_state("Write the parts"), turn(operation()))
+    for text in ["More work remains.", "A changed reply"]:
+        messages = turn(text=text)
+        expected = compute_no_progress_count(goal, EVALUATION, evidence_signature=latest_visible_assistant_signature(messages))
+        assert evaluate(goal, messages)["no_progress_count"] == expected
+
+
+def test_tool_failure_cannot_be_overridden_by_stale_success_metadata():
+    goal = evaluate(build_goal_state("Write the parts"), turn(operation()))
+    failed = operation(2, status="error")
+    failed[1].additional_kwargs["deerflow_tool_meta"] = {"status": "success"}
+    assert evaluate(goal, turn(failed))["no_progress_count"] == 1
+
+
+def test_persistence_merges_operations_from_fresh_goal_state():
+    goal = evaluate(build_goal_state("Write the parts"), turn(operation()))
+    old_observation = observe_goal_operations(goal, turn(operation(2)))
+    current = evaluate(goal, turn(operation(3)))
+    result = attach_goal_evaluation(current, EVALUATION, run_id="racing-evaluation", operation_progress=old_observation)
+    assert len(result["last_evaluation"]["operation_progress"]["seen_keys"]) == 3
+
+
+def test_serialized_message_dicts_have_same_operation_identity():
+    messages = turn(operation())
+    goal = evaluate(build_goal_state("Write the parts"), messages)
+    assert evaluate(goal, [message.model_dump() for message in messages])["no_progress_count"] == 1
+
+
+def test_matching_failed_receipt_cannot_gain_progress():
+    goal = evaluate(build_goal_state("Write the parts"), turn(operation()))
+    failed = operation(2, receipt=True)
+    failed[1].additional_kwargs["deerflow_tool_receipt"]["status"] = "error"
+    assert evaluate(goal, turn(failed))["no_progress_count"] == 1
+
+
+def test_foreign_receipt_is_not_used_to_classify_current_result():
+    goal = evaluate(build_goal_state("Write the parts"), turn(operation()))
+    current = operation(2, receipt=True)
+    current[1].additional_kwargs["deerflow_tool_receipt"].update(status="error", tool_call_id="different")
+    assert evaluate(goal, turn(current))["no_progress_count"] == 0
+
+
+@pytest.mark.parametrize(
+    "name,args",
+    [
+        ("bash", {"command": "printf OK"}),
+        ("read_file", {"path": "/tmp/part.txt"}),
+        ("write_file", {"path": "/tmp/part.txt", "content": "Part"}),
+        ("str_replace", {"path": "/tmp/part.txt", "old_str": "Part", "new_str": "Revised"}),
+        ("ls", {"path": "/tmp"}),
+        ("glob", {"pattern": "*.txt"}),
+        ("grep", {"pattern": "Part", "path": "/tmp"}),
+    ],
+)
+def test_sandbox_ui_description_does_not_change_operation_identity(name, args):
+    first = operation(name=name, args={**args, "description": "Inspect the source"}, receipt=True)
+    repeated = operation(name=name, args={**args, "description": "Inspect it again"}, call_id="another-call", receipt=True)
+    assert first[1].additional_kwargs["deerflow_tool_receipt"]["args_sha256"] != repeated[1].additional_kwargs["deerflow_tool_receipt"]["args_sha256"]
+    original = copy.deepcopy(repeated)
+    goal = evaluate(build_goal_state("Finish the task"), turn(first))
+    assert evaluate(goal, turn(repeated))["no_progress_count"] == 1
+    assert repeated == original
+
+
+@pytest.mark.parametrize(
+    "name,args,changed",
+    [
+        ("read_file", {"path": "/tmp/part.txt", "start_line": 1}, {"start_line": 2}),
+        ("write_file", {"path": "/tmp/part.txt", "content": "Part"}, {"content": "Revised"}),
+        ("str_replace", {"path": "/tmp/part.txt", "old_str": "Part", "new_str": "First"}, {"new_str": "Second"}),
+        ("update_agent", {"description": "Summarize documents"}, {"description": "Summarize documents and code"}),
+        ("create_issue", {"description": "First issue body"}, {"description": "A different issue body"}),
+    ],
+)
+def test_functional_arguments_still_distinguish_operations(name, args, changed):
+    first = operation(name=name, args=args, receipt=True)
+    different = operation(name=name, args={**args, **changed}, call_id="different-call", receipt=True)
+    goal = evaluate(build_goal_state("Finish the task"), turn(first))
+    goal = evaluate(goal, turn(first))
+    assert goal["no_progress_count"] == 1
+    assert evaluate(goal, turn(different))["no_progress_count"] == 0
+
+
+def test_legacy_fetched_error_page_reuses_existing_tool_classification():
+    goal = evaluate(build_goal_state("Read the sources"), turn(operation()))
+    call = {"name": "web_fetch", "args": {"url": "https://example.invalid/missing"}, "id": "fetch"}
+    messages = turn([AIMessage(content="", tool_calls=[call]), ToolMessage(content="# 404 Not Found\nnginx", tool_call_id="fetch")])
+    assert evaluate(goal, messages)["no_progress_count"] == 1
+
+
+@pytest.mark.parametrize("serialized", [False, True])
+def test_native_todo_reminder_cannot_hide_successful_operations(serialized):
+    from langgraph.runtime import Runtime
+
+    from deerflow.agents.middlewares.todo_middleware import TodoMiddleware
+
+    goal = evaluate(build_goal_state("Write the parts"), turn(operation()))
+    goal = evaluate(goal, turn(operation()))
+    assert goal["no_progress_count"] == 1
+    messages = turn(operation(2))[:-1]
+    reminder_update = TodoMiddleware().before_model({"messages": messages, "todos": [{"content": "Write the next part", "status": "pending"}]}, Runtime())
+    assert reminder_update is not None
+    messages.extend(reminder_update["messages"])
+    assert messages[-1].name == "todo_reminder"
+    messages.append(AIMessage(content="More work remains."))
+    if serialized:
+        messages = [message.model_dump() for message in messages]
+    assert evaluate(goal, messages)["no_progress_count"] == 0
+
+
+@pytest.mark.parametrize("serialized", [False, True])
+def test_valid_hidden_card_reply_starts_a_new_interval(serialized):
+    goal = evaluate(build_goal_state("Write the parts"), turn(operation()))
+    answer = HumanMessage(
+        content="Use option A",
+        additional_kwargs={
+            "hide_from_ui": True,
+            "human_input_response": {"version": 1, "kind": "human_input_response", "source": "ask_clarification", "request_id": "request-1", "response_kind": "text", "value": "Use option A"},
+        },
+    )
+    messages = [*turn(operation(2))[:-1], answer, *operation(), AIMessage(content="More work remains.")]
+    if serialized:
+        messages = [message.model_dump() for message in messages]
+    assert evaluate(goal, messages)["no_progress_count"] == 1
+
+
+def _externalized_bash_operation(tmp_path, *, call_id, receipts=True, content=None, description="Inspect the file"):
+    from deerflow.agents.middlewares.tool_output_budget_middleware import _patch_tool_message
+    from deerflow.agents.middlewares.tool_receipt import make_tool_receipt
+    from deerflow.agents.middlewares.tool_result_meta import normalize_tool_message
+    from deerflow.config.tool_output_config import ToolOutputConfig
+
+    call = {"name": "bash", "args": {"command": "cat /tmp/large-input.txt", "description": description}, "id": call_id}
+    raw = normalize_tool_message(ToolMessage(content=content or ("same data\n" * 1400 + "\nExit Code: 0"), tool_call_id=call_id, name="bash"))
+    if receipts:
+        raw.additional_kwargs["deerflow_tool_receipt"] = make_tool_receipt(call, raw)
+    visible = _patch_tool_message(raw, ToolOutputConfig(), str(tmp_path))
+    assert visible.content != raw.content
+    return [AIMessage(content="", tool_calls=[call]), visible]
+
+
+@pytest.mark.parametrize("receipts", [False, True])
+def test_externalized_output_path_and_call_id_cannot_manufacture_progress(tmp_path, monkeypatch, receipts):
+    from deerflow.agents.middlewares import tool_output_budget_middleware
+
+    monkeypatch.setattr(tool_output_budget_middleware, "get_blob_store_if_enabled", lambda: None)
+    first = _externalized_bash_operation(tmp_path, call_id="first", receipts=receipts)
+    repeat = _externalized_bash_operation(tmp_path, call_id="repeat", receipts=receipts, description="Check it again")
+    assert first[1].content != repeat[1].content
+    goal = evaluate(build_goal_state("Read the source"), turn(first))
+    goal = evaluate(goal, turn(repeat, text="Rephrased progress"))
+    assert goal["no_progress_count"] == 1
+    assert len(goal["last_evaluation"]["operation_progress"]["seen_keys"]) == (1 if receipts else 0)
+
+
+def test_changed_raw_externalized_result_is_new_progress_without_reading_saved_file(tmp_path, monkeypatch):
+    from pathlib import Path
+
+    from deerflow.agents.middlewares import tool_output_budget_middleware
+
+    monkeypatch.setattr(tool_output_budget_middleware, "get_blob_store_if_enabled", lambda: None)
+    first = _externalized_bash_operation(tmp_path, call_id="first")
+    changed = _externalized_bash_operation(tmp_path, call_id="changed", content="different data\n" * 1400 + "\nExit Code: 0", description="Read the changed file")
+
+    def denied(*args, **kwargs):
+        raise AssertionError("Goal observation must not read externalized tool files")
+
+    monkeypatch.setattr(Path, "read_bytes", denied)
+    monkeypatch.setattr(Path, "read_text", denied)
+    goal = evaluate(build_goal_state("Read the source"), turn(first))
+    goal = evaluate(goal, turn(first))
+    assert goal["no_progress_count"] == 1
+    assert evaluate(goal, turn(changed))["no_progress_count"] == 0
+
+
+def test_transformed_receipt_must_still_match_raw_arguments(tmp_path, monkeypatch):
+    from deerflow.agents.middlewares import tool_output_budget_middleware
+
+    monkeypatch.setattr(tool_output_budget_middleware, "get_blob_store_if_enabled", lambda: None)
+    first = _externalized_bash_operation(tmp_path, call_id="first")
+    mismatched = _externalized_bash_operation(tmp_path, call_id="changed", content="changed result\n" * 1400)
+    mismatched[0].tool_calls[0]["args"]["description"] = "Not the recorded invocation"
+    goal = evaluate(build_goal_state("Read the source"), turn(first))
+    assert evaluate(goal, turn(mismatched))["no_progress_count"] == 1

backend/tests/test_goal_operation_progress_worker.py

@@ -0,0 +1,396 @@
+"""Goal operation observations at the real durable worker decision boundary.
+
+Only completion verdicts are scripted. Checkpoint writes, continuation admission,
+progress counting, and all existing worker guards execute production code.
+"""
+
+from __future__ import annotations
+
+import os
+import socket
+from types import SimpleNamespace
+
+import pytest
+from langchain_core.language_models.fake_chat_models import FakeListChatModel
+from langchain_core.messages import AIMessage, HumanMessage, RemoveMessage, ToolMessage
+from langgraph.checkpoint.memory import InMemorySaver
+from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver
+from langgraph.graph import END, START, StateGraph
+from langgraph.graph.message import REMOVE_ALL_MESSAGES
+from langgraph.prebuilt import ToolNode
+
+from deerflow.agents.middlewares.summarization_middleware import DeerFlowSummarizationMiddleware
+from deerflow.agents.middlewares.tool_error_handling_middleware import ToolErrorHandlingMiddleware
+from deerflow.agents.middlewares.tool_receipt_middleware import ToolReceiptMiddleware
+from deerflow.agents.thread_state import ThreadState
+from deerflow.runtime.checkpoint_state import CheckpointStateAccessor, build_state_mutation_graph
+from deerflow.runtime.goal import (
+    GoalEvaluation,
+    attach_goal_evaluation,
+    build_goal_state,
+    latest_visible_assistant_signature,
+    read_thread_goal,
+    write_thread_goal,
+)
+from deerflow.runtime.runs import worker
+from deerflow.sandbox.local.local_sandbox import LocalSandbox, PathMapping
+from deerflow.sandbox.tools import bash_tool, read_file_tool, write_file_tool
+
+
+class _Bridge:
+    async def publish(self, *_args):
+        pass
+
+
+def _config(thread_id):
+    return {"configurable": {"thread_id": thread_id, "checkpoint_ns": ""}}
+
+
+def _accessor(saver, mode):
+    graph = build_state_mutation_graph("goal_evaluator", mode, snapshot_frequency=1000)
+    return CheckpointStateAccessor.bind(graph, saver, mode=mode)
+
+
+def _operations(indices, *, prefix="call", text="Work remains", status="success"):
+    messages = []
+    for index in indices:
+        call = {"name": "read_file", "id": f"{prefix}-{index}", "args": {"path": f"/part-{index}.txt"}}
+        messages.extend(
+            [
+                AIMessage(content="", tool_calls=[call], id=f"ai-{prefix}-{index}"),
+                ToolMessage(
+                    content=f"Observed part {index}",
+                    tool_call_id=call["id"],
+                    name=call["name"],
+                    status=status,
+                    id=f"tool-{prefix}-{index}",
+                ),
+            ]
+        )
+    messages.append(AIMessage(content=text, id=f"final-{prefix}"))
+    return messages
+
+
+@pytest.fixture(autouse=True)
+def _no_models_or_network(monkeypatch):
+    def denied(*_args, **_kwargs):
+        raise AssertionError("Goal worker regression attempted a provider/network call")
+
+    from deerflow import models
+    from deerflow.runtime import goal
+
+    monkeypatch.setattr(socket.socket, "connect", denied)
+    monkeypatch.setattr(socket.socket, "connect_ex", denied)
+    monkeypatch.setattr(models, "create_chat_model", denied)
+    monkeypatch.setattr(goal, "create_chat_model", denied)
+
+    async def unresolved(_goal, _messages, **_kwargs):
+        return GoalEvaluation(satisfied=False, blocker="goal_not_met_yet", reason="Fixed unresolved control", evidence_summary="")
+
+    monkeypatch.setattr(worker, "evaluate_goal_completion", unresolved)
+
+
+async def _seed(saver, mode, thread_id, messages):
+    accessor = _accessor(saver, mode)
+    await accessor.aupdate(_config(thread_id), {"messages": [HumanMessage(content="Finish all parts", id="user"), *messages]}, as_node="goal_evaluator")
+    await write_thread_goal(saver, thread_id, build_goal_state("Finish all parts"))
+    return accessor
+
+
+async def _decide(saver, accessor, thread_id, **kwargs):
+    result = await worker._prepare_goal_continuation_input(
+        bridge=_Bridge(),
+        accessor=accessor,
+        checkpointer=saver,
+        thread_id=thread_id,
+        run_id="local-control",
+        model_name=None,
+        app_config=None,
+        **kwargs,
+    )
+    return result, await read_thread_goal(saver, thread_id)
+
+
+async def _append_turn(accessor, thread_id, messages, *, turn):
+    await accessor.aupdate(_config(thread_id), {"messages": [HumanMessage(content="Continue", id=f"hidden-{turn}", additional_kwargs={"hide_from_ui": True, "deerflow_goal_continuation": True}), *messages]}, as_node="goal_evaluator")
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("mode", ["full", "delta"])
+async def test_worker_keeps_exact_seen_operations_above_64_without_fallback(mode):
+    count = 65
+    saver = InMemorySaver()
+    thread_id = f"over-64-{mode}-{count}"
+    accessor = await _seed(saver, mode, thread_id, _operations(range(count)))
+    first, goal = await _decide(saver, accessor, thread_id)
+    assert first is not None and goal["no_progress_count"] == 0
+    keys = goal["last_evaluation"]["operation_progress"]["seen_keys"]
+    assert len(keys) == count
+    assert len(set(keys)) == count
+    # Rotating and shrinking previously observed sets must not buy continuations.
+    await _append_turn(accessor, thread_id, _operations(reversed(range(count)), prefix="rotated", text="Another wording"), turn=2)
+    second, goal = await _decide(saver, accessor, thread_id)
+    assert second is not None and goal["no_progress_count"] == 1
+    assert goal["last_evaluation"]["operation_progress"]["seen_keys"] == keys
+    await _append_turn(accessor, thread_id, _operations(range(1), prefix="subset", text="Yet another wording"), turn=3)
+    third, goal = await _decide(saver, accessor, thread_id)
+    assert third is None and goal["no_progress_count"] == 2
+    assert goal["last_evaluation"]["stand_down_reason"] == "no_progress_detected"
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("mode", ["full", "delta"])
+async def test_worker_novel_operation_beyond_64_resets_count(mode):
+    saver = InMemorySaver()
+    accessor = await _seed(saver, mode, "novel", _operations(range(65)))
+    await _decide(saver, accessor, "novel")
+    await _append_turn(accessor, "novel", _operations([0], prefix="repeat"), turn=2)
+    _, goal = await _decide(saver, accessor, "novel")
+    assert goal["no_progress_count"] == 1
+    await _append_turn(accessor, "novel", _operations([65], prefix="new"), turn=3)
+    continuation, goal = await _decide(saver, accessor, "novel")
+    assert continuation is not None and goal["no_progress_count"] == 0
+    assert len(goal["last_evaluation"]["operation_progress"]["seen_keys"]) == 66
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("mode", ["full", "delta"])
+async def test_worker_text_only_turns_preserve_original_counter(mode):
+    saver = InMemorySaver()
+    accessor = await _seed(saver, mode, "text", [AIMessage(content="First explanation", id="text-1")])
+    outcomes = [await _decide(saver, accessor, "text")]
+    await _append_turn(accessor, "text", [AIMessage(content="Different explanation", id="text-2")], turn=2)
+    outcomes.extend([await _decide(saver, accessor, "text") for _ in range(3)])
+    assert [r is not None for r, _ in outcomes] == [True, True, True, False]
+    assert [g["no_progress_count"] for _, g in outcomes] == [0, 0, 1, 2]
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("mode", ["full", "delta"])
+async def test_worker_sqlite_reopen_preserves_exact_history(tmp_path, mode):
+    db = tmp_path / "goal.sqlite"
+    for phase in range(3):
+        async with AsyncSqliteSaver.from_conn_string(str(db)) as saver:
+            await saver.setup()
+            if phase == 0:
+                accessor = await _seed(saver, mode, "restart", _operations(range(65)))
+            else:
+                accessor = _accessor(saver, mode)
+            result, goal = await _decide(saver, accessor, "restart")
+            assert (result is not None) == (phase < 2)
+            assert goal["no_progress_count"] == phase
+            assert len(goal["last_evaluation"]["operation_progress"]["seen_keys"]) == 65
+
+
+def _summary_middleware(*, keep):
+    return DeerFlowSummarizationMiddleware(
+        model=FakeListChatModel(responses=["Earlier operations summarized. Work remains."]),
+        trigger=("messages", 4),
+        keep=("messages", keep),
+        trim_tokens_to_summarize=2000,
+        token_counter=lambda messages: sum(len(message.content) for message in messages),
+        summary_prompt="{messages}",
+        anchor_model_name="primary",
+        run_model_name="primary",
+    )
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("mode", ["full", "delta"])
+async def test_worker_real_compaction_and_old_replay_do_not_reset_count(mode):
+    saver = InMemorySaver()
+    accessor = await _seed(saver, mode, "compact", _operations(range(65)))
+    await _decide(saver, accessor, "compact")
+    snapshot = await accessor.aget(_config("compact"))
+    middleware = _summary_middleware(keep=1)
+    compact = await middleware.acompact_state(dict(snapshot.values), SimpleNamespace(context={}), force=True, raise_on_failure=True)
+    assert compact is not None and middleware.summary_call_count > 0
+    await accessor.aupdate(_config("compact"), {"messages": [RemoveMessage(id=REMOVE_ALL_MESSAGES), *compact.preserved_messages], "summary_text": compact.summary_text}, as_node="goal_evaluator")
+    second, goal = await _decide(saver, accessor, "compact")
+    assert second is not None and goal["no_progress_count"] == 1
+    await _append_turn(accessor, "compact", _operations([0], prefix="replayed", text="Rephrased old work"), turn=3)
+    third, goal = await _decide(saver, accessor, "compact")
+    assert third is None and goal["no_progress_count"] == 2
+    assert len(goal["last_evaluation"]["operation_progress"]["seen_keys"]) == 65
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("mode", ["full", "delta"])
+async def test_worker_first_turn_compaction_counts_retained_successful_pairs(mode):
+    saver = InMemorySaver()
+    prefix = [AIMessage(content="Earlier context " * 50, id=f"prefix-{index}") for index in range(20)]
+    accessor = await _seed(saver, mode, "first-compact", [*prefix, *_operations(range(65))])
+    snapshot = await accessor.aget(_config("first-compact"))
+    middleware = _summary_middleware(keep=131)
+    compact = await middleware.acompact_state(dict(snapshot.values), SimpleNamespace(context={}), force=True, raise_on_failure=True)
+    assert compact is not None and middleware.summary_call_count > 0
+    assert sum(isinstance(message, ToolMessage) for message in compact.preserved_messages) == 65
+    await accessor.aupdate(_config("first-compact"), {"messages": [RemoveMessage(id=REMOVE_ALL_MESSAGES), *compact.preserved_messages], "summary_text": compact.summary_text}, as_node="goal_evaluator")
+    continuation, goal = await _decide(saver, accessor, "first-compact")
+    assert continuation is not None
+    assert len(goal["last_evaluation"]["operation_progress"]["seen_keys"]) == 65
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("mode", ["full", "delta"])
+@pytest.mark.parametrize("with_tool", [False, True])
+async def test_worker_legacy_changed_reply_then_repeat_keeps_native_counter(mode, with_tool):
+    saver = InMemorySaver()
+    thread_id = f"legacy-changed-{mode}-{with_tool}"
+    previous_messages = [AIMessage(content="Earlier reply", id="legacy-final")]
+    accessor = await _seed(saver, mode, thread_id, previous_messages)
+    goal = await read_thread_goal(saver, thread_id)
+    evaluation = GoalEvaluation(satisfied=False, blocker="goal_not_met_yet", reason="Work remains", evidence_summary="")
+    legacy = attach_goal_evaluation(goal, evaluation, run_id="old-version", no_progress_count=1, evidence_signature=latest_visible_assistant_signature(previous_messages))
+    await write_thread_goal(saver, thread_id, legacy)
+    for turn, expected_count in [(1, 0), (2, 1), (3, 2)]:
+        messages = _operations([1], prefix=f"legacy-changed-{turn}", text="A genuinely changed reply") if with_tool else [AIMessage(content="A genuinely changed reply", id=f"changed-{turn}")]
+        await _append_turn(accessor, thread_id, messages, turn=turn)
+        continuation, current = await _decide(saver, accessor, thread_id)
+        assert current["no_progress_count"] == expected_count
+        assert (continuation is not None) is (expected_count < 2)
+
+
+@pytest.mark.asyncio
+async def test_replacing_or_clearing_goal_drops_its_operation_history():
+    saver = InMemorySaver()
+    accessor = await _seed(saver, "full", "replacement", _operations(range(65)))
+    await _decide(saver, accessor, "replacement")
+    await write_thread_goal(saver, "replacement", None)
+    assert await read_thread_goal(saver, "replacement") is None
+    await write_thread_goal(saver, "replacement", build_goal_state("A new goal"))
+    await _append_turn(accessor, "replacement", _operations([0], prefix="replacement"), turn=2)
+    continuation, goal = await _decide(saver, accessor, "replacement")
+    assert continuation is not None and goal["no_progress_count"] == 0
+    assert len(goal["last_evaluation"]["operation_progress"]["seen_keys"]) == 1
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("mode", ["full", "delta"])
+async def test_worker_existing_goal_migration_does_not_reset_count(mode):
+    saver = InMemorySaver()
+    messages = _operations([0])
+    accessor = await _seed(saver, mode, "migration", messages)
+    goal = await read_thread_goal(saver, "migration")
+    evaluation = GoalEvaluation(satisfied=False, blocker="goal_not_met_yet", reason="Work remains", evidence_summary="")
+    legacy = attach_goal_evaluation(goal, evaluation, run_id="old-version", no_progress_count=1, evidence_signature=latest_visible_assistant_signature(messages))
+    legacy["last_evaluation"].pop("operation_progress", None)
+    await write_thread_goal(saver, "migration", legacy)
+    result, goal = await _decide(saver, accessor, "migration")
+    assert result is None and goal["no_progress_count"] == 2
+    assert goal["last_evaluation"]["stand_down_reason"] == "no_progress_detected"
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("stop_reason", ["loop_capped", "token_capped"])
+async def test_new_successful_work_cannot_override_existing_hard_stop(stop_reason):
+    saver = InMemorySaver()
+    accessor = await _seed(saver, "full", "hard-stop", _operations(range(65)))
+    result, goal = await _decide(saver, accessor, "hard-stop", run_stop_reason=stop_reason)
+    assert result is None
+    assert goal["continuation_count"] == 0
+    assert goal["last_evaluation"]["stand_down_reason"] == stop_reason
+
+
+def _native_tool_graph(tmp_path, monkeypatch, *, receipts):
+    (tmp_path / "workspace").mkdir()
+    (tmp_path / "workspace/input.txt").write_text("Stable input contents.\n", encoding="utf-8")
+    sandbox = LocalSandbox("local", path_mappings=[PathMapping("/mnt/user-data/workspace", str(tmp_path / "workspace"))])
+    monkeypatch.setattr("deerflow.sandbox.local.local_sandbox.build_sandbox_env", lambda _injected: {"PATH": os.defpath})
+    # Bind only execution policy and the test-owned sandbox; tools, their real
+    # results, normalization, receipt stamping and worker decisions stay native.
+    monkeypatch.setattr("deerflow.sandbox.tools.ensure_sandbox_initialized", lambda runtime: sandbox)
+
+    async def initialized(_runtime):
+        return sandbox
+
+    monkeypatch.setattr("deerflow.sandbox.tools.ensure_sandbox_initialized_async", initialized)
+    monkeypatch.setattr("deerflow.sandbox.tools.is_host_bash_allowed", lambda: True)
+    errors = ToolErrorHandlingMiddleware()
+    receipt = ToolReceiptMiddleware() if receipts else None
+
+    async def wrap(request, handler):
+        async def normalized(current):
+            return await errors.awrap_tool_call(current, handler)
+
+        return await receipt.awrap_tool_call(request, normalized) if receipt else await normalized(request)
+
+    builder = StateGraph(ThreadState)
+    builder.add_node("tools", ToolNode([bash_tool, read_file_tool, write_file_tool], awrap_tool_call=wrap, handle_tool_errors=False))
+    builder.add_edge(START, "tools")
+    builder.add_edge("tools", END)
+    return builder.compile()
+
+
+async def _native_tool_turn(graph, tmp_path, *, thread_id, turn, kind):
+    args = (
+        {"command": f"env -i /bin/sh -c 'exit 2' # attempt-{turn}", "description": "Controlled failing command"}
+        if kind == "failed-bash"
+        else {"path": "/mnt/user-data/workspace/input.txt", "description": f"Read attempt {turn}"}
+        if kind == "same-read"
+        else {"path": "/mnt/user-data/workspace/result.txt", "content": "The same file contents.\n", "description": f"Write attempt {turn}"}
+    )
+    tool_name = {"failed-bash": "bash", "same-read": "read_file", "same-write": "write_file"}[kind]
+    state = await graph.ainvoke(
+        {
+            "sandbox": {"sandbox_id": "local"},
+            "thread_data": {f"{name}_path": str(tmp_path / name) for name in ("workspace", "uploads", "outputs")},
+            "messages": [AIMessage(content="", tool_calls=[{"id": f"native-{turn}", "name": tool_name, "args": args}])],
+        },
+        context={"thread_id": thread_id},
+    )
+    result = next(message for message in state["messages"] if isinstance(message, ToolMessage))
+    return [*state["messages"], AIMessage(content="Work remains.", id=f"native-final-{turn}")], result
+
+
+@pytest.mark.skipif(os.name == "nt", reason="Controlled native LocalSandbox commands use a POSIX shell")
+@pytest.mark.asyncio
+@pytest.mark.parametrize("mode", ["full", "delta"])
+@pytest.mark.parametrize("receipts", [False, True], ids=["receipts-off", "receipts-on"])
+@pytest.mark.parametrize("kind", ["failed-bash", "same-read"])
+async def test_native_tool_results_cannot_buy_goal_continuations(tmp_path, monkeypatch, mode, receipts, kind):
+    graph = _native_tool_graph(tmp_path, monkeypatch, receipts=receipts)
+    async with AsyncSqliteSaver.from_conn_string(str(tmp_path / "goal.sqlite")) as saver:
+        await saver.setup()
+        accessor = await _seed(saver, mode, "native-progress", [])
+        counts, decisions, results = [], [], []
+        for turn in range(3):
+            messages, result = await _native_tool_turn(graph, tmp_path, thread_id="native-progress", turn=turn, kind=kind)
+            assert ("deerflow_tool_receipt" in result.additional_kwargs) is receipts
+            if kind == "failed-bash":
+                assert str(result.content).endswith("Exit Code: 2")
+            else:
+                assert "Stable input contents." in str(result.content)
+            await _append_turn(accessor, "native-progress", messages, turn=turn)
+            continuation, goal = await _decide(saver, accessor, "native-progress")
+            counts.append(goal["no_progress_count"])
+            decisions.append(continuation is not None)
+            results.append(result)
+        assert counts == [0, 1, 2]
+        assert decisions == [True, True, False]
+        assert goal["last_evaluation"]["stand_down_reason"] == "no_progress_detected"
+        assert len(goal["last_evaluation"]["operation_progress"]["seen_keys"]) == (0 if kind == "failed-bash" else 1)
+        assert all(result.additional_kwargs["deerflow_tool_meta"]["status"] == ("error" if kind == "failed-bash" else "success") for result in results)
+        assert (tmp_path / "workspace/input.txt").read_text(encoding="utf-8") == "Stable input contents.\n"
+
+
+@pytest.mark.skipif(os.name == "nt", reason="Controlled native LocalSandbox commands use a POSIX shell")
+@pytest.mark.asyncio
+@pytest.mark.parametrize("mode", ["full", "delta"])
+@pytest.mark.parametrize("receipts", [False, True], ids=["receipts-off", "receipts-on"])
+async def test_native_success_after_failed_tools_resets_goal_stall(tmp_path, monkeypatch, mode, receipts):
+    graph = _native_tool_graph(tmp_path, monkeypatch, receipts=receipts)
+    async with AsyncSqliteSaver.from_conn_string(str(tmp_path / "goal.sqlite")) as saver:
+        await saver.setup()
+        accessor = await _seed(saver, mode, "native-recovery", [])
+        counts = []
+        for turn, kind in enumerate(["failed-bash", "failed-bash", "same-write"]):
+            messages, result = await _native_tool_turn(graph, tmp_path, thread_id="native-recovery", turn=turn, kind=kind)
+            assert result.additional_kwargs["deerflow_tool_meta"]["status"] == ("error" if kind == "failed-bash" else "success")
+            await _append_turn(accessor, "native-recovery", messages, turn=turn)
+            continuation, goal = await _decide(saver, accessor, "native-recovery")
+            assert continuation is not None
+            counts.append(goal["no_progress_count"])
+        assert counts == [0, 1, 0]
+        assert len(goal["last_evaluation"]["operation_progress"]["seen_keys"]) == 1
+        assert (tmp_path / "workspace/result.txt").read_text(encoding="utf-8") == "The same file contents.\n"

backend/tests/test_goal_worker.py

@@ -15,6 +15,41 @@ from deerflow.runtime.runs.manager import RunRecord, RunStartOutcome
 from deerflow.runtime.runs.schemas import DisconnectMode, RunStatus
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize("successful_work", [True, False], ids=["new-work-same-reply", "failed-work-paraphrased-reply"])
+async def test_goal_progress_uses_operation_results_instead_of_reply_wording(monkeypatch, successful_work):
+    saver = InMemorySaver()
+    thread_id = "operation-progress-behavior"
+    await _seed_goal_thread(saver, thread_id=thread_id, goal_text="Finish the controlled task")
+    goal = await read_thread_goal(saver, thread_id)
+    goal["max_continuations"] = 8
+    await write_thread_goal(saver, thread_id, goal)
+
+    async def fixed_evaluator(*_args, **_kwargs):
+        return GoalEvaluation(satisfied=False, blocker="goal_not_met_yet", reason="Work remains", evidence_summary="Fixed local result")
+
+    monkeypatch.setattr(worker, "evaluate_goal_completion", fixed_evaluator)
+    messages = [HumanMessage(content="Finish the controlled task")]
+    decisions = []
+    for turn in range(3):
+        call = {"id": f"call-{turn}", "name": "write_file" if successful_work else "read_file", "args": {"path": f"/outputs/part-{turn}.txt"}}
+        if turn:
+            messages.append(HumanMessage(content="Continue", additional_kwargs={"hide_from_ui": True, "deerflow_goal_continuation": True}))
+        messages.extend(
+            [
+                AIMessage(content="", tool_calls=[call]),
+                ToolMessage(content=f"Created part {turn}" if successful_work else "File not found", tool_call_id=call["id"], name=call["name"], status="success" if successful_work else "error"),
+                AIMessage(content="More work remains." if successful_work else f"Investigating another approach, attempt {turn}."),
+            ]
+        )
+        await _write_messages(saver, thread_id=thread_id, messages=messages)
+        result = await worker._prepare_goal_continuation_input(accessor=_full_accessor(saver), bridge=_CollectingBridge(), checkpointer=saver, thread_id=thread_id, run_id="controlled-run", model_name=None, app_config=None)
+        decisions.append(result is not None)
+    saved = await read_thread_goal(saver, thread_id)
+    assert decisions == ([True, True, True] if successful_work else [True, True, False])
+    assert saved["no_progress_count"] == (0 if successful_work else 2)
+
+
 def _full_accessor(checkpointer) -> CheckpointStateAccessor:
     """Bind a full-mode accessor over a state-only graph for materialized reads."""
     graph = build_state_mutation_graph("goal_evaluator", "full")

backend/tests/test_tool_result_meta.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import json
+import os
 
 import pytest
 from langchain_core.messages import ToolMessage
@@ -43,6 +44,85 @@ def test_existing_meta_is_preserved():
     assert result.additional_kwargs[TOOL_META_KEY] is existing
 
 
+@pytest.mark.parametrize("name", ["bash", "bash_tool"])
+@pytest.mark.parametrize("content", ["\nExit Code: 1", "12 passed, 1 error\nExit Code: 7", "Command exited with code 3", "partial output\nExit Code: -9", "Command timed out\nExit Code: 124"])
+def test_bash_nonzero_exit_marker_is_a_recoverable_error(name, content):
+    msg = ToolMessage(content=content, tool_call_id="bash-1", name=name)
+    normalize_tool_message(msg)
+    assert _meta(msg) == {
+        "status": "error",
+        "error_type": "unknown",
+        "recoverable_by_model": True,
+        "recommended_next_action": "try_alternative",
+        "source": "content_analysis",
+    }
+
+
+def test_bash_nonzero_exit_repairs_legacy_analysis_success_without_changing_producer_meta():
+    legacy = {"status": "success", "source": "content_analysis"}
+    msg = ToolMessage(content="\nExit Code: 1", tool_call_id="bash-1", name="bash", additional_kwargs={TOOL_META_KEY: legacy, "deerflow_tool_receipt": {"status": "success"}})
+    normalize_tool_message(msg)
+    assert _meta(msg)["status"] == "error"
+    assert legacy == {"status": "success", "source": "content_analysis"}
+    assert msg.additional_kwargs["deerflow_tool_receipt"]["status"] == "success"
+
+
+@pytest.mark.parametrize("source", ["tool_return", "exception", "progress_middleware", "custom"])
+def test_bash_nonzero_exit_preserves_explicit_producer_meta(source):
+    existing = {"status": "success", "source": source}
+    msg = ToolMessage(content="\nExit Code: 1", tool_call_id="bash-1", name="bash", additional_kwargs={TOOL_META_KEY: existing})
+    assert _meta(normalize_tool_message(msg)) is existing
+
+
+@pytest.mark.parametrize(
+    "name,content",
+    [
+        ("bash", "ok\nExit Code: 0"),
+        ("bash", "Command exited with code 0"),
+        ("bash", "Exit Code: 1\nordinary final output"),
+        ("bash", "validating error path: Command exited with code 3\n5 passed"),
+        ("bash", "A document says Exit Code: 1"),
+        ("read_file", "\nExit Code: 1"),
+        ("web_fetch", "Command exited with code 3"),
+    ],
+)
+def test_bash_exit_marker_does_not_classify_other_tool_content_or_prose(name, content):
+    msg = ToolMessage(content=content, tool_call_id="call-1", name=name)
+    assert _meta(normalize_tool_message(msg))["status"] == "success"
+
+
+def test_bash_exit_status_does_not_classify_arbitrary_stdout_as_an_auth_error():
+    msg = ToolMessage(content="Processed row 403\nExit Code: 1", tool_call_id="bash-1", name="bash")
+    assert _meta(normalize_tool_message(msg))["error_type"] == "unknown"
+
+
+@pytest.mark.skipif(os.name == "nt", reason="POSIX inert shell command")
+@pytest.mark.parametrize(
+    "command,output,status",
+    [
+        (": attempt-0; exit 1", "\nExit Code: 1", "error"),
+        ("printf 'x\\n' | grep z", "\nExit Code: 1", "error"),
+        ("printf 'x\\n' | grep z || true", "(no output)", "success"),
+    ],
+)
+def test_native_bash_exit_is_stamped_before_receipt(monkeypatch, command, output, status):
+    from types import SimpleNamespace
+
+    from deerflow.agents.middlewares.tool_error_handling_middleware import ToolErrorHandlingMiddleware
+    from deerflow.agents.middlewares.tool_receipt_middleware import ToolReceiptMiddleware
+    from deerflow.sandbox.local import local_sandbox
+
+    monkeypatch.setattr(local_sandbox, "build_sandbox_env", lambda _injected: {"PATH": os.defpath})
+    actual = local_sandbox.LocalSandbox("exit-status-probe").execute_command(command, timeout=5)
+    assert actual == output
+    request = SimpleNamespace(tool_call={"id": "bash-1", "name": "bash", "args": {"command": command}})
+    errors = ToolErrorHandlingMiddleware()
+    result = ToolReceiptMiddleware().wrap_tool_call(request, lambda req: errors.wrap_tool_call(req, lambda _: ToolMessage(content=actual, tool_call_id="bash-1", name="bash")))
+    assert _meta(result)["status"] == status
+    assert _meta(result)["recoverable_by_model"] is True
+    assert result.additional_kwargs["deerflow_tool_receipt"]["status"] == status
+
+
 # ---------------------------------------------------------------------------
 # Error prefix (tool_return path)