fix(builder): retain safe repair validation details in trace

api/services/dify_builder/agent/fix.py

@@ -449,7 +449,7 @@ def propose_repair(
     # surface to the human; a proposal the engine still refuses is not a fix.
     vetted = preflight.vet_intents(graph, intents, _ALLOWED_NODE_TYPES, evidence=diagnosis.condition_evidence)
     if vetted.rejections:
-        diagnostics.report("validation_refused", attempt=1)
+        diagnostics.report("validation_refused", attempt=1, validation_details=vetted.validation_details)
         reasons = "\n".join(vetted.rejections)
         retry_user = f"{user}\n\nYour previous intents were invalid:\n{reasons}\nReturn corrected intents."
         if redact_text is not None:
@@ -464,6 +464,8 @@ def propose_repair(
         if not intents:
             return _empty_repair(risk, graph, redact_text)
         vetted = preflight.vet_intents(graph, intents, _ALLOWED_NODE_TYPES, evidence=diagnosis.condition_evidence)
+        if vetted.rejections:
+            diagnostics.report("validation_refused", attempt=2, validation_details=vetted.validation_details)
     # Parity with Edit (``agent.edit._refuse_a_batch_already_judged_wrong``), and
     # written as its own check rather than left to the broader one below.
     # ``would_run_wrong`` is the one refusal the ENGINE DOES NOT MAKE: the draft
@@ -477,7 +479,6 @@ def propose_repair(
     if vetted.would_run_wrong:
         return _no_fix(_judged_wrong_reason(vetted))
     if vetted.rejections:
-        diagnostics.report("validation_refused")
         return _no_fix()
     if not vetted.applicable:
         diagnostics.report("repair_not_applicable")

api/services/dify_builder/graph_ops.py

@@ -11,6 +11,7 @@ import copy
 import hashlib
 import json
 from collections.abc import Callable
+from contextlib import suppress
 from typing import Any, NamedTuple
 
 from core.dify_builder.models import Graph, MutationIntent
@@ -59,6 +60,19 @@ _REDACTABLE_ARGS = ("value", "config")
 _MODE_KEY = "type"
 
 
+class OperationValidationError(ValueError):
+    """A structural refusal with a source-owned, value-free observation code.
+
+    Message and ValueError compatibility remain the mutation caller's contract;
+    the code is observation only and never controls applicability.
+    """
+
+    def __init__(self, message: str, reason_code: str) -> None:
+        """Preserve the original refusal text and its fixed source classification."""
+        super().__init__(message)
+        self.reason_code = reason_code
+
+
 def validate_intent_args(intent: MutationIntent) -> None:
     """Raise ``ValueError`` if ``intent.args`` is missing a required key for
     ``intent.op``, or if ``intent.op`` isn't one of ``MUTATION_ARG_KEYS``'s
@@ -77,24 +91,28 @@ def validate_intent_args(intent: MutationIntent) -> None:
     """
     required = MUTATION_ARG_KEYS.get(intent.op)
     if required is None:
-        raise ValueError(f"unknown mutation op: {intent.op!r}")
+        raise OperationValidationError(f"unknown mutation op: {intent.op!r}", "unknown_operation")
     missing = [key for key in required if key not in intent.args]
     if missing:
-        raise ValueError(f"missing required arg(s) {missing} for op {intent.op!r}")
+        raise OperationValidationError(f"missing required arg(s) {missing} for op {intent.op!r}", "missing_argument")
 
     for key in _NODE_ID_ARGS:
         if key not in intent.args:
             continue
         value = intent.args[key]
         if not isinstance(value, str) or not value:
-            raise ValueError(f"{key} must be a non-empty string for op {intent.op!r}, got {value!r}")
+            raise OperationValidationError(
+                f"{key} must be a non-empty string for op {intent.op!r}, got {value!r}", "invalid_argument"
+            )
 
     for key in _STRING_ARGS:
         if key not in intent.args:
             continue
         value = intent.args[key]
         if not isinstance(value, str) or not value:
-            raise ValueError(f"{key} must be a non-empty string for op {intent.op!r}, got {value!r}")
+            raise OperationValidationError(
+                f"{key} must be a non-empty string for op {intent.op!r}, got {value!r}", "invalid_argument"
+            )
 
     # KNOWN AND ACCEPTED, deliberately not closed here: this stops the model
     # handing the PLACEHOLDER back, not a rewrite that simply OMITS the
@@ -106,10 +124,11 @@ def validate_intent_args(intent: MutationIntent) -> None:
     # which is a larger design change and would break legitimate deletion.
     for key in _REDACTABLE_ARGS:
         if key in intent.args and credentials.carries_redaction(intent.args[key]):
-            raise ValueError(
+            raise OperationValidationError(
                 f"{key} for op {intent.op!r} contains the redaction placeholder "
                 f"{credentials.REDACTED!r}: that is not the real secret. Leave the field "
-                "unchanged, or set it to a real value."
+                "unchanged, or set it to a real value.",
+                "redacted_value",
             )
 
 
@@ -129,35 +148,45 @@ def _resolve_path(container: dict[str, Any], path: str) -> tuple[Any, str | int]
     """
     segments = path.split(".")
     if not path or any(segment == "" for segment in segments):
-        raise ValueError(f"empty segment in path: {path!r}")
+        raise OperationValidationError(f"empty segment in path: {path!r}", "invalid_path")
 
     cursor: Any = container
     for depth, segment in enumerate(segments[:-1]):
         walked = ".".join(segments[: depth + 1])
         if segment.isdigit():
             if not isinstance(cursor, list):
-                raise ValueError(f"path {path!r}: {walked} indexes a {type(cursor).__name__}, not a list")
+                raise OperationValidationError(
+                    f"path {path!r}: {walked} indexes a {type(cursor).__name__}, not a list", "invalid_path"
+                )
             index = int(segment)
             if index >= len(cursor):
-                raise ValueError(f"path {path!r}: index {index} out of range at {walked} (len {len(cursor)})")
+                raise OperationValidationError(
+                    f"path {path!r}: index {index} out of range at {walked} (len {len(cursor)})", "invalid_path"
+                )
             cursor = cursor[index]
             continue
         if not isinstance(cursor, dict):
-            raise ValueError(f"path {path!r}: {walked} is not a mapping")
+            raise OperationValidationError(f"path {path!r}: {walked} is not a mapping", "invalid_path")
         if segment not in cursor:
-            raise ValueError(f"path {path!r}: no key {segment!r} at {walked}")
+            raise OperationValidationError(f"path {path!r}: no key {segment!r} at {walked}", "invalid_path")
         cursor = cursor[segment]
 
     last = segments[-1]
     if last.isdigit():
         if not isinstance(cursor, list):
-            raise ValueError(f"path {path!r}: final segment indexes a {type(cursor).__name__}, not a list")
+            raise OperationValidationError(
+                f"path {path!r}: final segment indexes a {type(cursor).__name__}, not a list", "invalid_path"
+            )
         index = int(last)
         if index >= len(cursor):
-            raise ValueError(f"path {path!r}: index {index} out of range (len {len(cursor)})")
+            raise OperationValidationError(
+                f"path {path!r}: index {index} out of range (len {len(cursor)})", "invalid_path"
+            )
         return cursor, index
     if not isinstance(cursor, dict):
-        raise ValueError(f"path {path!r}: final container is a {type(cursor).__name__}, not a mapping")
+        raise OperationValidationError(
+            f"path {path!r}: final container is a {type(cursor).__name__}, not a mapping", "invalid_path"
+        )
     return cursor, last
 
 
@@ -207,7 +236,7 @@ def apply_set_node_config(graph: Graph, node_id: str, path: str, value: Any) ->
             parent[key] = copy.deepcopy(value)
             return new_graph, [node_id]
 
-    raise ValueError(f"node not found: {node_id}")
+    raise OperationValidationError(f"node not found: {node_id}", "node_not_found")
 
 
 def _next_unique_id(base: str, existing_ids: set) -> str:
@@ -419,7 +448,7 @@ def apply_delete_node(graph: Graph, node_id: str) -> tuple[Graph, list[str]]:
     new_graph = copy.deepcopy(graph)
     nodes = new_graph.get("nodes", [])
     if not any(n.get("id") == node_id for n in nodes):
-        raise ValueError(f"node not found: {node_id}")
+        raise OperationValidationError(f"node not found: {node_id}", "node_not_found")
 
     new_graph["nodes"] = [n for n in nodes if n.get("id") != node_id]
     new_graph["edges"] = [
@@ -465,9 +494,9 @@ def apply_connect(
     new_graph = copy.deepcopy(graph)
     nodes_by_id = {n.get("id"): n for n in new_graph.get("nodes", [])}
     if from_node not in nodes_by_id:
-        raise ValueError(f"node not found: {from_node}")
+        raise OperationValidationError(f"node not found: {from_node}", "node_not_found")
     if to_node not in nodes_by_id:
-        raise ValueError(f"node not found: {to_node}")
+        raise OperationValidationError(f"node not found: {to_node}", "node_not_found")
 
     # A branch node (if-else / question-classifier / human-input / fail-branch)
     # only routes along the handles it declares. An edge on any other handle
@@ -477,7 +506,9 @@ def apply_connect(
     declared = declared_branch_handles(nodes_by_id[from_node])
     handle = source_handle or "source"
     if declared and handle not in declared:
-        raise ValueError(f"branch node {from_node!r} has no handle {handle!r}; it declares {declared}")
+        raise OperationValidationError(
+            f"branch node {from_node!r} has no handle {handle!r}; it declares {declared}", "invalid_branch_handle"
+        )
 
     existing_edge_ids = {e.get("id") for e in new_graph.get("edges", [])}
     edge = _make_edge(from_node, to_node, source_handle, target_handle, existing_edge_ids)
@@ -820,6 +851,8 @@ def filter_applicable(
     graph: Graph,
     intents: list[MutationIntent],
     allowed_node_types: set[str] | None = None,
+    *,
+    on_rejection: Callable[[MutationIntent, str], None] | None = None,
 ) -> DryRun:
     """Dry-run each intent through the real validate_intent_args + apply_* on a
     working deep copy, in order (so a connect sees a node an earlier create_node
@@ -841,6 +874,9 @@ def filter_applicable(
     present" means -- so the graph returned here is still what the port would
     write.
 
+    ``on_rejection`` receives only source-owned codes. Observation failures are
+    suppressed and cannot change any existing mutation verdict or error text.
+
     The working copy then goes through the shared deterministic heal set
     (``services.workflow_generation.graph_normalizers.heal_nodes_for_preflight``), exactly as
     ``dify_port.apply_repair`` runs it over the real draft after its own apply
@@ -867,10 +903,18 @@ def filter_applicable(
                     intent.op == "create_node" and node_type == "" and intent.args.get("flow_type") == "custom-note"
                 ):
                     rejected.append((intent, f"node_type not allowed: {node_type!r}"))
+                    if on_rejection is not None:
+                        with suppress(Exception):
+                            on_rejection(intent, "unsupported_node_type")
                     continue
             working, changed = APPLY_FNS[intent.op](working, **intent.args)
         except Exception as exc:
             rejected.append((intent, str(exc)))
+            if on_rejection is not None:
+                with suppress(Exception):
+                    on_rejection(
+                        intent, exc.reason_code if isinstance(exc, OperationValidationError) else "operation_invalid"
+                    )
             continue
         # A connect writes an EDGE; its endpoints' node data is untouched (see
         # ``DryRun.changed_nodes``). Mirrored in ``dify_port.apply_repair``.

api/services/dify_builder/preflight.py

@@ -31,7 +31,7 @@ from typing import Any, NamedTuple
 from pydantic import ValidationError
 
 from core.dify_builder.models import Graph, MutationIntent
-from services.dify_builder import credentials, graph_ops
+from services.dify_builder import credentials, graph_ops, trace_diagnostics
 
 # Only for the node-type knowledge the rejoin guard's REASON needs -- which
 # variable a node publishes -- stated once there so the sentence Edit's prompt
@@ -222,6 +222,31 @@ class _NodeProblem(NamedTuple):
     node: tuple[str, str]
     locations: frozenset[tuple[str, ...]]
     message: str
+    validation_details: tuple[dict[str, Any], ...] = ()
+
+
+def _native_validation_details(exc: BaseException, node_index: int) -> tuple[dict[str, Any], ...]:
+    """Observe the actual native exception without values, repr or prose.
+
+    Follow at most eight wrapped causes. Non-Pydantic exceptions have only a
+    fixed category; their private exception message is never metadata.
+    """
+    current: BaseException | None = exc
+    for _ in range(8):
+        if isinstance(current, ValidationError):
+            return tuple(
+                trace_diagnostics.validation_detail(
+                    {
+                        "stage": "native_config",
+                        "reason_code": error["type"],
+                        "node_index": node_index,
+                        "path": error["loc"],
+                    }
+                )
+                for error in current.errors(include_input=False, include_context=False, include_url=False)
+            )
+        current = (current.__cause__ or current.__context__) if current is not None else None
+    return ({"stage": "native_config", "reason_code": "native_validation_exception", "node_index": node_index},)
 
 
 def _node_problems(graph: Graph) -> list[_NodeProblem]:
@@ -231,19 +256,43 @@ def _node_problems(graph: Graph) -> list[_NodeProblem]:
     touched-node validity and unchanged-draft exemptions apply to both.
     """
     problems: list[_NodeProblem] = []
-    for node in graph.get("nodes") or []:
+    for node_index, node in enumerate(graph.get("nodes") or []):
         if not isinstance(node, Mapping) or node.get("type") == _CANVAS_ONLY_NODE_TYPE:
             continue
         exc = _validation_error(node)
         if exc is None:
             continue
         locations = _error_locations(exc)
-        problems.append(_NodeProblem(_node_key(node), locations, _withheld_message(node, exc, locations)))
+        problems.append(
+            _NodeProblem(
+                _node_key(node),
+                locations,
+                _withheld_message(node, exc, locations),
+                _native_validation_details(exc, node_index),
+            )
+        )
     problems.extend(
         _NodeProblem(
             (problem.node_id, "if-else"),
             frozenset({problem.location}),
             f"node {problem.node_id!r} (if-else): {problem.detail}",
+            (
+                trace_diagnostics.validation_detail(
+                    {
+                        "stage": "semantic",
+                        "reason_code": "native_http_condition",
+                        "path": problem.location,
+                        "node_index": next(
+                            (
+                                index
+                                for index, node in enumerate(graph.get("nodes") or [])
+                                if isinstance(node, Mapping) and node.get("id") == problem.node_id
+                            ),
+                            None,
+                        ),
+                    }
+                ),
+            ),
         )
         for problem in native_http_condition_problems(graph.get("nodes") or [])
     )
@@ -314,12 +363,21 @@ def new_preflight_problems(before: Graph, after: Graph, touched: Collection[str]
     which is why ``touched`` has to be the same set on both sides too (the
     pre-heal ``changed_nodes``; see ``graph_ops.DryRun``).
     """
+    return [problem.message for problem in _new_node_problems(before, after, touched)]
+
+
+def _new_node_problems(before: Graph, after: Graph, touched: Collection[str]) -> list[_NodeProblem]:
+    """Return the same touched/new-location refusals with their source metadata.
+
+    Inputs and exemptions match ``new_preflight_problems`` exactly; no mutation
+    occurs and neither diagnostics nor their classification affects the verdict.
+    """
     written = frozenset(touched)
     known: dict[tuple[str, str], frozenset[tuple[str, ...]]] = {}
     for problem in _node_problems(before):
         known[problem.node] = known.get(problem.node, frozenset()) | problem.locations
     return [
-        problem.message
+        problem
         for problem in _node_problems(after)
         if problem.node[0] in written or problem.locations - known.get(problem.node, frozenset())
     ]
@@ -787,6 +845,10 @@ class VettedIntents(NamedTuple):
       validation accepts it. The node policy is also enforced at the write
       boundary; the other semantic guards protect model candidate selection.
 
+    ``validation_details`` adds value-free observations from the actual sources.
+    Its empty default preserves existing three-argument constructors; historical
+    absent observations stay unknown and never become reconstructed causes.
+
     Reading which is which off the rejection PROSE would be the same mistake as
     keying control flow on the model's prose: the text is for a human and a
     re-prompt, the list is for the code.
@@ -795,6 +857,7 @@ class VettedIntents(NamedTuple):
     applicable: list[MutationIntent]
     rejections: list[str]
     would_run_wrong: list[str]
+    validation_details: tuple[dict[str, Any], ...] = ()
 
 
 def vet_intents(
@@ -832,14 +895,42 @@ def vet_intents(
     their existing filtering behavior; isolated creates with no rejected wiring
     are allowed in either mode.
     """
-    dry_run = graph_ops.filter_applicable(graph, intents, allowed_node_types)
+    details: list[dict[str, Any]] = []
+
+    def observe_rejection(intent: MutationIntent, reason_code: str) -> None:
+        """Attach only canonical operation/path and the known graph position."""
+        node_id, path = intent.args.get("node_id"), intent.args.get("path")
+        details.append(
+            trace_diagnostics.validation_detail(
+                {
+                    "stage": "operation",
+                    "reason_code": reason_code,
+                    "op": intent.op,
+                    "node_index": next(
+                        (
+                            index
+                            for index, node in enumerate(graph.get("nodes") or [])
+                            if isinstance(node, Mapping) and node.get("id") == node_id
+                        ),
+                        None,
+                    )
+                    if isinstance(node_id, str)
+                    else None,
+                    "path": path.split(".") if isinstance(path, str) else None,
+                }
+            )
+        )
+
+    dry_run = graph_ops.filter_applicable(graph, intents, allowed_node_types, on_rejection=observe_rejection)
+    node_problems = _new_node_problems(graph, dry_run.graph, dry_run.changed_nodes)
+    details.extend(detail for problem in node_problems for detail in problem.validation_details)
     rejections = [f"- {intent.op} {_withheld_args(intent)}: {reason}" for intent, reason in dry_run.rejected]
     rejections += [
-        f"- the draft would then not start: {problem}"
+        f"- the draft would then not start: {problem.message}"
         # The dry run's own record of what the batch wrote, which is the port's
         # too -- a node this batch touches has to be startable, not merely no
         # worse than it was.
-        for problem in new_preflight_problems(graph, dry_run.graph, dry_run.changed_nodes)
+        for problem in node_problems
     ]
     # The third layer: Builder policy and defects native validation accepts.
     # They are keyed on proposed native effects, never the model's verdict.
@@ -856,7 +947,8 @@ def vet_intents(
     if atomic_added_node_wiring:
         would_run_wrong += _partial_added_node_wiring_reasons(graph, dry_run)
     rejections += [f"- {reason}" for reason in would_run_wrong]
-    return VettedIntents(dry_run.applicable, rejections, would_run_wrong)
+    details.extend({"stage": "semantic", "reason_code": "semantic_guard"} for _ in would_run_wrong)
+    return VettedIntents(dry_run.applicable, rejections, would_run_wrong, tuple(details))
 
 
 def _partial_added_node_wiring_reasons(before: Graph, dry_run: graph_ops.DryRun) -> list[str]:

api/services/dify_builder/trace_diagnostics.py

@@ -80,6 +80,157 @@ _MESSAGES = {
 }
 
 
+# Paths are schema vocabulary, never arbitrary user dictionary keys or numeric
+# values. Unknown fields stay unknown, including custom Code output names.
+_VALIDATION_PATH_FIELDS = frozenset(
+    [
+        "type",
+        "title",
+        "code",
+        "code_language",
+        "outputs",
+        "variables",
+        "variable",
+        "value",
+        "value_type",
+        "variable_selector",
+        "cases",
+        "conditions",
+        "comparison_operator",
+        "logical_operator",
+        "case_id",
+        "id",
+        "authorization",
+        "config",
+        "headers",
+        "body",
+        "data",
+        "method",
+        "url",
+        "params",
+        "timeout",
+        "connect",
+        "read",
+        "write",
+        "retry_config",
+        "delivery_methods",
+        "inputs",
+        "user_actions",
+        "form_content",
+        "enabled",
+        "default",
+        "selector",
+        "classes",
+        "metadata_filtering_conditions",
+        "prompt_template",
+        "model",
+        "provider",
+        "name",
+        "api_key",
+        "query",
+        "memory",
+        "context",
+        "vision",
+        "max_tokens",
+        "temperature",
+        "options",
+        "required",
+        "label",
+    ]
+) | {"*", "unknown"}
+_VALIDATION_REASONS = {
+    "operation": frozenset(
+        {
+            "unknown_operation",
+            "missing_argument",
+            "invalid_argument",
+            "redacted_value",
+            "invalid_path",
+            "node_not_found",
+            "invalid_branch_handle",
+            "unsupported_node_type",
+            "operation_invalid",
+        }
+    ),
+    "native_config": frozenset(
+        {
+            "missing",
+            "literal_error",
+            "string_type",
+            "int_type",
+            "int_parsing",
+            "float_type",
+            "float_parsing",
+            "bool_type",
+            "bool_parsing",
+            "list_type",
+            "dict_type",
+            "model_type",
+            "enum",
+            "value_error",
+            "union_tag_invalid",
+            "union_tag_not_found",
+            "extra_forbidden",
+            "finite_number",
+            "greater_than",
+            "greater_than_equal",
+            "less_than",
+            "less_than_equal",
+            "too_short",
+            "too_long",
+            "uuid_parsing",
+            "native_validation_error",
+            "native_validation_exception",
+        }
+    ),
+    "semantic": frozenset({"semantic_guard", "native_http_condition"}),
+}
+_VALIDATION_OPS = frozenset({"set_node_config", "create_node", "delete_node", "connect", "insert_between"})
+
+
+def validation_detail(record: Mapping[str, Any]) -> dict[str, Any]:
+    """Project a source observation to bounded schema vocabulary only.
+
+    Accept source metadata or stored records; return no proposal values, ids,
+    exception text or arbitrary field names. Unknown classifications remain
+    explicit unknown within their source stage. This never grants authority.
+    """
+    stage = record.get("stage")
+    if not isinstance(stage, str) or stage not in _VALIDATION_REASONS:
+        return {}
+    reason = record.get("reason_code")
+    fallback = {
+        "operation": "operation_invalid",
+        "native_config": "native_validation_error",
+        "semantic": "semantic_guard",
+    }
+    safe: dict[str, Any] = {
+        "stage": stage,
+        "reason_code": reason if isinstance(reason, str) and reason in _VALIDATION_REASONS[stage] else fallback[stage],
+    }
+    op = record.get("op")
+    if isinstance(op, str) and op in _VALIDATION_OPS:
+        safe["op"] = op
+    # Node position identifies the actual validator target without persisting an
+    # arbitrary graph/model-authored id that might itself contain a secret.
+    index = record.get("node_index")
+    if type(index) is int and 0 <= index <= 1_000_000:
+        safe["node_index"] = index
+    path = record.get("path")
+    if isinstance(path, (list, tuple)):
+        safe["path"] = [
+            part
+            if isinstance(part, str) and part in _VALIDATION_PATH_FIELDS
+            else "*"
+            if type(part) is int or (isinstance(part, str) and part.isdecimal())
+            else "unknown"
+            for part in path[:8]
+        ]
+        if len(path) > 8 or record.get("path_truncated") is True:
+            safe["path_truncated"] = True
+    return safe
+
+
 def project(record: Mapping[str, Any]) -> dict[str, Any]:
     """Ignore arbitrary text/extra fields, including legacy generator errors."""
     code = record.get("code")
@@ -148,6 +299,16 @@ def project(record: Mapping[str, Any]) -> dict[str, Any]:
         "unknown",
     }:
         result["reason"] = reason
+    details = record.get("validation_details")
+    if code == "validation_refused" and isinstance(details, (list, tuple)):
+        result["validation_details"] = [
+            safe for detail in details[:16] if isinstance(detail, Mapping) and (safe := validation_detail(detail))
+        ]
+        count = record.get("validation_detail_count")
+        result["validation_detail_count"] = (
+            max(len(details), count) if type(count) is int and 0 <= count <= 1_000_000_000 else len(details)
+        )
+        result["validation_details_truncated"] = result["validation_detail_count"] > len(result["validation_details"])
     codes = record.get("validation_codes")
     allowed = {
         "UNRESOLVED_REFERENCE",
@@ -279,6 +440,7 @@ def report(
     attempt: int | None = None,
     fallback_used: bool | None = None,
     validation_code: str | None = None,
+    validation_details: Sequence[Mapping[str, Any]] | None = None,
     node_id: str | None = None,
     provider: str | None = None,
     model: str | None = None,
@@ -295,6 +457,7 @@ def report(
             "fallback_used": fallback_used,
             "reason": failure_reason(error) if error else None,
             "validation_codes": [validation_code] if validation_code else None,
+            "validation_details": validation_details,
             "node_id": node_id,
             "provider": provider,
             "model": model,

api/tests/unit_tests/services/dify_builder/test_fix_cognition.py

@@ -1737,6 +1737,8 @@ def test_rejected_nonempty_repair_remains_validation_refused() -> None:
     assert intents == []
     assert len(model.calls) == 2
     assert [record["code"] for record in records] == ["validation_refused", "validation_refused"]
+    assert [record["attempt"] for record in records] == [1, 2]
+    assert all(record["validation_details"][0]["stage"] == "native_config" for record in records)
 
 
 @pytest.mark.parametrize("risk_payload", [{}, {"reason": "   \n"}])
@@ -1776,3 +1778,82 @@ def test_empty_repair_redacts_overlapping_credentials_with_complete_runtime_prof
     )
 
     assert risk.reason == "403 permission required: __DIFY_BUILDER_REDACTED__ __DIFY_BUILDER_REDACTED__"
+
+
+def test_refusals_keep_distinct_actual_stage_and_reason_per_attempt() -> None:
+    from services.dify_builder import trace_diagnostics as diagnostics
+
+    bad_path = json.loads(_operator_repair("=="))
+    bad_path["intents"][0]["args"]["path"] = "cases.900.conditions"
+    model = _RecordingInstance([json.dumps(bad_path), _operator_repair("==")])
+    records = []
+    with diagnostics.scope(records.append, "propose-repair"):
+        intents, risk = fix.propose_repair(model, _IF_ELSE_DIAG, _IF_ELSE_GRAPH)
+    assert intents == []
+    assert risk.level == "high"
+    assert len(model.calls) == 2
+    assert [
+        (r["attempt"], r["validation_details"][0]["stage"], r["validation_details"][0]["reason_code"]) for r in records
+    ] == [
+        (1, "operation", "invalid_path"),
+        (2, "native_config", "literal_error"),
+    ]
+
+
+def test_semantic_final_early_return_keeps_both_refusal_observations() -> None:
+    from services.dify_builder import trace_diagnostics as diagnostics
+
+    model = _RecordingInstance([_UNFED_BRANCH_REPAIR, _UNFED_BRANCH_REPAIR])
+    records = []
+    with diagnostics.scope(records.append, "propose-repair"):
+        intents, risk = fix.propose_repair(model, _REJOIN_DIAG, _REJOIN_GRAPH)
+    assert intents == []
+    assert "node5" in risk.reason
+    assert len(model.calls) == 2
+    assert [r["attempt"] for r in records] == [1, 2]
+    assert all(r["validation_details"] == [{"stage": "semantic", "reason_code": "semantic_guard"}] for r in records)
+
+
+def test_valid_code_syntax_repair_after_native_refusal_preserves_declarations() -> None:
+    from services.dify_builder import graph_ops
+    from services.dify_builder import trace_diagnostics as diagnostics
+
+    graph = deepcopy(_RG)
+    declarations = {
+        "outputs": {"result": {"type": "number", "children": None}},
+        "variables": [{"variable": "x", "value_selector": ["start", "x"]}],
+    }
+    graph["nodes"][0]["data"].update(declarations)
+    graph["nodes"][0]["data"]["code"] = "def main(x)\n    return {'result': x}"
+    payload = {
+        "intents": [
+            {
+                "op": "set_node_config",
+                "args": {"node_id": "code1", "path": "code_language", "value": "PRIVATE_INVALID_LANGUAGE"},
+            }
+        ]
+    }
+    fixed = {
+        "intents": [
+            {
+                "op": "set_node_config",
+                "args": {"node_id": "code1", "path": "code", "value": "def main(x):\n    return {'result': x}"},
+            }
+        ]
+    }
+    model = _RecordingInstance([json.dumps(payload), json.dumps(fixed)])
+    records = []
+    with diagnostics.scope(records.append, "propose-repair"):
+        intents, risk = fix.propose_repair(model, _DIAG, graph)
+    assert len(intents) == 1
+    assert len(model.calls) == 2
+    assert risk.level == "low"
+    after = graph_ops.filter_applicable(graph, intents).graph
+    assert after["nodes"][0]["data"]["outputs"] == declarations["outputs"]
+    assert after["nodes"][0]["data"]["variables"] == declarations["variables"]
+    assert after["nodes"][0]["data"]["code"] == "def main(x):\n    return {'result': x}"
+    assert records[0]["attempt"] == 1
+    assert records[0]["validation_details"][0]["path"] == ["code_language"]
+    assert len(records) == 1
+    assert "PRIVATE_INVALID_LANGUAGE" not in json.dumps(records)
+    assert "def main" not in json.dumps(records)

api/tests/unit_tests/services/dify_builder/test_preflight.py

@@ -1626,3 +1626,121 @@ def test_edge_only_wiring_of_historical_retrieval_does_not_change_native_binding
     assert vetted.applicable == [intent]
     assert vetted.rejections == []
     assert graph == before
+
+
+def test_repair_validation_details_distinguish_structural_native_and_semantic_without_values() -> None:
+    structural = vet_intents(
+        _scoring_graph(),
+        [
+            MutationIntent(
+                op="set_node_config",
+                args={"node_id": "branch", "path": "cases.900.conditions", "value": "SECRET_VALUE"},
+            )
+        ],
+    )
+    assert structural.validation_details[0]["stage"] == "operation"
+    assert structural.validation_details[0]["reason_code"] == "invalid_path"
+    assert structural.validation_details[0]["op"] == "set_node_config"
+    assert structural.validation_details[0]["path"] == ["cases", "*", "conditions"]
+
+    native = vet_intents(
+        _scoring_graph(),
+        [
+            MutationIntent(
+                op="set_node_config",
+                args={"node_id": "branch", "path": "cases.0.conditions.0.comparison_operator", "value": "SECRET_VALUE"},
+            )
+        ],
+    )
+    assert native.validation_details[0]["stage"] == "native_config"
+    assert native.validation_details[0]["reason_code"] == "literal_error"
+    assert native.validation_details[0]["path"] == ["cases", "*", "conditions", "*", "comparison_operator"]
+    assert native.validation_details[0]["node_index"] == 1
+    assert native.would_run_wrong == []
+    assert "SECRET_VALUE" not in json.dumps(native.validation_details)
+    output_graph = _code_node(code="def main():\n    return {}", code_language="python3", outputs={}, variables=[])
+    output_refusal = vet_intents(
+        output_graph,
+        [
+            MutationIntent(
+                op="set_node_config",
+                args={
+                    "node_id": "code1",
+                    "path": "outputs",
+                    "value": {
+                        "SECRET_OUTPUT_NAME": {"type": "SECRET_TYPE", "children": {"SECRET_NESTED": "SECRET_VALUE"}}
+                    },
+                },
+            )
+        ],
+    )
+    assert output_refusal.validation_details
+    assert all(detail["stage"] == "native_config" for detail in output_refusal.validation_details)
+    assert "SECRET" not in json.dumps(output_refusal.validation_details)
+
+
+def test_validation_metadata_never_copies_unknown_operation_node_or_path_strings() -> None:
+    marker = "SECRET_MARKER"
+    graph = _scoring_graph()
+    graph["nodes"][1]["id"] = marker
+    refused = vet_intents(
+        graph,
+        [
+            MutationIntent(
+                op="set_node_config",
+                args={"node_id": marker, "path": marker + "." + "x" * 10000, "value": {marker: marker}},
+            ),
+            MutationIntent(op=marker, args={marker: marker}),
+        ],
+    )
+    assert len(refused.validation_details) == 2
+    assert refused.validation_details[0]["node_index"] == 1
+    assert refused.validation_details[0]["path"] == ["unknown", "unknown"]
+    assert refused.validation_details[1]["reason_code"] == "unknown_operation"
+    assert "op" not in refused.validation_details[1]
+    assert marker not in json.dumps(refused.validation_details)
+    assert "x" * 100 not in json.dumps(refused.validation_details)
+
+
+@pytest.mark.parametrize(
+    ("op", "args", "reason"),
+    [
+        ("private-unknown-op", {"value": "__DIFY_BUILDER_REDACTED__"}, "unknown_operation"),
+        ("set_node_config", {"value": "__DIFY_BUILDER_REDACTED__"}, "missing_argument"),
+        (
+            "set_node_config",
+            {"node_id": "branch", "path": "title", "value": "__DIFY_BUILDER_REDACTED__"},
+            "redacted_value",
+        ),
+    ],
+)
+def test_validation_observation_preserves_existing_first_refusal_priority(
+    op: str, args: dict[str, object], reason: str
+) -> None:
+    vetted = vet_intents(_scoring_graph(), [MutationIntent(op=op, args=args)])
+    assert vetted.applicable == []
+    assert len(vetted.rejections) == 1
+    assert vetted.validation_details[0]["reason_code"] == reason
+    assert "__DIFY_BUILDER_REDACTED__" not in json.dumps(vetted.validation_details)
+
+
+def test_native_validation_crash_is_fixed_unknown_category_without_exception_text() -> None:
+    graph: Graph = {"nodes": [_http_node("h", {"type": "no-auth"})], "edges": []}
+    vetted = vet_intents(
+        graph,
+        [
+            MutationIntent(
+                op="set_node_config",
+                args={
+                    "node_id": "h",
+                    "path": "authorization",
+                    "value": {"type": "PRIVATE_INVALID_AUTH", "config": {"type": "bearer", "api_key": "PRIVATE_TOKEN"}},
+                },
+            )
+        ],
+    )
+    assert vetted.validation_details == (
+        {"stage": "native_config", "reason_code": "native_validation_exception", "node_index": 0},
+    )
+    assert "PRIVATE_TOKEN" not in json.dumps(vetted.validation_details)
+    assert vetted.would_run_wrong == []

api/tests/unit_tests/services/dify_builder/test_trace_diagnostics.py

@@ -539,3 +539,111 @@ def test_resource_plan_missing_id_code_survives_projection_report_and_history()
     assert history[1]["validation_codes"] == []
     assert "private-resource-id" not in json.dumps(history)
     assert "provider answer" not in json.dumps(history)
+
+
+def test_repair_detail_projection_bounds_counts_and_drops_arbitrary_metadata() -> None:
+    raw = {
+        "stage": "operation",
+        "reason_code": "invalid_path",
+        "op": "set_node_config",
+        "node_index": 1,
+        "node_id": "SECRET",
+        "path": ["cases", "*", "SECRET"],
+        "value": {"SECRET": "SECRET"},
+        "message": "SECRET",
+        "exception_type": "SECRET",
+    }
+    projected = diagnostics.project({"code": "validation_refused", "validation_details": [raw] * 40})
+    assert len(projected["validation_details"]) == 16
+    assert projected["validation_detail_count"] == 40
+    assert projected["validation_details_truncated"] is True
+    assert projected["validation_details"][0]["path"] == ["cases", "*", "unknown"]
+    assert "SECRET" not in json.dumps(projected)
+    assert diagnostics.project(projected)["validation_detail_count"] == 40
+    assert "validation_details" not in diagnostics.project({"code": "validation_refused"})
+    deep = diagnostics.project(
+        {"code": "validation_refused", "validation_details": [{**raw, "path": ["cases"] * 10000}]}
+    )
+    assert len(deep["validation_details"][0]["path"]) == 8
+    assert deep["validation_details"][0]["path_truncated"] is True
+    from core.dify_builder.models import MutationIntent
+    from services.dify_builder.preflight import vet_intents
+
+    actual = vet_intents({"nodes": [], "edges": []}, [MutationIntent(op="SECRET", args={}) for _ in range(40)])
+    records = []
+    with diagnostics.scope(records.append, "propose-repair"):
+        diagnostics.report("validation_refused", attempt=1, validation_details=actual.validation_details)
+    assert len(records[0]["validation_details"]) == 16
+    assert records[0]["validation_detail_count"] == 40
+    assert records[0]["validation_details_truncated"] is True
+    for unknown in ("SECRET", {"SECRET": "SECRET"}, ["SECRET"], None):
+        detail = diagnostics.validation_detail({"stage": "native_config", "reason_code": unknown, "path": [unknown]})
+        assert detail == {"stage": "native_config", "reason_code": "native_validation_error", "path": ["unknown"]}
+    assert "SECRET" not in json.dumps(records)
+
+
+def test_repair_details_survive_real_cas_item_history_and_http_projection(repo: SqlDifyBuilderRepository) -> None:
+    from controllers.console.dify_builder_fields import DifyBuilderConversationPageResponse
+    from core.dify_builder.contract import AssistantTurnItem, ExecutionProgress
+    from core.dify_builder.models import Action, DifyBuilderContext, MutationIntent, Turn
+    from core.dify_builder.state import PcState
+    from libs.helper import dump_response
+    from services.dify_builder.application.runner import Runner, StepResult
+    from services.dify_builder.preflight import vet_intents
+    from tests.unit_tests.core.dify_builder.test_runner import _actor
+
+    env, _ = _new_env()
+    env.repo = repo
+    events = []
+    env.emit_item = events.append
+    session = _session()
+    repo.create_session(session, DifyBuilderContext(), [])
+
+    def handler(_env: Env, _turn: Turn, current: Session, context: DifyBuilderContext) -> StepResult:
+        vetted = vet_intents(
+            {"nodes": [], "edges": []}, [MutationIntent(op="PRIVATE_OP", args={"value": "PRIVATE_VALUE"})]
+        )
+        diagnostics.report("validation_refused", attempt=2, validation_details=vetted.validation_details)
+        item = AssistantTurnItem(
+            turn_id="repair-trace",
+            stage_id=str(current.current_state),
+            execution=ExecutionProgress(status="completed"),
+            reply_text="Review repair",
+            reply_mode="complete",
+        ).to_item(seq=0, at_version=2)
+        return StepResult(next=PcState.FIX_AWAIT_APPROVAL, context=context, items=[item])
+
+    Runner(env, {PcState.FIX_DIAGNOSE: handler}).advance(
+        session.id, Turn(action=Action(kind="request_fix", base_version=1, command_id="repair-command"), actor=_actor())
+    )
+    assistant = repo.list_conversation(session.id)[-1]
+    _, context = repo.get_session(session.id)
+    assert context.last_command_id == "repair-command"
+    assert events[-1].command_id == "repair-command"
+    stored = assistant.payload["diagnostics"][0]
+    assert stored["command_id"] == "repair-command"
+    assert stored["attempt"] == 2
+    assert stored["validation_details"] == [{"stage": "operation", "reason_code": "unknown_operation"}]
+    assert stored["validation_detail_count"] == 1
+    assert stored["validation_details_truncated"] is False
+    assert events[-1].item.payload["diagnostics"] == [stored]
+    history = diagnostics.from_history([assistant], session_id=session.id, version=2)
+    assert history[0]["validation_details"] == stored["validation_details"]
+    http = dump_response(
+        DifyBuilderConversationPageResponse,
+        {
+            "data": [
+                {
+                    "seq": assistant.seq,
+                    "kind": assistant.kind,
+                    "payload": assistant.payload,
+                    "at_version": assistant.at_version,
+                }
+            ],
+            "has_more": False,
+            "first_seq": 0,
+            "last_seq": 0,
+        },
+    )
+    assert http["data"][0]["payload"]["diagnostics"][0]["validation_details"] == stored["validation_details"]
+    assert "PRIVATE" not in json.dumps(history)