"""MDM scheduled jobs — device sync, and the cleanup todos it closes. ``jamf.device_sync`` keeps the device ledger and the licence-utilisation snapshot in step with the tenant. ``mdm.cleanup_notify`` announces and chases the "this device was disposed of but is still enrolled" todos. Provider-named where the API is provider-specific, provider-neutral where the behaviour is not — a second MDM adds a sync job and reuses the rest. This job used to be ``jamf.seat_sync`` and used to hand out per-person seat allocations, which is how a managed Mac ended up looking like a software licence somebody holds until they resign. It doesn't any more. An enrolment is a property of the device; the job writes ``asset_mdm_devices`` (see ``mdm_device_coordinator``) and nothing about a person's employment changes what it does. What it still feeds the shared reconciliation surface is the licence question, which is genuinely per-device: the tenant reports N enrolled devices against a contract sized for M. That number belongs next to Feishu / Jira / Zoom on the reconciliation page, so the four buckets are projected there from the same classification pass: * matched — an active employee holds it * leaver_residual — a departed employee's name is still on it in Jamf * external — an address that is not a Helix person * unowned — pool / lab / shared, nobody named (count-only) The buckets are a read on the MDM's own data quality, not a work queue. ``leaver_residual`` in particular is "Jamf still says Chris owns this Mac", which resolves itself when the machine is re-imaged for the next hire — it is not a departure that needs actioning, and nothing here raises a todo for it. Per-device dedup keys on the tenant's device identity (``holder_key = device_type:external_id``), so an employee with two Macs is two devices rather than one collapsed row — and a re-enrolment, which leaves two records sharing one serial, does not collapse either. """ from __future__ import annotations import logging import uuid from datetime import UTC, datetime from typing import Any from sqlalchemy.exc import OperationalError from sqlalchemy.ext.asyncio import AsyncSession from app.core.txn import savepoint from app.domains.asset.models import ( MDM_MATCH_EXTERNAL, MDM_MATCH_LEAVER_RESIDUAL, SoftwareExternalHolder, SoftwareSeatSyncStat, ) from app.scheduled_jobs._seat_sync_helpers import ( ExternalHolderObservation, SeatReconciliationIdentityError, raise_if_total_failure, record_seat_reconciliation, write_sync_audit, ) from app.scheduled_jobs.registry import register_scheduled_job logger = logging.getLogger(__name__) @register_scheduled_job( key="jamf.device_sync", display_name="Jamf · 设备同步", description=( "拉取 Jamf 已纳管 Apple 设备写入设备台账(按序列号关联资产、解析当前" "使用人、本次未出现即判定已从 Jamf 移除并自动结案清理待办),并把设备数" "按在职/离职残留/外部/无归属四桶喂给统一对账用于合同用量。" "纳管关系属于设备,不随离职回收。" ), ) async def run_jamf_device_sync( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: from app.scheduled_jobs.handlers import run_with_lock_retry return await run_with_lock_retry( db, lambda session: _run_jamf_device_sync_once(session, config), label="jamf.device_sync", ) async def _run_jamf_device_sync_once( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: from app.core.coordinators.jamf_seat_coordinator import ( get_jamf_reconciliation_plans, load_jamf_driver, ) from app.core.coordinators.mdm_device_coordinator import ( sync_devices_into_ledger, ) now = datetime.now(UTC) summary: dict[str, Any] = { "devices": 0, "ledger_created": 0, "ledger_updated": 0, "marked_gone": 0, "returned": 0, "cleanup_auto_confirmed": 0, "cleanup_reconciled_assets": 0, "cleanup_reconciled_tasks": 0, "plans_reconciled": 0, "per_plan": [], "errors": [], } driver = await load_jamf_driver(db) if driver is None: raise RuntimeError( "Jamf MDM integration (kind='mdm' provider='jamf') not " "configured or not enabled — device sync cannot proceed.", ) integration_id = uuid.UUID(driver.integration_id) devices = await driver.list_devices() # ── 1. the ledger — the authoritative write ─────────────────── # Runs whether or not any software plan is bound to the integration. # The reconciliation snapshot below is a licence-reporting extra; the # device inventory is the point of the job and must not depend on an # admin having wired up a connector. min_ratio, allow_zero_overlap, ratio_note = _resolve_retire_overrides( config, now=now, ) if ratio_note: summary["retire_ratio_note"] = ratio_note logger.warning("jamf.device_sync: %s", ratio_note) outcome, observations = await sync_devices_into_ledger( db, integration_id=integration_id, provider="jamf", devices=devices, now=now, min_ratio=min_ratio, allow_zero_overlap=allow_zero_overlap, ) summary["devices"] = outcome.total_devices summary["roster_entries"] = outcome.roster_entries summary["duplicates_ignored"] = outcome.duplicates_ignored summary["ledger_created"] = outcome.created summary["ledger_updated"] = outcome.updated summary["marked_gone"] = outcome.marked_gone summary["returned"] = outcome.returned summary["cleanup_auto_confirmed"] = outcome.cleanup_auto_confirmed summary["matched"] = outcome.matched summary["leaver_residual"] = outcome.leaver_residual summary["external"] = outcome.external summary["unowned"] = outcome.unowned summary["linked_to_asset"] = outcome.linked_to_asset if outcome.retirement_refused: # A deliberate refusal, and it goes in BOTH places on purpose. The # dedicated key says what happened; ``errors`` is what surfaces it # on the job row an operator actually looks at — because "we chose # not to retire anything tonight" and "nothing needed retiring" # are different facts and only one of them wants investigating, # and a refusal that only ever appears in a summary field is a # tenant frozen behind a green job. ``raise_if_total_failure`` # recognises this class as a defensive refusal rather than a # failure, so it does not page — same posture as the seat-sync # family's ``safe_to_drift_revoke``. summary["retirement_refused"] = outcome.retirement_refused summary["errors"].append( f"jamf.device_sync: {outcome.retirement_refused}", ) # ── 1b. close the cleanup todos the vendor just confirmed ───── summary["cleanup_tasks_closed"] = await _close_confirmed_cleanup_tasks( db, task_ids=outcome.auto_confirmed_task_ids, now=now, ) # ── 1c. put back the closes the roster has overturned ───────── summary["cleanup_tasks_reopened"] = await _reopen_cleanup_tasks( db, task_ids=outcome.cleanup_reopened_task_ids, reason=( "the device reappeared in the MDM tenant — the earlier close " "rested on a disappearance that did not last" ), now=now, ) # A tick-box said the device was removed; the tenant still lists it. # The roster is the fact, so the request goes back on the queue. summary["cleanup_claims_contradicted"] = await _reopen_cleanup_tasks( db, task_ids=outcome.contradicted_claim_task_ids, reason=( "marked done, but the MDM still reports the device — the " "removal did not take" ), now=now, ) # ── 1d. release devices whose todo was cancelled elsewhere ──── summary["cleanup_orphans_released"] = await _release_cancelled_cleanups( db, now=now, ) # ── 1e. raise the cleanups the disposal hook never got to ───── # Runs AFTER the ledger refresh above, because it reads the very # ``asset_id`` that refresh just resolved. Makes the loop converge in # both directions: ``withdraw_stale_cleanups`` already retracted a # request whose asset came back, and until this existed nothing ever # OPENED one for an asset that was disposed while Helix was not # looking — every pre-feature buyback, every hook whose best_effort # savepoint ate a DB error, every writer that sets a disposed status # without going through AssetService. from app.core.coordinators.mdm_device_retirement_coordinator import ( reconcile_disposed_assets_without_cleanup, ) recon = await reconcile_disposed_assets_without_cleanup( db, now=now, trace_id="jamf.device_sync", ) summary["cleanup_reconciled_assets"] = recon["assets"] summary["cleanup_reconciled_tasks"] = recon["tasks_created"] if recon["unroutable"]: # A todo with nobody to file it against is also a todo with nobody # to tell, so it is not written half-formed — but it is a real # gap, and a counter alone is not an error report. summary["errors"].append( f"jamf.device_sync: {recon['unroutable']} disposed asset(s) are " f"still enrolled but have no routable owner — give the MDM " f"software catalogue an active owner, or check the tenant has " f"active global admins" ) # ── 2. licence-utilisation snapshot (optional) ──────────────── plans = await get_jamf_reconciliation_plans(db) if not plans: # No software bound to the Jamf integration — the ledger is still # correct and the run is a success. Recorded as a note rather than # an error so a tenant that never wired up the licence side does # not page anyone every night. summary["reconciliation"] = ( "no active software plan bound to the Jamf integration — " "licence snapshot skipped (device ledger unaffected)" ) else: # One observation per device that deviates from the baseline. # ``matched`` and ``unowned`` are count-only by the shared # helper's contract: the first is "everything is fine" and the # second has no holder identity to annotate. externals: list[ExternalHolderObservation] = [] for obs in observations: if obs.category not in ( MDM_MATCH_LEAVER_RESIDUAL, MDM_MATCH_EXTERNAL, ): continue row = obs.row externals.append(ExternalHolderObservation( # The resolved address, not Jamf's raw field: a leaver's # Mac that Jamf could not attribute is still annotated # against the person the asset ledger named, which is how # the cross-vendor registry recognises the same account. # # ``""`` for the addressless case, same convention as # jira / feishu. It is not quite unreachable: a # ``leaver_residual`` whose holder came from the asset # ledger rather than from Jamf's own field carries only # whatever ``work_email`` holds, and that column is NOT # NULL but not non-blank. Emitted rather than skipped — # the recorder normalises empty to NULL and the row is # keyed on ``holder_key`` below, whereas dropping the # observation would unbalance the per-bucket identity # check, which is fail-loud for this provider and would # roll back the whole ledger write. email=obs.owner_email or "", # The TENANT's device identity, not the serial. Per-device # keying is the point — a two-Mac employee must be two # seats — but the serial is not reliably unique: a # re-enrolment leaves two records for one machine (the # ledger's own index is deliberately non-unique on serial # for exactly that reason), and Jamf numbers computers and # mobiles in separate sequences. Either collision makes the # shared recorder dedup two devices into one, which fails # the reconciliation identity — and because that check is # fail-loud for this provider, it would roll back the # entire ledger write this job exists to make. holder_key=f"{row.device_type}:{row.external_id}".lower(), vendor_account_id=row.external_id, account_status="managed" if row.managed else "unmanaged", display_name=( row.device_name or row.model or row.serial_number ), category=obs.category, person_id=row.assigned_person_id, )) for plan in plans: plan_name = getattr(plan, "name", str(plan.id)) try: # SAVEPOINT per plan: the generic handler below swallows # and continues, and the run then writes an audit row and # commits. Without isolation a DB failure on one plan # poisons the session, every later plan dies on its first # query, and the trailing commit discards the ledger write # this job exists to make. async with savepoint(db, label=f"jamf.device_sync {plan_name}"): await record_seat_reconciliation( db, holder_model=SoftwareExternalHolder, stat_model=SoftwareSeatSyncStat, integration_id=integration_id, software_id=plan.software_id, plan_id=plan.id, provider="jamf", product_key=None, site=None, vendor_license_holders=outcome.total_devices, holders_matched=outcome.matched, holders_no_email=outcome.unowned, externals=externals, holders_leaver_residual=outcome.leaver_residual, now=now, ) summary["plans_reconciled"] += 1 summary["per_plan"].append({ "plan": plan_name, "devices": outcome.total_devices, "matched": outcome.matched, "leaver_residual": outcome.leaver_residual, "external": outcome.external, "unowned": outcome.unowned, }) except OperationalError: raise except SeatReconciliationIdentityError: # The four buckets are assigned by a single pass over the # roster and are mutually exclusive, so they sum to the # device count by construction. A violation can only be a # real counting bug — fail loud rather than bury it in # summary["errors"] (unlike jira.seat_sync, where # shared-email dedup can legitimately unbalance). raise except Exception as e: # noqa: BLE001 summary["errors"].append( f"{plan_name}: {type(e).__name__}: {e}", ) await write_sync_audit(db, job_key="jamf.device_sync", summary=summary) await db.commit() # Gate on real mutation counters only. ``plans_reconciled`` is # structural — it advances for any plan that finishes iterating, so # including it would let a run where every write failed still record # status="ok" and never page. raise_if_total_failure( job_key="jamf.device_sync", summary=summary, progress_keys=( "ledger_created", "ledger_updated", "marked_gone", # A run whose only real mutation was raising a backlog # cleanup did do work; leaving it out would let that run be # judged a total failure and page on a success. "cleanup_reconciled_tasks", ), ) return summary @register_scheduled_job( key="mdm.cleanup_notify", display_name="MDM · 设备清理待办触达", description=( "把「资产已处置但设备仍在 MDM 中」的清理待办发给 MDM 软件目录的负责人" "(负责人缺失或已离职则扇出给管理员):新待办首次发卡,超过 7 天未处理" "每周催办一次,超过 21 天同时抄送全部管理员。设备从 MDM 消失即自动结案," "不再催。仅在工作时段运行,避免深夜打扰。" ), ) async def run_mdm_cleanup_notify( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: """Announce and chase the MDM cleanups nobody has closed yet.""" from app.scheduled_jobs.handlers import run_with_lock_retry return await run_with_lock_retry( db, lambda session: _run_mdm_cleanup_notify_once(session, config), label="mdm.cleanup_notify", ) async def _run_mdm_cleanup_notify_once( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: """One tick. Deliberately a separate job from the disposal itself. Disposal runs inside the caller's transaction — sometimes an approval callback holding row locks — and delivering a Feishu card from there would put an external round-trip inside an open transaction. So disposal writes the todo and this job does the talking, which also means the first card and the weekly nudge travel the same, single, retried path instead of one being a special case of the other. """ from app.core.coordinators.mdm_device_retirement_coordinator import ( NOTIFY_BURST_CAP, deliver_enqueued, due_cleanup_notifications, notify_cleanup, withdraw_stale_cleanups, ) now = datetime.now(UTC) summary: dict[str, Any] = { "due": 0, "first_notice": 0, "reminders": 0, "deferred": 0, "withdrawn": 0, "recipients": 0, "delivered": 0, "errors": [], } # Re-check the premise before speaking. A request whose asset is no # longer disposed would have an administrator delete a machine the # company still owns — so the queue verifies its own reason to exist # rather than trusting every future writer of the asset status to # remember this queue. summary["withdrawn"] = await withdraw_stale_cleanups(db, now=now) due = await due_cleanup_notifications(db, now=now) summary["due"] = len(due) # Bounded fan-out. ``due`` arrives oldest-waiting first, so taking a # prefix defers the least-overdue rather than an arbitrary subset, # and the next of the day's ten ticks picks them up. The count is # REPORTED: a silent cap reads as "5 due, 5 sent, all healthy" while # a backlog sits behind it, which is the same shape as a truncation # that calls itself full coverage. cap = _positive_int(config.get("notify_burst_cap"), NOTIFY_BURST_CAP) if len(due) > cap: summary["deferred"] = len(due) - cap summary["note"] = ( f"{len(due)} cleanups due, sending {cap} this tick and " f"deferring {len(due) - cap} to the next one (oldest first) — " f"raise `notify_burst_cap` on the job config to widen" ) due = due[:cap] # ── phase 1: write the outbox rows + reminder stamps, then commit ── # Nothing here talks to Feishu. Committing the stamp together with the # intent-to-send is what makes a crash between them impossible to turn # into a duplicate card on the next tick. queued: list[tuple[Any, Any]] = [] for device, task, is_reminder in due: try: # Per-device SAVEPOINT: one device whose enqueue hits a DB # error must not poison the session and silence every device # behind it in the batch. async with savepoint(db, label=f"mdm.cleanup_notify {device.id}"): result = await notify_cleanup( db, device=device, task=task, trace_id=f"mdm.cleanup_notify:{device.id}", is_reminder=is_reminder, now=now, ) summary["recipients"] += len(result.queued) queued.extend(result.queued) if result.problem: # A cleanup nobody can be told about is a real gap, and a # run that quietly does nothing about it is indis- # tinguishable from a run with nothing to do. summary["errors"].append(result.problem) if result.queued: if is_reminder: summary["reminders"] += 1 else: summary["first_notice"] += 1 except OperationalError: raise except Exception as e: # noqa: BLE001 summary["errors"].append(f"{device.id}: {type(e).__name__}: {e}") await db.commit() # ── phase 2: deliver, outside any transaction ───────────────────── # Best-effort. Every row above is durably PENDING, so whatever this # drain misses is delivered by ``notification.outbox_retry``. summary["delivered"] = await deliver_enqueued( db, queued=queued, trace_id="mdm.cleanup_notify", ) raise_if_total_failure( job_key="mdm.cleanup_notify", summary=summary, # ``deferred`` is deliberately NOT progress: a tick that only # deferred did no work, and counting it would let a permanently # blocked queue report success every time. progress_keys=("first_notice", "reminders", "withdrawn"), ) return summary async def _close_confirmed_cleanup_tasks( db: AsyncSession, *, task_ids: list[uuid.UUID], now: datetime, ) -> int: """Complete the cleanup todos whose devices have left the tenant. The ledger sync has just stamped ``cleanup_confirmed_via='sync'`` on every device that disappeared while a cleanup was open. That is the vendor confirming the removal, which is stronger evidence than the button on the notification card — so the todo is closed here rather than left waiting for a human to also tick it. """ if not task_ids: return 0 from sqlalchemy import select from app.audit.service import AuditService from app.domains.provisioning.models import ( STATUS_CANCELLED, STATUS_COMPLETED, ProvisioningTask, ) tasks = ( await db.execute( select(ProvisioningTask).where( ProvisioningTask.id.in_(set(task_ids)), ProvisioningTask.status.notin_( (STATUS_COMPLETED, STATUS_CANCELLED), ), ) ) ).scalars().all() audit = AuditService(db) for task in tasks: task.status = STATUS_COMPLETED task.completed_at = now task.external_ref = "confirmed by Jamf inventory (device removed)" # This path bypasses ``complete_manual_task``, so it has to write # its own audit row. It is also the path that makes the STRONGEST # claim on the ledger — "the vendor confirmed this" — and a claim # that strong with no trail behind it is the one an auditor will # ask about first (CLAUDE.md invariant 6). await audit.log( trace_id=f"jamf.device_sync:{task.id}", actor_type="system", actor_user_id=None, action="mdm.cleanup.auto_confirmed", status="success", resource_type="provisioning_task", resource_id=str(task.id), detail={ "device_id": str(task.target_id) if task.target_id else None, "serial_number": (task.context_data or {}).get( "serial_number", ), "closed_by": "jamf.device_sync", "evidence": "device no longer reported by the MDM tenant", }, ) await db.flush() return len(tasks) def _positive_int(value: object, default: int) -> int: """A job-config integer, or the default when it is not usable. Validated rather than trusted, for the reason the retire overrides were: ``job.config`` is a persisted row a human edits by hand, and a silently-accepted ``0`` here would stop every cleanup card without changing anything an operator would think to look at. ``bool`` is rejected explicitly — ``True`` is an ``int`` in Python and would read as a cap of one. """ if isinstance(value, bool) or not isinstance(value, (int, str)): return default try: parsed = int(str(value).strip()) except (TypeError, ValueError): return default return parsed if parsed > 0 else default def _resolve_retire_overrides( config: dict[str, Any], *, now: datetime, ) -> tuple[float, bool, str | None]: """The retirement floor for this run, the fleet-swap authorisation, why. Two real mass events exist and they are different. An office closes and thirty machines are scrapped in one night — that is a ratio question. A tenant is migrated and every device comes back with a new id — that is "retire devices the roster no longer mentions at all", which no ratio expresses. The guard refuses both, and then refuses them again every night after, because the refusal is precisely what stops those machines from ever being marked gone. So each needs a way through, and they need SEPARATE ones: deriving the second from the first said yes to a TIGHTENING ratio (0.9 unlocked the whole-fleet bypass this guard exists to hold shut) and said no to a deliberate override that happened to equal the default. Both ways through **expire by themselves**, on one shared ``retire_override_until`` (ISO date or datetime). Neither is honoured without it. Job config is a persisted row, so an override with no expiry is not "for one run" however it is worded — it is a safety floor quietly lowered forever, discovered the next time something goes wrong. Self-expiry means forgetting to put it back costs nothing. """ from app.core.coordinators.mdm_device_coordinator import RETIRE_MIN_RATIO raw_ratio = config.get("retire_min_ratio") raw_zero = config.get("retire_allow_zero_overlap") if raw_ratio is None and not raw_zero: return RETIRE_MIN_RATIO, False, None ratio = RETIRE_MIN_RATIO if raw_ratio is not None: try: ratio = float(raw_ratio) except (TypeError, ValueError): return RETIRE_MIN_RATIO, False, ( f"retire_min_ratio={raw_ratio!r} is not a number — ignoring " "every retirement override and using the default floor." ) if not 0 < ratio <= 1: return RETIRE_MIN_RATIO, False, ( f"retire_min_ratio={ratio} is outside (0, 1] — ignoring " "every retirement override and using the default floor." ) until_raw = config.get("retire_override_until") if not until_raw: return RETIRE_MIN_RATIO, False, ( "a retirement override is set but retire_override_until is not " "— ignoring it. An override with no expiry is not a one-run " "exception, it is a permanently weaker guard that nobody will " "remember to restore." ) try: until = datetime.fromisoformat(str(until_raw)) except ValueError: return RETIRE_MIN_RATIO, False, ( f"retire_override_until={until_raw!r} is not an ISO date — " "ignoring every retirement override." ) if until.tzinfo is None: until = until.replace(tzinfo=UTC) if until <= now: return RETIRE_MIN_RATIO, False, ( f"the retirement override expired at {until.isoformat()} — back " "to the default. Remove retire_min_ratio / " "retire_allow_zero_overlap / retire_override_until from the job " "config to silence this." ) from app.core.coordinators.mdm_device_coordinator import coerce_bool allow_zero_overlap = coerce_bool(raw_zero) parts = [] if raw_ratio is not None: parts.append(f"floor lowered to {ratio:.0%}") if allow_zero_overlap: parts.append("retiring a roster with NO known devices authorised") return ratio, allow_zero_overlap, ( f"retirement override in force until {until.isoformat()}: " + "; ".join(parts) + " — the default returns on its own after that." ) async def _release_cancelled_cleanups( db: AsyncSession, *, now: datetime, ) -> int: """Un-stick devices whose cleanup todo was cancelled from outside. ``POST /provisioning/tasks/{id}/cancel`` is a global-admin action on the generic queue that knows nothing about this ledger. Cancelling a hardware cleanup there leaves the device ``pending_cleanup`` pointing at a terminal task: the reminder query skips terminal tasks, the sync will not re-open a deliberate cancel, and the page shows an open request nobody is chasing. Reconciling here treats the cancel as what it is — a decision that this device does not need removing — rather than leaving two domains disagreeing in silence. """ from sqlalchemy import select from app.audit.service import AuditService from app.domains.asset.models import ( MDM_STATE_ENROLLED, MDM_STATE_PENDING_CLEANUP, MdmManagedDevice, ) from app.domains.provisioning.models import ( STATUS_CANCELLED, ProvisioningTask, ) rows = ( await db.execute( select(MdmManagedDevice, ProvisioningTask) .join( ProvisioningTask, ProvisioningTask.id == MdmManagedDevice.cleanup_task_id, ) .where( MdmManagedDevice.enrollment_state == MDM_STATE_PENDING_CLEANUP, ProvisioningTask.status == STATUS_CANCELLED, ) ) ).all() if not rows: return 0 audit = AuditService(db) released = 0 for device, task in rows: device.enrollment_state = MDM_STATE_ENROLLED device.cleanup_task_id = None device.cleanup_requested_at = None device.cleanup_reason = None device.cleanup_confirmed_at = None device.cleanup_confirmed_via = None device.cleanup_contradicted_at = None device.cleanup_last_reminded_at = None released += 1 await audit.log( trace_id=f"jamf.device_sync:{device.id}", actor_type="system", actor_user_id=None, action="mdm.cleanup.released", status="success", resource_type="asset_mdm_device", resource_id=str(device.id), detail={ "serial_number": device.serial_number, "cancelled_task_id": str(task.id), "reason": ( "the cleanup todo was cancelled on the provisioning " "queue — the device is released rather than left " "pending against a terminal task" ), }, ) await db.flush() return released async def _reopen_cleanup_tasks( db: AsyncSession, *, task_ids: list[uuid.UUID], reason: str, now: datetime, ) -> int: """Put a wrongly-closed cleanup back in play, as a NEW todo. Two ways a close turns out to be wrong, both established by the same authority — the tenant's own roster: * the device vanished for a night and came back, so the "vendor confirmed the removal" close rested on a disappearance that did not last; * an admin ticked "I removed it" and the next sync still finds the device, so the claim is contradicted by the fact. Either way the disposal behind the request still stands and the machine is still enrolled, so the request is live again. Leaving it closed loses it permanently: nothing re-scans historical disposals. A FRESH task rather than flipping the old one back out of ``completed`` — and that is a security property, not tidiness. The one-click confirm link in each card is a signed token bound to a task id, valid for two weeks and NOT single-use. Recycling the id would leave the card that closed this cleanup the first time able to close it again from somebody's Feishu history, silently re-burying the machine the sync just told us is still there. Against a new id that stale token addresses a terminal task and does nothing. """ if not task_ids: return 0 from sqlalchemy import select from app.audit.service import AuditService from app.core.coordinators.mdm_device_retirement_coordinator import ( _resolve_assignee, ) from app.domains.asset.models import MdmManagedDevice from app.domains.provisioning.models import ( STATUS_CANCELLED, STATUS_NOTIFIED, ProvisioningTask, ) tasks = ( await db.execute( select(ProvisioningTask).where( ProvisioningTask.id.in_(set(task_ids)), # A cleanup an admin deliberately cancelled stays cancelled # — only a close the evidence has overturned is reversed. ProvisioningTask.status != STATUS_CANCELLED, ) ) ).scalars().all() audit = AuditService(db) reopened = 0 for old in tasks: if old.status == STATUS_NOTIFIED: # Already open; the claim/return handling upstream cleared the # device stamps and there is nothing to re-raise. continue # Re-resolved, not inherited. ``_resolve_assignee`` returns None # when the catalogue owner has left, so the card fans out to # admins; a task reopened months later would otherwise name a # departed person as the one responsible. assignee_id = await _resolve_assignee(db, old.software_id) fresh = ProvisioningTask( user_id=old.user_id, software_id=old.software_id, action=old.action, target_type=old.target_type, target_id=old.target_id, source=old.source, execution_mode=old.execution_mode, status=STATUS_NOTIFIED, assignee_id=assignee_id, context_data={ **(old.context_data or {}), "reopened_from_task_id": str(old.id), "reopened_reason": reason, }, ) db.add(fresh) await db.flush() old.status = STATUS_CANCELLED old.completed_at = now old.error_message = ( f"superseded by {fresh.id}: {reason}" )[:2000] # Located by the relation being repaired, not by ``target_id``. # If this repoint is missed, the device still points at a task # now CANCELLED — and ``_release_cancelled_cleanups``, later in # this same tick, would wipe the cleanup we just re-raised and # orphan the fresh task where nothing device-driven finds it. pointed = ( await db.execute( select(MdmManagedDevice).where( MdmManagedDevice.cleanup_task_id == old.id, ) ) ).scalars().all() if not pointed: logger.error( 'mdm cleanup: no device points at task %s — cannot ' 'repoint to the replacement %s; abandoning the reopen ' 'rather than orphaning it', old.id, fresh.id, ) fresh.status = STATUS_CANCELLED fresh.completed_at = now fresh.error_message = ( 'reopen abandoned: no device row referenced task ' f'{old.id}' ) # The old task keeps whatever state it had — saying it was # 'superseded by' a task that is itself cancelled would be a # false trail. And this branch audits like its four siblings: # a state transition nobody can reconstruct is the one an # auditor asks about. await audit.log( trace_id=f'jamf.device_sync:{old.id}', actor_type='system', actor_user_id=None, action='mdm.cleanup.reopen_abandoned', status='failure', resource_type='provisioning_task', resource_id=str(old.id), detail={ 'reason': ( 'no device row referenced this task, so the ' 'replacement could not be reached by anything ' 'device-driven' ), 'abandoned_replacement_id': str(fresh.id), }, ) continue for device in pointed: device.cleanup_task_id = fresh.id # The one-shot that makes the next tick speak, set here so # it cannot be separated from the marker above that decides # how that card reads. Self-limiting: the notify loop writes # the stamp back as soon as it sends. device.cleanup_last_reminded_at = None reopened += 1 await audit.log( trace_id=f"jamf.device_sync:{fresh.id}", actor_type="system", actor_user_id=None, action="mdm.cleanup.reopened", status="success", resource_type="provisioning_task", resource_id=str(fresh.id), detail={ "device_id": str(old.target_id) if old.target_id else None, "serial_number": (old.context_data or {}).get( "serial_number", ), "superseded_task_id": str(old.id), "reason": reason, }, ) await db.flush() return reopened