Automating Checklist Synchronization Between EDC and CTMS

Site-activation checklists live in both your EDC and your CTMS, and they drift apart the moment a coordinator updates one without the other. This guide builds a production-ready Python synchronization engine that reconciles checklist state across both systems with explicit source-of-truth precedence, conflict detection, idempotent upserts, resilient API clients, and a tamper-evident 21 CFR Part 11 audit trail.

The pattern below is the deep, code-first companion to Checklist Sync & Gap Analysis, which sits within the end-to-end Automated Document Ingestion & Validation Workflows architecture. If your sync runs against portals that intermittently fail, pair it with configuring fallback routing when clinical portals timeout; for the bulk-ingestion side of the same problem, see handling async batch processing for multi-site document ingestion; and when an incoming status enum is unknown, route it to Schema Validation & Error Categorization rather than guessing.

Why EDC and CTMS checklists diverge #

An EDC (Electronic Data Capture) system records the operational truth of what a site has actually completed — regulatory documents uploaded, training acknowledged, lab certifications attached. A CTMS (Clinical Trial Management System) tracks the milestone view used by study managers and sponsors for site-activation reporting. The same logical fact — “site 1042 has a fully executed FDA Form 1572” — exists as a checklist item in both systems, with different field names, different enumerations, and independent edit histories.

Divergence is structural, not exceptional:

• Independent writes. A CRA marks an item complete in the CTMS during a monitoring visit while the underlying EDC record is still in_review.
• Taxonomy skew. The EDC exposes status: SUBMITTED; the CTMS expects state: "Pending QC". A mapping layer must normalize both into one canonical vocabulary.
• Clock skew and ordering. Updates carry timestamps from different servers. You cannot trust wall-clock order alone to decide which write is newer.
• Partial failures. A batch upsert commits some items, then hits a rate limit, leaving the two systems in a state that is neither fully synced nor cleanly rolled back.

Treating this as a naive two-way copy produces flapping records and silent data loss. The correct framing is reconciliation against a declared source of truth, with every divergence either resolved deterministically or quarantined as a conflict for a human to adjudicate.

Reconciliation model and source-of-truth precedence #

Define one canonical checklist record, then compute the desired state from both systems according to an explicit precedence policy. A field-level precedence map is more honest than a blanket “EDC always wins,” because operational reality is mixed: completion evidence is authoritative in the EDC, but milestone sign-off and projected dates are owned by study managers in the CTMS.

For each item we classify the pair of records into one of four reconciliation outcomes:

Conflict detection needs a baseline. We store the canonical hash from the last successful sync per item; a side “changed” if its current hash differs from that baseline. If both sides changed fields they each own, no precedence rule can safely auto-resolve it — that is a genuine conflict and must be surfaced, never silently overwritten.

The reconciliation decision is deterministic given three inputs — the EDC record, the CTMS record, and the stored baseline:

$\text{decision} = f(h_{\text{edc}},\; h_{\text{ctms}},\; h_{\text{base}})$

where each $h$ is a canonical SHA-256 digest over the normalized, precedence-relevant fields. Hashing a normalized projection (sorted keys, compact separators, only owned fields) eliminates false positives from whitespace, key ordering, or fields neither system controls.

Architecture overview #

The whole run is a deterministic pipeline: fetch both sides, normalize each item into one canonical shape, hash the precedence-relevant fields, compare against the stored baseline, then route to an idempotent upsert or a conflict quarantine — appending an audit entry for every outcome, including no-ops.

Setup and configuration #

Install the async HTTP client, validation, and retry stack. Everything here is a maintained package — no deprecated PDF or HTTP libraries are involved on this path:

pip install "pydantic>=2.7" "aiohttp>=3.9" "tenacity>=8.3"

Both API clients read their base URL and bearer token from the environment, so no credential is ever committed or baked into an image. Missing secrets fail fast at start-up rather than mid-run:

Initialise structured, single-line logging once at process start so every reconciliation decision is machine-parseable and timestamped for the audit trail:

import logging
import os
import sys

logging.basicConfig(
    level=os.environ.get("SYNC_LOG_LEVEL", "INFO"),
    format="%(message)s",
    stream=sys.stdout,
)
logger = logging.getLogger("edc_ctms_sync")

Canonical models with pydantic #

We model the canonical item and the reconciliation result with pydantic, which gives us validation, coercion, and a single place to enforce required regulatory fields. The FIELD_OWNER map encodes precedence per field.

from __future__ import annotations

import hashlib
import json
from datetime import datetime, timezone
from enum import Enum
from typing import Optional

from pydantic import BaseModel, Field, field_validator


class System(str, Enum):
    EDC = "edc"
    CTMS = "ctms"


class ItemStatus(str, Enum):
    """Canonical status vocabulary that both systems map into."""

    NOT_STARTED = "not_started"
    IN_REVIEW = "in_review"
    COMPLETE = "complete"
    REJECTED = "rejected"


# Field-level source of truth. Completion evidence is owned by the EDC;
# milestone sign-off and planned dates are owned by study managers in the CTMS.
FIELD_OWNER: dict[str, System] = {
    "status": System.EDC,
    "evidence_doc_id": System.EDC,
    "milestone_signed_off": System.CTMS,
    "planned_activation_date": System.CTMS,
}


class CanonicalItem(BaseModel):
    """One checklist item expressed in the canonical vocabulary."""

    site_id: str
    item_code: str
    checklist_version: str
    status: ItemStatus
    evidence_doc_id: Optional[str] = None
    milestone_signed_off: bool = False
    planned_activation_date: Optional[str] = None  # ISO-8601 date
    source_updated_utc: datetime
    operator_id: str = Field(..., min_length=1)

    @field_validator("source_updated_utc")
    @classmethod
    def _require_tz(cls, value: datetime) -> datetime:
        """Reject naive timestamps; ALCOA+ requires unambiguous time."""
        if value.tzinfo is None:
            raise ValueError("source_updated_utc must be timezone-aware")
        return value.astimezone(timezone.utc)

    def owned_projection(self, owner: System) -> dict[str, object]:
        """The subset of fields a given system is the source of truth for."""
        return {
            name: getattr(self, name)
            for name, field_owner in FIELD_OWNER.items()
            if field_owner is owner
        }

    def canonical_hash(self) -> str:
        """Stable digest over all precedence-relevant fields."""
        relevant = {name: getattr(self, name) for name in FIELD_OWNER}
        payload = json.dumps(
            relevant, sort_keys=True, separators=(",", ":"), default=str
        )
        return hashlib.sha256(payload.encode("utf-8")).hexdigest()


class Decision(str, Enum):
    IN_SYNC = "in_sync"
    EDC_AUTHORITATIVE = "edc_authoritative"
    CTMS_AUTHORITATIVE = "ctms_authoritative"
    CONFLICT = "conflict"


class Reconciliation(BaseModel):
    site_id: str
    item_code: str
    decision: Decision
    target: Optional[System] = None
    desired: Optional[CanonicalItem] = None
    reason: str = ""

The reconciliation function #

Reconciliation is pure: given two canonical items and the stored baseline hash, it returns a Reconciliation with no side effects. Keeping it pure makes it trivially unit-testable and removes any temptation to mutate state mid-decision.

def reconcile(
    edc: CanonicalItem,
    ctms: CanonicalItem,
    baseline_hash: Optional[str],
) -> Reconciliation:
    """Decide how to reconcile one item using field-level precedence.

    A side is considered "changed" when its current hash differs from the
    baseline captured at the last successful sync. When both sides changed
    fields they each own, no precedence rule can resolve it safely.
    """
    edc_hash = edc.canonical_hash()
    ctms_hash = ctms.canonical_hash()

    if edc_hash == ctms_hash:
        return Reconciliation(
            site_id=edc.site_id,
            item_code=edc.item_code,
            decision=Decision.IN_SYNC,
        )

    edc_changed = baseline_hash is None or edc_hash != baseline_hash
    ctms_changed = baseline_hash is None or ctms_hash != baseline_hash

    # Did the divergence touch EDC-owned fields, CTMS-owned fields, or both?
    edc_owns_diff = edc.owned_projection(System.EDC) != ctms.owned_projection(System.EDC)
    ctms_owns_diff = edc.owned_projection(System.CTMS) != ctms.owned_projection(System.CTMS)

    if edc_owns_diff and ctms_owns_diff and edc_changed and ctms_changed:
        return Reconciliation(
            site_id=edc.site_id,
            item_code=edc.item_code,
            decision=Decision.CONFLICT,
            reason="Both systems changed owned fields since last sync",
        )

    # Build the desired canonical record: take each field from its owner.
    desired = edc.model_copy(
        update={
            "milestone_signed_off": ctms.milestone_signed_off,
            "planned_activation_date": ctms.planned_activation_date,
        }
    )

    if edc_owns_diff:
        return Reconciliation(
            site_id=edc.site_id,
            item_code=edc.item_code,
            decision=Decision.EDC_AUTHORITATIVE,
            target=System.CTMS,
            desired=desired,
            reason="EDC-owned fields changed",
        )

    return Reconciliation(
        site_id=edc.site_id,
        item_code=edc.item_code,
        decision=Decision.CTMS_AUTHORITATIVE,
        target=System.EDC,
        desired=desired,
        reason="CTMS-owned fields changed",
    )

A resilient async API client #

EDC and CTMS APIs are remote, paginated, and rate-limited. A correct client must: read credentials from the environment (never hardcode them), retry transient failures with exponential backoff and jitter, respect server Retry-After headers on 429, paginate fully, and cap concurrency with a semaphore so one batch never stampedes the server. We use aiohttp and tenacity.

import asyncio
import os
import random
from collections.abc import AsyncIterator, Callable
from typing import Any

import aiohttp
from tenacity import (
    retry,
    retry_if_exception_type,
    stop_after_attempt,
    wait_exponential_jitter,
)


class RetryableHTTPError(Exception):
    """Raised for status codes that warrant a retry (429 / 5xx)."""


class ClinicalApiClient:
    """Async client for an EDC or CTMS REST API.

    Secrets are read from the environment. Set, for the EDC:
        EDC_BASE_URL, EDC_API_TOKEN
    and for the CTMS:
        CTMS_BASE_URL, CTMS_API_TOKEN
    """

    def __init__(self, system: System, *, max_concurrency: int = 8) -> None:
        prefix = "EDC" if system is System.EDC else "CTMS"
        try:
            self._base_url = os.environ[f"{prefix}_BASE_URL"].rstrip("/")
            self._token = os.environ[f"{prefix}_API_TOKEN"]
        except KeyError as exc:
            raise RuntimeError(f"Missing required environment variable: {exc}") from exc
        self.system = system
        self._semaphore = asyncio.Semaphore(max_concurrency)
        self._session: aiohttp.ClientSession | None = None

    async def __aenter__(self) -> "ClinicalApiClient":
        self._session = aiohttp.ClientSession(
            base_url=self._base_url,
            headers={
                "Authorization": f"Bearer {self._token}",
                "Accept": "application/json",
            },
            timeout=aiohttp.ClientTimeout(total=30),
            connector=aiohttp.TCPConnector(limit=100, limit_per_host=10),
        )
        return self

    async def __aexit__(self, *exc: object) -> None:
        if self._session is not None:
            await self._session.close()

    @retry(
        retry=retry_if_exception_type((RetryableHTTPError, aiohttp.ClientConnectionError)),
        wait=wait_exponential_jitter(initial=0.5, max=20.0),
        stop=stop_after_attempt(5),
        reraise=True,
    )
    async def _request(self, method: str, path: str, **kwargs: Any) -> dict[str, Any]:
        assert self._session is not None, "client must be used as an async context manager"
        async with self._semaphore:
            async with self._session.request(method, path, **kwargs) as resp:
                if resp.status == 429:
                    delay = float(resp.headers.get("Retry-After", "1"))
                    await asyncio.sleep(delay + random.uniform(0, 0.5))
                    raise RetryableHTTPError("rate limited")
                if resp.status >= 500:
                    raise RetryableHTTPError(f"server error {resp.status}")
                resp.raise_for_status()  # raises ClientResponseError for 4xx
                return await resp.json()

    async def iter_checklist(self, study_id: str) -> AsyncIterator[dict[str, Any]]:
        """Yield raw checklist items across all pages (cursor pagination)."""
        cursor: str | None = None
        while True:
            params: dict[str, str] = {"study_id": study_id, "limit": "200"}
            if cursor:
                params["cursor"] = cursor
            page = await self._request("GET", "/v1/checklist-items", params=params)
            for item in page.get("items", []):
                yield item
            cursor = page.get("next_cursor")
            if not cursor:
                return

    async def upsert_item(
        self, payload: dict[str, Any], idempotency_key: str
    ) -> dict[str, Any]:
        """Idempotent create-or-update keyed by a deterministic hash."""
        return await self._request(
            "PUT",
            f"/v1/checklist-items/{payload['item_code']}",
            json=payload,
            headers={"Idempotency-Key": idempotency_key},
        )

PUT to a deterministic resource path (/checklist-items/{item_code}) is naturally idempotent, and the Idempotency-Key header lets the server collapse retried writes that carry the same canonical payload. The key is derived from the payload hash, so an accidental double-send never produces a second audit event or a duplicate milestone.

Mapping each system into the canonical model #

Each system gets a small adapter that translates its native shape and enums into a CanonicalItem. Centralizing the vocabulary here is what defeats taxonomy skew — the reconciliation logic only ever sees canonical values.

from datetime import date

_EDC_STATUS = {
    "DRAFT": ItemStatus.NOT_STARTED,
    "SUBMITTED": ItemStatus.IN_REVIEW,
    "APPROVED": ItemStatus.COMPLETE,
    "RETURNED": ItemStatus.REJECTED,
}

_CTMS_STATUS = {
    "Pending": ItemStatus.NOT_STARTED,
    "Pending QC": ItemStatus.IN_REVIEW,
    "Verified": ItemStatus.COMPLETE,
    "Rework": ItemStatus.REJECTED,
}


def from_edc(raw: dict[str, Any]) -> CanonicalItem:
    return CanonicalItem(
        site_id=str(raw["siteId"]),
        item_code=str(raw["code"]),
        checklist_version=str(raw["version"]),
        status=_EDC_STATUS[raw["status"]],
        evidence_doc_id=raw.get("documentId"),
        source_updated_utc=datetime.fromisoformat(raw["updatedAt"]),
        operator_id=str(raw["lastEditedBy"]),
    )


def from_ctms(raw: dict[str, Any]) -> CanonicalItem:
    planned = raw.get("plannedActivation")
    return CanonicalItem(
        site_id=str(raw["site"]),
        item_code=str(raw["taskCode"]),
        checklist_version=str(raw["formVersion"]),
        status=_CTMS_STATUS[raw["state"]],
        milestone_signed_off=bool(raw.get("signedOff", False)),
        planned_activation_date=date.fromisoformat(planned).isoformat() if planned else None,
        source_updated_utc=datetime.fromisoformat(raw["modifiedUtc"]),
        operator_id=str(raw["modifiedBy"]),
    )

A missing or unexpected status raises a KeyError here rather than silently defaulting — an unknown enum is a mapping defect that belongs in categorizing validation errors in regulatory document pipelines, not a value to guess at.

Tamper-evident audit trail (21 CFR Part 11 / ALCOA+) #

Every sync decision — including no-ops and conflicts — produces an audit entry. To make the trail tamper-evident we hash-chain entries: each entry includes the hash of the previous one, so any retroactive edit breaks the chain from that point forward. This satisfies the 21 CFR Part 11 expectation of a secure, computer-generated, time-stamped audit trail and the ALCOA+ attributes (Attributable, Legible, Contemporaneous, Original, Accurate, Complete, Consistent, Enduring, Available).

class AuditEntry(BaseModel):
    sequence: int
    correlation_id: str
    site_id: str
    item_code: str
    decision: Decision
    target: Optional[System]
    payload_hash: str
    operator_id: str
    recorded_utc: datetime
    previous_hash: str
    entry_hash: str = ""

    def compute_hash(self) -> str:
        body = self.model_dump(exclude={"entry_hash"}, mode="json")
        canonical = json.dumps(body, sort_keys=True, separators=(",", ":"))
        return hashlib.sha256(canonical.encode("utf-8")).hexdigest()


class AuditLedger:
    """Append-only, hash-chained ledger. Concurrency-safe via a single lock."""

    GENESIS = "0" * 64

    def __init__(self) -> None:
        self._tip = self.GENESIS
        self._sequence = 0
        self._lock = asyncio.Lock()
        self._entries: list[AuditEntry] = []

    async def append(
        self, rec: Reconciliation, correlation_id: str, payload_hash: str, operator_id: str
    ) -> AuditEntry:
        async with self._lock:
            self._sequence += 1
            entry = AuditEntry(
                sequence=self._sequence,
                correlation_id=correlation_id,
                site_id=rec.site_id,
                item_code=rec.item_code,
                decision=rec.decision,
                target=rec.target,
                payload_hash=payload_hash,
                operator_id=operator_id,
                recorded_utc=datetime.now(timezone.utc),
                previous_hash=self._tip,
            )
            sealed = entry.model_copy(update={"entry_hash": entry.compute_hash()})
            self._tip = sealed.entry_hash
            self._entries.append(sealed)
            return sealed

    def verify(self) -> bool:
        """Re-walk the chain; returns False if any entry was altered."""
        prev = self.GENESIS
        for entry in self._entries:
            if entry.previous_hash != prev:
                return False
            if entry.compute_hash() != entry.entry_hash:
                return False
            prev = entry.entry_hash
        return True

In production the ledger is not an in-memory list. Persist sealed entries to write-once storage — an append-only table with row-level security, or object storage with an immutability lock — and run verify() as a scheduled integrity check. The in-memory version above is the reference implementation you unit-test against.

Orchestrating a full sync run #

The orchestrator pairs items by (site_id, item_code), reconciles each, performs idempotent upserts for authoritative decisions, quarantines conflicts, and records every outcome. Concurrency is bounded by the client semaphores; asyncio.gather collects results without leaking exceptions because each task wraps its own error handling.

import logging

logger = logging.getLogger("edc_ctms_sync")


def _idempotency_key(item: CanonicalItem) -> str:
    return f"{item.site_id}:{item.item_code}:{item.canonical_hash()}"


async def sync_study(
    study_id: str,
    correlation_id: str,
    ledger: AuditLedger,
    baselines: dict[tuple[str, str], str],
) -> list[AuditEntry]:
    """Reconcile one study's checklist between EDC and CTMS.

    ``baselines`` maps (site_id, item_code) to the canonical hash recorded at
    the previous successful sync; it is the basis for conflict detection.
    """
    async with ClinicalApiClient(System.EDC) as edc, ClinicalApiClient(System.CTMS) as ctms:
        edc_items = {
            (i.site_id, i.item_code): i
            async for i in _mapped(edc.iter_checklist(study_id), from_edc)
        }
        ctms_items = {
            (i.site_id, i.item_code): i
            async for i in _mapped(ctms.iter_checklist(study_id), from_ctms)
        }

        results: list[AuditEntry] = []
        for key in edc_items.keys() & ctms_items.keys():
            edc_item, ctms_item = edc_items[key], ctms_items[key]
            rec = reconcile(edc_item, ctms_item, baselines.get(key))

            try:
                if rec.decision in (Decision.EDC_AUTHORITATIVE, Decision.CTMS_AUTHORITATIVE):
                    assert rec.desired is not None and rec.target is not None
                    target = ctms if rec.target is System.CTMS else edc
                    payload = rec.desired.model_dump(mode="json")
                    await target.upsert_item(payload, _idempotency_key(rec.desired))
                    baselines[key] = rec.desired.canonical_hash()
                elif rec.decision == Decision.CONFLICT:
                    logger.warning(
                        "CONFLICT site=%s item=%s corr=%s", rec.site_id, rec.item_code, correlation_id
                    )
                else:  # IN_SYNC
                    baselines[key] = edc_item.canonical_hash()
            except aiohttp.ClientResponseError as exc:
                logger.error(
                    "UPSERT_FAILED site=%s item=%s status=%s", rec.site_id, rec.item_code, exc.status
                )
                rec = rec.model_copy(update={"decision": Decision.CONFLICT, "reason": "upsert failed"})

            payload_hash = rec.desired.canonical_hash() if rec.desired else edc_item.canonical_hash()
            entry = await ledger.append(rec, correlation_id, payload_hash, edc_item.operator_id)
            results.append(entry)

        return results


async def _mapped(
    source: AsyncIterator[dict[str, Any]], mapper: "Callable[[dict[str, Any]], CanonicalItem]"
) -> AsyncIterator[CanonicalItem]:
    async for raw in source:
        yield mapper(raw)

Items present in only one system (a key in the symmetric difference rather than the intersection) are not silently dropped — extend the loop to emit a CONFLICT-style audit entry for them so gap analysis can flag the asymmetry. That cross-system completeness check is exactly the work covered by the parent Checklist Sync & Gap Analysis topic.

Operational checklist #

Before promoting this engine to a regulated environment:

• Credentials for both systems are injected from a secrets manager into EDC_API_TOKEN / CTMS_API_TOKEN, never committed.
• The FIELD_OWNER precedence map is signed off by clinical operations and reflects who actually owns each data point.
• Conflicts route to a human-review queue with SLA, not an auto-resolve fallback.
• Audit entries persist to write-once storage and verify() runs on a schedule.
• Baselines are stored durably so conflict detection survives restarts.
• Rate limits and Retry-After handling are validated against each vendor’s documented quotas.
• All timestamps are timezone-aware UTC; naive datetimes are rejected at the model boundary.

Validation and edge-case handling #

The reconciler is only as trustworthy as the edge cases you exercise against it. Each of these is a real divergence shape a mixed EDC/CTMS estate produces:

• First-ever sync (no baseline). With baseline_hash is None, every field looks “changed.” The reconciler treats both sides as changed, so a same-value first run still resolves cleanly by owned-field diff rather than falsely quarantining. Seed baselines from a known-good side before go-live to avoid a wave of spurious conflicts.
• Taxonomy skew that is not a real change. SUBMITTED in the EDC and Pending QC in the CTMS both normalize to IN_REVIEW; because the canonical hash is computed on normalized values, the pair reports IN_SYNC with no write. Verify this — an adapter bug that leaks native enums into the hash produces endless phantom upserts.
• Unknown enum. A status the mapping table has never seen raises KeyError in from_edc / from_ctms. Catch it at the batch boundary and route it to categorizing validation errors in regulatory document pipelines as a mapping defect; never default it to NOT_STARTED.
• One-sided item. An item present in only the EDC or only the CTMS falls outside the key intersection. Emit a CONFLICT-style audit entry so gap analysis surfaces the asymmetry instead of silently skipping it.
• Partial batch failure. An upsert that returns 4xx after some items have already synced must not roll the whole run back. The orchestrator downgrades that single item to CONFLICT with reason="upsert failed", leaving the successfully-synced items and their baselines intact — the run is resumable, not all-or-nothing.
• Clock skew. Two servers disagree on wall-clock order, so the engine never decides authority by timestamp. Authority is decided purely by which owned fields changed versus the baseline, which is immune to skewed clocks.

Testing and verification #

The behaviours worth pinning with pytest are the ones that fail silently in production: normalization defeating taxonomy skew, the conflict rule firing only when both sides changed owned fields, and the audit chain detecting tampering. These are pure functions, so the tests need no network:

from datetime import datetime, timezone

_NOW = datetime(2026, 1, 1, tzinfo=timezone.utc)


def _edc_item(**over: object) -> CanonicalItem:
    base = dict(
        site_id="1042", item_code="FORM_1572", checklist_version="v3",
        status=ItemStatus.COMPLETE, evidence_doc_id="doc-9",
        source_updated_utc=_NOW, operator_id="cra-7",
    )
    base.update(over)
    return CanonicalItem(**base)


def test_taxonomy_skew_is_not_a_change() -> None:
    # Native EDC 'APPROVED' and CTMS 'Verified' both normalize to COMPLETE.
    edc = from_edc({"siteId": "1042", "code": "FORM_1572", "version": "v3",
                    "status": "APPROVED", "documentId": "doc-9",
                    "updatedAt": "2026-01-01T00:00:00+00:00", "lastEditedBy": "cra-7"})
    ctms = from_ctms({"site": "1042", "taskCode": "FORM_1572", "formVersion": "v3",
                      "state": "Verified", "signedOff": False, "plannedActivation": None,
                      "modifiedUtc": "2026-01-01T00:00:00+00:00", "modifiedBy": "pm-2"})
    assert reconcile(edc, ctms, baseline_hash=edc.canonical_hash()).decision is Decision.IN_SYNC


def test_conflict_only_when_both_sides_change_owned_fields() -> None:
    base = _edc_item()
    baseline = base.canonical_hash()
    edc = _edc_item(status=ItemStatus.REJECTED)          # EDC changed an EDC-owned field
    ctms = _edc_item(milestone_signed_off=True)          # CTMS changed a CTMS-owned field
    assert reconcile(edc, ctms, baseline).decision is Decision.CONFLICT


def test_edc_authoritative_when_only_edc_owned_field_changed() -> None:
    base = _edc_item()
    edc = _edc_item(status=ItemStatus.REJECTED)
    rec = reconcile(edc, base, base.canonical_hash())
    assert rec.decision is Decision.EDC_AUTHORITATIVE and rec.target is System.CTMS


def test_naive_timestamp_is_rejected() -> None:
    import pytest
    with pytest.raises(ValueError):
        _edc_item(source_updated_utc=datetime(2026, 1, 1))  # tz-naive


def test_audit_chain_detects_tampering() -> None:
    import asyncio

    async def _run() -> None:
        ledger = AuditLedger()
        rec = Reconciliation(site_id="1042", item_code="FORM_1572",
                             decision=Decision.IN_SYNC)
        await ledger.append(rec, "corr-1", "hash-a", "cra-7")
        await ledger.append(rec, "corr-1", "hash-b", "cra-7")
        assert ledger.verify() is True
        # Mutate a sealed entry in place: the chain must no longer verify.
        object.__setattr__(ledger._entries[0], "payload_hash", "tampered")
        assert ledger.verify() is False

    asyncio.run(_run())

Run them with pytest -q. The tamper test deliberately reaches past the model’s immutability to prove verify() catches an edit that a naive append-only log would miss.

FAQ #

Should the EDC or the CTMS be the source of truth? #

Neither, globally. Use field-level precedence: completion evidence and document status are authoritative in the EDC, while milestone sign-off and planned dates are owned by study managers in the CTMS. A blanket “one system wins” policy inevitably overwrites legitimate edits in the other.

How does the engine avoid duplicate writes on retry? #

Upserts use PUT against a deterministic resource path plus an Idempotency-Key derived from the canonical payload hash. A retried request carries the identical key, so the server collapses it to a single effect and the audit ledger records one event, not two.

What makes the audit trail tamper-evident rather than just append-only? #

Each entry embeds the hash of the previous entry, forming a chain. Editing any historical field changes that entry’s hash and breaks every link after it, which verify() detects. Combined with write-once storage, this meets the 21 CFR Part 11 requirement for a secure, time-stamped, computer-generated audit trail.

How are genuine conflicts handled? #

When both systems have changed fields they each own since the last successful sync, the reconciler returns CONFLICT and writes nothing. The item is quarantined for human adjudication and logged in the audit ledger, preserving ALCOA+ accuracy and completeness instead of guessing.

Does this scale to many sites and large checklists? #

Yes. The API client paginates with cursors, bounds concurrency with a semaphore, and streams items through async iterators rather than buffering whole studies. For very large multi-site onboarding waves, drive the orchestrator from the batch patterns in handling async batch processing for multi-site document ingestion.

Related #

• Checklist Sync & Gap Analysis — the parent topic framing cross-system completeness checks and activation-blocker detection.
• Schema Validation & Error Categorization — where an unknown status enum from either system is triaged as a mapping defect instead of guessed.
• Handling async batch processing for multi-site document ingestion — the bounded-concurrency worker that drives this reconciler across large onboarding waves.
• Configuring fallback routing when clinical portals timeout — transport-layer recovery when an EDC or CTMS portal is unreachable rather than slow.
• Automated Document Ingestion & Validation Workflows — how checklist reconciliation fits the wider ingest-to-submission lifecycle.

Up one level: this is a deep how-to under Checklist Sync & Gap Analysis.