[19.0][FIX] edi_queue_oca: propagate failed jobs to exchanges

edi_queue_oca/models/__init__.py

@@ -1,3 +1,4 @@
 from . import edi_exchange_record
 from . import edi_exchange_type
 from . import edi_backend
+from . import queue_job

edi_queue_oca/models/edi_exchange_record.py

@@ -66,6 +66,33 @@ def delayable(self, **kw):
     def _job_retry_params(self):
         return {}
 
+    def _mark_failed_from_queue_job(self, job):
+        """Set the EDI error matching a terminal queue job failure.
+
+        :param job: failed ``queue.job`` record
+        """
+        failure_mapping = {
+            "action_exchange_process": ("input_processed_error", "process_ko"),
+            "action_exchange_receive": ("input_receive_error", "receive_ko"),
+            "action_exchange_send": ("output_error_on_send", "send_ko"),
+        }
+        failure = failure_mapping.get(job.method_name)
+        if not failure:
+            return
+        state, message_key = failure
+        for record in self:
+            state_changed = record.edi_exchange_state != state
+            record.write(
+                {
+                    "edi_exchange_state": state,
+                    "exchange_error": job.exc_message,
+                    "exchange_error_traceback": job.exc_info,
+                    "exchanged_on": fields.Datetime.now(),
+                }
+            )
+            if state_changed:
+                record._notify_error(message_key)
+
     def _compute_related_queue_jobs_count(self):
         for rec in self:
             # TODO: We should refactor the object field on queue_job to use jsonb field

edi_queue_oca/models/queue_job.py

@@ -0,0 +1,28 @@
+# Copyright 2026 Camptocamp SA
+# License LGPL-3.0 or later (https://www.gnu.org/licenses/lgpl).
+
+from odoo import models
+
+
+class QueueJob(models.Model):
+    _inherit = "queue.job"
+
+    def write(self, vals):
+        result = super().write(vals)
+        if vals.get("state") == "failed":
+            self._mark_related_edi_exchanges_failed()
+        return result
+
+    def _mark_related_edi_exchanges_failed(self):
+        """Propagate terminal EDI job failures to their exchange records."""
+        supported_methods = {
+            "action_exchange_process",
+            "action_exchange_receive",
+            "action_exchange_send",
+        }
+        jobs = self.filtered(
+            lambda job: job.model_name == "edi.exchange.record"
+            and job.method_name in supported_methods
+        )
+        for job in jobs:
+            job.records.sudo()._mark_failed_from_queue_job(job)

edi_queue_oca/tests/test_backend_jobs.py

@@ -94,6 +94,7 @@ def test_output(self):
         self.assertEqual(created, self._get_related_jobs(record))
 
     def test_output_fail_retry(self):
+        """Test a retryable send failure keeps the exchange pending."""
         job_counter = self.job_counter()
         vals = {
             "model": self.partner._name,
@@ -106,8 +107,113 @@ def test_output_fail_retry(self):
         job_counter.search_created()
         with mock.patch.object(type(self.backend), "_exchange_send") as mocked:
             mocked.side_effect = ReqConnectionError("Connection broken")
-            with self.assertRaises(RetryableJobError):
+            with self.assertRaisesRegex(RetryableJobError, "Connection broken"):
                 job.perform()
+        self.assertEqual(record.edi_exchange_state, "output_pending")
+
+    def test_failed_send_job_marks_exchange_as_error(self):
+        """Test a terminal send job failure marks its exchange as failed."""
+        record = self.backend.create_record(
+            "test_csv_output",
+            {
+                "model": self.partner._name,
+                "res_id": self.partner.id,
+                "edi_exchange_state": "output_pending",
+            },
+        )
+        record._set_file_content("ABC")
+        job = record.with_delay().action_exchange_send()
+
+        job.db_record().write(
+            {
+                "state": "failed",
+                "exc_message": "Connection broken",
+                "exc_info": "Traceback of the connection failure",
+            }
+        )
+
+        self.assertEqual(record.edi_exchange_state, "output_error_on_send")
+        self.assertEqual(record.exchange_error, "Connection broken")
+        self.assertEqual(
+            record.exchange_error_traceback, "Traceback of the connection failure"
+        )
+        self.assertTrue(record.exchanged_on)
+
+    def test_failed_receive_job_marks_exchange_as_error(self):
+        """Test a terminal receive job failure marks its exchange as failed."""
+        record = self.backend.create_record(
+            "test_csv_input",
+            {
+                "model": self.partner._name,
+                "res_id": self.partner.id,
+                "edi_exchange_state": "input_pending",
+            },
+        )
+        job = record.with_delay().action_exchange_receive()
+
+        job.db_record().write(
+            {
+                "state": "failed",
+                "exc_message": "Receive failed",
+                "exc_info": "Traceback for receive",
+            }
+        )
+
+        self.assertEqual(record.edi_exchange_state, "input_receive_error")
+        self.assertEqual(record.exchange_error, "Receive failed")
+        self.assertEqual(record.exchange_error_traceback, "Traceback for receive")
+
+    def test_failed_process_job_marks_exchange_as_error(self):
+        """Test a terminal process job failure marks its exchange as failed."""
+        record = self.backend.create_record(
+            "test_csv_input",
+            {
+                "model": self.partner._name,
+                "res_id": self.partner.id,
+                "edi_exchange_state": "input_received",
+            },
+        )
+        job = record.with_delay().action_exchange_process()
+
+        job.db_record().write(
+            {
+                "state": "failed",
+                "exc_message": "Process failed",
+                "exc_info": "Traceback for process",
+            }
+        )
+
+        self.assertEqual(record.edi_exchange_state, "input_processed_error")
+        self.assertEqual(record.exchange_error, "Process failed")
+        self.assertEqual(record.exchange_error_traceback, "Traceback for process")
+
+    def test_unsupported_failed_jobs_do_not_mark_exchange_as_error(self):
+        """Test generate and non-exchange jobs do not alter the exchange state."""
+        record = self.backend.create_record(
+            "test_csv_output",
+            {
+                "model": self.partner._name,
+                "res_id": self.partner.id,
+                "edi_exchange_state": "output_pending",
+            },
+        )
+        jobs = (
+            record.with_delay().action_exchange_generate(),
+            self.backend.with_delay().exchange_send(record),
+        )
+
+        for job in jobs:
+            job.db_record().write(
+                {
+                    "state": "failed",
+                    "exc_message": "Unsupported job failed",
+                    "exc_info": "Unsupported job traceback",
+                }
+            )
+
+        self.assertEqual(record.edi_exchange_state, "output_pending")
+        self.assertFalse(record.exchange_error)
+        self.assertFalse(record.exchange_error_traceback)
 
     def test_input(self):
         job_counter = self.job_counter()