docs+fix: record the Xero-first company merge model, align the stragglers (KAN-325)

apps/company/management/commands/merge_companies.py

@@ -1,9 +1,8 @@
 from django.core.management.base import BaseCommand
-from django.db import transaction
 
 from apps.accounts.models import Staff
 from apps.company.models import Company
-from apps.company.services.company_merge_service import reassign_company_fk_records
+from apps.company.services.company_merge_service import merge_companies
 from apps.job.models import Job
 from apps.workflow.models import CompanyDefaults
 
@@ -39,10 +38,11 @@ def handle(self, *args, **options):
             f"Looking for duplicate companies with name: '{company_name}'"
         )
 
-        # Find all companies with this name
-        duplicate_companies = Company.objects.filter(name=company_name).order_by(
-            "django_created_at"
-        )
+        # Find all companies with this name. Tombstones are already resolved —
+        # they are neither duplicates to fix nor eligible primaries.
+        duplicate_companies = Company.objects.filter(
+            name=company_name, merged_into__isnull=True
+        ).order_by("django_created_at")
         count = duplicate_companies.count()
 
         if count == 0:
@@ -95,30 +95,19 @@ def handle(self, *args, **options):
                 self.stdout.write(self.style.WARNING("Operation cancelled"))
                 return
 
-        # Merge duplicates
-        with transaction.atomic():
-            for company, job_count in companies_with_job_counts[1:]:
-                self.stdout.write(
-                    f"Merging company {company.pk} into {primary_company.pk}..."
-                )
-
-                # Reassign every company-FK record (Jobs, Invoices, Bills,
-                # Credit Notes, Quotes, POs, supplier references). The prior
-                # implementation only moved Jobs; the others ended up orphaned
-                # on the deleted company via the PROTECT constraint failure
-                # path — or silently lost on cascade with the old pointer.
-                counts = reassign_company_fk_records(
-                    company,
-                    primary_company,
-                    Staff.get_automation_user(),
-                    logger_prefix="[manual-merge] ",
-                )
-                self.stdout.write(f"  Reassigned records: {counts}")
-
-                # Delete the duplicate company — safe now that every PROTECTed
-                # FK has been moved onto the primary.
-                company.delete()
-                self.stdout.write(f"  Deleted duplicate company {company.pk}")
+        # Merge duplicates. Same tombstone semantics as the Xero-driven merge
+        # (ADR 0034): the loser keeps its row with merged_into set, so any
+        # late-arriving reference to it still resolves to the primary.
+        for company, _ in companies_with_job_counts[1:]:
+            self.stdout.write(
+                f"Merging company {company.pk} into {primary_company.pk}..."
+            )
+            counts = merge_companies(
+                company.pk,
+                primary_company.pk,
+                Staff.get_automation_user(),
+            )
+            self.stdout.write(f"  Reassigned records: {counts}")
 
         self.stdout.write(
             self.style.SUCCESS(

apps/company/migrations/0013_alter_company_allow_jobs.py

@@ -0,0 +1,21 @@
+# Generated by Django 6.0.7 on 2026-08-02 02:00
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("company", "0012_text_unset_constraints"),
+    ]
+
+    operations = [
+        migrations.AlterField(
+            model_name="company",
+            name="allow_jobs",
+            field=models.BooleanField(
+                default=True,
+                help_text="If False, this company cannot be selected as the company on a Job. Use for Xero contacts that must exist (tax authorities, internal accounts, etc.) but should never appear on a job. Automatically set to False when a company is archived or merged in Xero, and restored on un-archive (merged companies stay blocked).",
+            ),
+        ),
+    ]

apps/company/models.py

@@ -98,7 +98,8 @@ class Company(models.Model):
             "If False, this company cannot be selected as the company on a Job. "
             "Use for Xero contacts that must exist (tax authorities, internal "
             "accounts, etc.) but should never appear on a job. Automatically "
-            "set to False when a company is archived or merged in Xero."
+            "set to False when a company is archived or merged in Xero, and "
+            "restored on un-archive (merged companies stay blocked)."
         ),
     )
     xero_last_modified = models.DateTimeField(null=False, blank=False)

apps/company/services/company_rest_service.py

@@ -151,8 +151,11 @@ def list_companies(
             if sort_dir.lower() == "desc":
                 sort_field = f"-{sort_field}"
 
+            # Merged tombstones are excluded: their data lives on the winner
+            # (ADR 0034). They stay reachable by id on the detail endpoint.
             queryset = (
                 Company.objects.with_invoice_summary()
+                .filter(merged_into__isnull=True)
                 .defer("raw_json")
                 .annotate(
                     phone=ContactMethod.primary_phone_annotation(
@@ -164,7 +167,7 @@ def list_companies(
             # Apply search filter if query provided
             if query:
                 ranked_ids = CompanyRestService._rank_matching_company_ids(
-                    Company.objects.all(), query
+                    Company.objects.filter(merged_into__isnull=True), query
                 )
                 total_count = len(ranked_ids)
                 offset = (page - 1) * page_size

apps/company/tests/test_company_fts_search.py

@@ -301,3 +301,21 @@ def test_search_companies_short_query_is_rejected(db):
     """The 3-character minimum-query guard returns [] without raising."""
     assert CompanyRestService.search_companies("ab", limit=10) == []
     assert CompanyRestService.search_companies("", limit=10) == []
+
+
+def test_merged_tombstone_is_excluded_from_browse_and_search(db: None) -> None:
+    """A merged company's data lives on the winner (ADR 0034); listing the
+    tombstone alongside it would show the same business twice."""
+    winner = _make_company("Tombstone Winner Ltd")
+    _make_company("Tombstone Loser Ltd", merged_into=winner, allow_jobs=False)
+
+    browsed = CompanyRestService.list_companies(page=1, page_size=50)
+    browsed_names = [c["name"] for c in browsed["results"]]
+    assert "Tombstone Winner Ltd" in browsed_names
+    assert "Tombstone Loser Ltd" not in browsed_names
+
+    searched = CompanyRestService.list_companies(
+        query="Tombstone", page=1, page_size=50
+    )
+    searched_names = [c["name"] for c in searched["results"]]
+    assert searched_names == ["Tombstone Winner Ltd"]

apps/company/tests/test_company_merge_service.py

@@ -11,6 +11,7 @@
 from decimal import Decimal
 from unittest.mock import patch
 
+from django.core.management import call_command
 from django.db import IntegrityError
 from django.utils import timezone
 
@@ -749,3 +750,29 @@ def test_rolls_back_tombstone_when_reassignment_fails(self) -> None:
         self.source.refresh_from_db()
         self.assertIsNone(self.source.merged_into_id)
         self.assertTrue(self.source.allow_jobs)
+
+
+class MergeCompaniesCommandTests(BaseTestCase):
+    """The management command must produce the same end state as a
+    Xero-driven merge (ADR 0034): a tombstone, never a deletion."""
+
+    def test_command_tombstones_duplicates_and_moves_data(self) -> None:
+        primary = make_company("Dup Co")
+        duplicate = make_company("Dup Co")
+        primary_job = make_job(primary, self.test_staff)
+        duplicate_job = make_job(duplicate, self.test_staff)
+
+        call_command("merge_companies", "--name", "Dup Co", "--auto")
+
+        duplicate.refresh_from_db()  # the old implementation deleted this row
+        self.assertEqual(duplicate.merged_into_id, primary.id)
+        self.assertFalse(duplicate.allow_jobs)
+        primary_job.refresh_from_db()
+        duplicate_job.refresh_from_db()
+        self.assertEqual(primary_job.company_id, primary.id)
+        self.assertEqual(duplicate_job.company_id, primary.id)
+
+        # Re-running must treat the tombstone as resolved, not as a duplicate.
+        call_command("merge_companies", "--name", "Dup Co", "--auto")
+        duplicate.refresh_from_db()
+        self.assertEqual(duplicate.merged_into_id, primary.id)

apps/workflow/api/xero/reprocess_xero.py

@@ -285,20 +285,31 @@ def set_company_fields(company: Company, new_from_xero: bool = False) -> None:
     if xero_contact_id_from_json:
         company.xero_contact_id = xero_contact_id_from_json
 
-    # Check for archived/merged status from raw_json
-    contact_status = raw_json.get("_contact_status", "ACTIVE")
-    if contact_status == "ARCHIVED":
-        company.xero_archived = True
+    # xero_archived mirrors Xero in both directions. allow_jobs follows only
+    # on the transitions — archiving disables it, un-archiving restores it —
+    # never on a steady-state sync, so a manual block on an active company
+    # survives routine syncs (ADR 0034). A merged tombstone stays blocked
+    # regardless: its jobs belong to the winner.
+    # No fallback for a missing status: guessing ACTIVE would un-archive the
+    # company and re-open it for jobs. Malformed data fails this company's
+    # sync (ADR 0015 — consumers stay strict). GDPRREQUEST is deliberately
+    # not accepted: it should never occur on an NZ org, and treating it as
+    # active would re-enable jobs for an erased contact — if it ever appears,
+    # fail loudly and decide its handling then.
+    contact_status = raw_json.get("_contact_status")
+    if contact_status not in ("ACTIVE", "ARCHIVED"):
+        raise ValueError(
+            f"Company {company.id} raw_json has missing or unhandled "
+            f"_contact_status {contact_status!r}"
+        )
+    was_archived = bool(company.xero_archived)
+    company.xero_archived = contact_status == "ARCHIVED"
+    if company.xero_archived:
         company.allow_jobs = False
-        # FIXME: asymmetric -- un-archiving in Xero does not reset either
-        # flag. If a contact is archived then un-archived, `xero_archived`
-        # and `allow_jobs` stay in the archived state until an admin toggles
-        # `allow_jobs` back on via the company detail UI. The un-archive
-        # path is rare enough that we accepted the asymmetry rather than
-        # introduce a "manually set" protection flag. If un-archive becomes
-        # common, revisit: (a) auto-reset both flags, which overwrites any
-        # manual admin-set `allow_jobs=False`; or (b) track admin overrides
-        # separately so they survive a sync.
+    elif was_archived and not company.merged_into_id:
+        company.allow_jobs = True
+    else:
+        pass  # no archive transition; leave allow_jobs as the admin set it
 
     # Check for merge information
     merged_to_contact_id = raw_json.get("_merged_to_contact_id")

apps/workflow/tests/test_sync_clients.py

@@ -150,6 +150,109 @@ def _make_xero_contact(contact_id, name, status="ACTIVE", merged_to=None):
     return contact
 
 
+class UnarchiveResetTests(TestCase):
+    """xero_archived mirrors Xero both ways; allow_jobs follows only on the
+    archive transitions, so a manual block survives routine syncs."""
+
+    def test_unarchive_restores_allow_jobs(self) -> None:
+        company = Company.objects.create(
+            name="Archive Roundtrip Ltd",
+            xero_last_modified=timezone.now(),
+            allow_jobs=True,
+            raw_json=_make_raw_json(
+                "contact-1", "Archive Roundtrip Ltd", status="ARCHIVED"
+            ),
+        )
+
+        set_company_fields(company)
+        self.assertTrue(company.xero_archived)
+        self.assertFalse(company.allow_jobs)
+
+        company.raw_json = _make_raw_json(
+            "contact-1", "Archive Roundtrip Ltd", status="ACTIVE"
+        )
+        set_company_fields(company)
+
+        self.assertFalse(company.xero_archived)
+        self.assertTrue(company.allow_jobs)
+
+    def test_missing_status_fails_rather_than_unarchiving(self) -> None:
+        """A malformed payload with no _contact_status must not be guessed as
+        ACTIVE — that would un-archive the company and re-open it for jobs."""
+        malformed = _make_raw_json("contact-4", "Malformed Status Ltd")
+        del malformed["_contact_status"]
+        company = Company.objects.create(
+            name="Malformed Status Ltd",
+            xero_last_modified=timezone.now(),
+            allow_jobs=False,
+            xero_archived=True,
+            raw_json=malformed,
+        )
+
+        with self.assertRaisesRegex(ValueError, "_contact_status"):
+            set_company_fields(company)
+
+        self.assertTrue(company.xero_archived)
+        self.assertFalse(company.allow_jobs)
+
+    def test_gdprrequest_status_fails_rather_than_reenabling_jobs(self) -> None:
+        """GDPRREQUEST is unhandled: treating an erased contact as active
+        would re-enable jobs for it. It must fail loudly instead."""
+        company = Company.objects.create(
+            name="GDPR Erased Ltd",
+            xero_last_modified=timezone.now(),
+            allow_jobs=False,
+            xero_archived=True,
+            raw_json=_make_raw_json(
+                "contact-5", "GDPR Erased Ltd", status="GDPRREQUEST"
+            ),
+        )
+
+        with self.assertRaisesRegex(ValueError, "_contact_status"):
+            set_company_fields(company)
+
+        self.assertTrue(company.xero_archived)
+        self.assertFalse(company.allow_jobs)
+
+    def test_manual_block_survives_steady_state_sync(self) -> None:
+        """No transition, no touch: an admin's allow_jobs=False on an active
+        company must not be flipped back by a routine sync."""
+        company = Company.objects.create(
+            name="Blocked But Active Ltd",
+            xero_last_modified=timezone.now(),
+            allow_jobs=False,
+            xero_archived=False,
+            raw_json=_make_raw_json(
+                "contact-2", "Blocked But Active Ltd", status="ACTIVE"
+            ),
+        )
+
+        set_company_fields(company)
+
+        self.assertFalse(company.xero_archived)
+        self.assertFalse(company.allow_jobs)
+
+    def test_unarchived_merged_tombstone_stays_blocked(self) -> None:
+        """A merged loser's jobs belong to the winner; un-archiving its Xero
+        contact must not re-open it for jobs."""
+        winner = Company.objects.create(
+            name="Tombstone Winner", xero_last_modified=timezone.now()
+        )
+        loser = Company.objects.create(
+            name="Tombstone Loser",
+            xero_last_modified=timezone.now(),
+            allow_jobs=False,
+            xero_archived=True,
+            merged_into=winner,
+            raw_json=_make_raw_json("contact-3", "Tombstone Loser", status="ACTIVE"),
+        )
+
+        set_company_fields(loser)
+
+        self.assertFalse(loser.xero_archived)
+        self.assertFalse(loser.allow_jobs)
+
+
 class SyncClientsArchivedContactTests(TestCase):
     """Regression tests for archived-contact name collisions during Xero sync.
 
@@ -306,14 +409,15 @@ def _client_with_phone(
             name=name,
             xero_last_modified=timezone.now(),
             raw_json={
+                "_contact_status": "ACTIVE",
                 "_phones": [
                     {
                         "_phone_type": phone_type,
                         "_phone_number": number,
                         "_phone_area_code": "",
                         "_phone_country_code": "",
                     }
-                ]
+                ],
             },
         )