diff --git a/edi_core_oca/models/edi_backend.py b/edi_core_oca/models/edi_backend.py
index a7a15e2b6..109d88fa0 100644
--- a/edi_core_oca/models/edi_backend.py
+++ b/edi_core_oca/models/edi_backend.py
@@ -297,12 +297,19 @@ def exchange_send(self, exchange_record):
         return res
 
     def _swallable_exceptions(self):
-        # TODO: improve this list
+        # These errors are permanent because retrying the same data will fail again.
+        # They should be swallowed so the exchange can move to an error state.
+        #
+        # SQL errors are handled separately: OperationalError may be transient,
+        # while IntegrityError requires process-specific transaction handling.
         return (
             ValueError,
             FileNotFoundError,
             exceptions.UserError,
             exceptions.ValidationError,
+            TypeError,
+            LookupError,  # covers KeyError / IndexError
+            AttributeError,
         )
 
     def _send_retryable_exceptions(self):
@@ -486,7 +493,14 @@ def exchange_process(self, exchange_record):
             error = _get_exception_msg(err)
             state = "input_processed_error"
             res = f"Error: {error}"
-        except (OperationalError, IntegrityError):
+        except IntegrityError as err:
+            if self.env.context.get("_edi_process_break_on_error"):
+                raise
+            traceback = _get_exception_traceback()
+            error = _get_exception_msg(err)
+            state = "input_processed_error"
+            res = f"Error: {error}"
+        except OperationalError:
             # We don't want the finally block to be executed in this case as
             # the cursor is already in an aborted state and any query will fail.
             res = "__sql_error__"
diff --git a/edi_core_oca/tests/test_backend_process.py b/edi_core_oca/tests/test_backend_process.py
index bb6f4ffeb..1c8092db2 100644
--- a/edi_core_oca/tests/test_backend_process.py
+++ b/edi_core_oca/tests/test_backend_process.py
@@ -5,7 +5,6 @@
 import base64
 
 from freezegun import freeze_time
-from psycopg2 import IntegrityError
 
 from odoo import fields
 from odoo.exceptions import UserError
@@ -110,13 +109,31 @@ def test_process_outbound_record(self):
         with self.assertRaises(UserError):
             record.action_exchange_process()
 
+    @mute_logger("odoo.sql_db")
     def test_process_record_with_integrity_error(self):
+        conflicting_record = self.backend.create_record(
+            "test_csv_input",
+            {
+                "model": self.partner._name,
+                "res_id": self.partner.id,
+                "exchange_file": base64.b64encode(b"1234"),
+            },
+        )
+        original_identifier = self.record.identifier
+        duplicate_identifier = conflicting_record.identifier
         self.record.write({"edi_exchange_state": "input_received"})
-        with self.assertRaises(IntegrityError):
-            self.backend.with_context(
-                test_break_process=IntegrityError("SQL error")
-            ).exchange_process(self.record)
-        self.assertRecordValues(self.record, [{"edi_exchange_state": "input_received"}])
-        self.assertFalse(self.record.exchange_error)
+
+        self.record.with_context(
+            fake_update_values={"identifier": duplicate_identifier}
+        ).action_exchange_process()
+        self.assertRecordValues(
+            self.record,
+            [
+                {
+                    "identifier": original_identifier,
+                    "edi_exchange_state": "input_processed_error",
+                }
+            ],
+        )
 
     # TODO: test ack file are processed