diff --git a/lib/crewai/src/crewai/agents/crew_agent_executor.py b/lib/crewai/src/crewai/agents/crew_agent_executor.py
index 729409ef6..1fae032d7 100644
--- a/lib/crewai/src/crewai/agents/crew_agent_executor.py
+++ b/lib/crewai/src/crewai/agents/crew_agent_executor.py
@@ -905,6 +905,14 @@ class CrewAgentExecutor(BaseAgentExecutor):
             func_args, func_name, call_id, original_tool
         )
         if parse_error is not None:
+            handle_tool_failure(
+                parse_error["tool_failure"],
+                tool_name=func_name,
+                tool_args=func_args,
+                agent=self.agent,
+                task=self.task,
+                crew=self.crew,
+            )
             return parse_error
 
         if original_tool is None:
diff --git a/lib/crewai/src/crewai/experimental/agent_executor.py b/lib/crewai/src/crewai/experimental/agent_executor.py
index 9ad771a0c..35e55bee7 100644
--- a/lib/crewai/src/crewai/experimental/agent_executor.py
+++ b/lib/crewai/src/crewai/experimental/agent_executor.py
@@ -1926,6 +1926,14 @@ class AgentExecutor(Flow[AgentExecutorState], BaseAgentExecutor):
         # Parse arguments
         parsed_args, parse_error = parse_tool_call_args(func_args, func_name, call_id)
         if parse_error is not None:
+            handle_tool_failure(
+                parse_error["tool_failure"],
+                tool_name=func_name,
+                tool_args=func_args,
+                agent=self.agent,
+                task=self.task,
+                crew=self.crew,
+            )
             return parse_error
         args_dict: dict[str, Any] = parsed_args or {}
 
diff --git a/lib/crewai/src/crewai/tools/tool_failure.py b/lib/crewai/src/crewai/tools/tool_failure.py
index 9606be8ee..46c36042a 100644
--- a/lib/crewai/src/crewai/tools/tool_failure.py
+++ b/lib/crewai/src/crewai/tools/tool_failure.py
@@ -246,6 +246,12 @@ def collect_tool_failures(agent: Any) -> list[ToolFailureRecord]:
     return [record for record in records if isinstance(record, ToolFailureRecord)]
 
 
+def _agent_id(agent: Any) -> str | None:
+    """Stringified agent id, for correlating events with the call."""
+    agent_id = getattr(agent, "id", None)
+    return str(agent_id) if agent_id is not None else None
+
+
 def _record_on_agent(agent: Any, record: ToolFailureRecord) -> None:
     """Append to the agent's per-execution failure list when it has one."""
     failures = getattr(agent, "_tool_failures", None)
@@ -320,6 +326,10 @@ def handle_tool_failure(
             policy=policy,
             agent_role=record.agent_role,
             agent_key=getattr(agent, "key", None),
+            # Set explicitly rather than via from_agent, which would also
+            # overwrite agent_role and lose the _original_role preference that
+            # the paired ToolUsage events use.
+            agent_id=_agent_id(agent),
             agent=agent,
             task_name=record.task_name,
             task_id=record.task_id,
diff --git a/lib/crewai/src/crewai/tools/tool_usage.py b/lib/crewai/src/crewai/tools/tool_usage.py
index fa9fb69a7..f87a170ac 100644
--- a/lib/crewai/src/crewai/tools/tool_usage.py
+++ b/lib/crewai/src/crewai/tools/tool_usage.py
@@ -1044,9 +1044,13 @@ class ToolUsage:
     def _prepare_event_data(
         self, tool: Any, tool_calling: ToolCalling | InstructorToolCalling
     ) -> dict[str, Any]:
+        agent_id = getattr(self.agent, "id", None) if self.agent else None
         event_data = {
             "run_attempts": self._run_attempts,
             "delegations": self.task.delegations if self.task else 0,
+            # agent_key alone cannot correlate an event with a specific agent
+            # instance; the native paths already carry agent_id via from_agent.
+            "agent_id": str(agent_id) if agent_id is not None else None,
             "tool_name": sanitize_tool_name(tool.name),
             "tool_args": tool_calling.arguments,
             "tool_class": tool.__class__.__name__,
diff --git a/lib/crewai/src/crewai/utilities/agent_utils.py b/lib/crewai/src/crewai/utilities/agent_utils.py
index 7e526d376..988834826 100644
--- a/lib/crewai/src/crewai/utilities/agent_utils.py
+++ b/lib/crewai/src/crewai/utilities/agent_utils.py
@@ -1828,21 +1828,28 @@ def parse_tool_call_args(
     Returns:
         ``(args_dict, None)`` on success, or ``(None, error_result)`` on
         JSON parse failure where ``error_result`` is a ready-to-return dict
-        with the same shape as ``_execute_single_native_tool_call`` return values.
+        with the same shape as ``_execute_single_native_tool_call`` return
+        values, carrying an ``INVALID_INPUT`` failure for the caller to report.
     """
     if isinstance(func_args, str):
         try:
             return json.loads(func_args), None
         except json.JSONDecodeError as e:
+            message = (
+                f"Error: Failed to parse tool arguments as JSON: {e}. "
+                f"Please provide valid JSON arguments for the '{func_name}' tool."
+            )
             return None, {
                 "call_id": call_id,
                 "func_name": func_name,
-                "result": (
-                    f"Error: Failed to parse tool arguments as JSON: {e}. "
-                    f"Please provide valid JSON arguments for the '{func_name}' tool."
-                ),
+                "result": message,
                 "from_cache": False,
                 "original_tool": original_tool,
+                "tool_failure": ToolFailure(
+                    message=message,
+                    reason=ToolFailureReason.INVALID_INPUT,
+                    code="json_decode_error",
+                ),
             }
     return func_args, None
 
diff --git a/lib/crewai/tests/tools/test_tool_failure.py b/lib/crewai/tests/tools/test_tool_failure.py
index 39e590ec6..6391a5c28 100644
--- a/lib/crewai/tests/tools/test_tool_failure.py
+++ b/lib/crewai/tests/tools/test_tool_failure.py
@@ -1192,6 +1192,88 @@ class TestDeliberateStopIsNotAnUnknownError:
             )
 
 
+class TestEventCarriesCorrelationIds:
+    """The failure event must be correlatable with the call it describes."""
+
+    def test_agent_and_task_ids_are_populated(self) -> None:
+        crew, agent = _build_crew(ToolFailurePolicy.WARN)
+        events: list[ToolFailureDetectedEvent] = []
+
+        with crewai_event_bus.scoped_handlers():
+
+            @crewai_event_bus.on(ToolFailureDetectedEvent)
+            def _(source: Any, event: ToolFailureDetectedEvent) -> None:
+                events.append(event)
+
+            crew.kickoff()
+            crewai_event_bus.flush(timeout=10.0)
+
+        assert events
+        event = events[0]
+        assert event.agent_id == str(agent.id)
+        assert event.agent_role == agent.role
+        assert event.task_id is not None
+        assert event.task_name == "post to slack"
+
+    def test_ids_match_the_paired_finished_event(self) -> None:
+        crew, _ = _build_crew(ToolFailurePolicy.WARN)
+        failures: list[ToolFailureDetectedEvent] = []
+        finished: list[ToolUsageFinishedEvent] = []
+
+        with crewai_event_bus.scoped_handlers():
+
+            @crewai_event_bus.on(ToolFailureDetectedEvent)
+            def _f(source: Any, event: ToolFailureDetectedEvent) -> None:
+                failures.append(event)
+
+            @crewai_event_bus.on(ToolUsageFinishedEvent)
+            def _d(source: Any, event: ToolUsageFinishedEvent) -> None:
+                if event.tool_name == "slackbot_send_message":
+                    finished.append(event)
+
+            crew.kickoff()
+            crewai_event_bus.flush(timeout=10.0)
+
+        assert failures and finished
+        assert failures[0].agent_id == finished[0].agent_id
+        assert failures[0].task_id == finished[0].task_id
+
+
+class TestMalformedArgumentsAreReported:
+    """A tool call with unparseable JSON args is a failure, not a silent skip."""
+
+    @staticmethod
+    def _parse_error() -> dict[str, Any]:
+        from crewai.utilities.agent_utils import parse_tool_call_args
+
+        args, error = parse_tool_call_args("{not json", "echo", "call_1")
+        assert args is None
+        assert error is not None
+        return error
+
+    def test_parse_error_carries_an_invalid_input_failure(self) -> None:
+        error = self._parse_error()
+        failure = error["tool_failure"]
+        assert isinstance(failure, ToolFailure)
+        assert failure.reason is ToolFailureReason.INVALID_INPUT
+        assert failure.code == "json_decode_error"
+
+    def test_valid_args_carry_no_failure(self) -> None:
+        from crewai.utilities.agent_utils import parse_tool_call_args
+
+        args, error = parse_tool_call_args('{"text": "hi"}', "echo", "call_1")
+        assert args == {"text": "hi"}
+        assert error is None
+
+    def test_reason_enum_member_is_used(self) -> None:
+        """INVALID_INPUT was declared but unreferenced before this."""
+        import inspect
+
+        from crewai.utilities import agent_utils
+
+        assert "INVALID_INPUT" in inspect.getsource(agent_utils.parse_tool_call_args)
+
+
 class TestMCPIsErrorPlumbing:
     """An MCP server flags a failed tool with isError on a 200 response."""
 
diff --git a/lib/crewai/tests/utilities/test_agent_utils.py b/lib/crewai/tests/utilities/test_agent_utils.py
index 0910fb86e..755befdbe 100644
--- a/lib/crewai/tests/utilities/test_agent_utils.py
+++ b/lib/crewai/tests/utilities/test_agent_utils.py
@@ -1031,7 +1031,14 @@ class TestParseToolCallArgs:
     def test_error_result_has_correct_keys(self) -> None:
         _, error = parse_tool_call_args("{bad json}", "tool", "call_7")
         assert error is not None
-        assert set(error.keys()) == {"call_id", "func_name", "result", "from_cache", "original_tool"}
+        assert set(error.keys()) == {
+            "call_id",
+            "func_name",
+            "result",
+            "from_cache",
+            "original_tool",
+            "tool_failure",
+        }
 
 
 class TestExecuteSingleNativeToolCall: