EAAPL-INT008 — Bidirectional AI Sync

Status: Proven Tags: event-driven traceability audit-logging high-complexity Version: 2.0.0 Last Updated: 2026-06-12

1. Pattern Identity

2. Executive Summary

The Bidirectional AI Sync pattern defines the architecture for maintaining consistency between AI systems and enterprise data stores when changes can originate from either side: source data changes trigger AI re-processing to update AI-derived outputs; AI system outputs are written back to target enterprise data stores and must be validated before acceptance. This bidirectional flow creates three distinct complexity challenges that unidirectional integration patterns do not address: conflict resolution when the same entity is updated by both a human and the AI at or near the same time; eventual consistency management across the full round trip; and a complete sync audit trail that can reconstruct the chain from a source data change through AI processing to a target record update. The pattern also governs data quality preservation — AI-generated writes to enterprise data stores must pass schema validation, business rule validation, and referential integrity checks before being committed, preventing AI hallucinations or quality regressions from corrupting authoritative data stores.

3. Problem Statement

3.1 Context

Enterprise AI systems increasingly operate in closed-loop architectures: they read from enterprise data stores (CRM, ERP, document management systems), process the data, and write derived results back to those same or adjacent stores (enriched records, risk scores, classification labels, generated summaries). When source data is updated by a human or another system, the AI's derived outputs become stale and must be re-computed. When the AI writes back, it must not overwrite concurrent human changes without conflict resolution.

3.2 Forces in Tension

• Freshness vs. processing cost. Re-processing AI outputs every time source data changes is expensive at scale. Batching changes reduces cost but increases staleness.
• AI authority vs. human authority. For some data fields, the AI's output should take precedence (classification labels, risk scores). For others, human edits must never be overwritten by AI (customer contact preferences, manually verified facts).
• Write-back quality vs. throughput. Validating AI outputs before writing back to enterprise stores prevents corruption but adds latency and processing overhead.
• Sync completeness vs. eventual consistency. Full consistency at every moment requires synchronous processing that cannot scale. Eventual consistency scales but requires all consumers to handle stale reads gracefully.

3.3 Failure Modes Without This Pattern

Without CDC triggering AI re-processing, AI-derived outputs grow progressively stale as source data is updated. Without conflict resolution, AI writes silently overwrite human edits, destroying work. Without sync audit logging, it is impossible to investigate data quality issues or prove regulatory compliance. Without pre-write validation, AI hallucinations are committed to authoritative data stores and propagate to downstream systems.

4. Solution

4.1 Bidirectional Sync Architecture

4.2 Conflict Resolution Decision Tree

5. Structure

5.1 Component Catalogue

5.2 Sync Event Record Schema

{
  "syncEventId": "uuid-v4",
  "eventType": "SOURCE_CHANGE_TRIGGERED | AI_OUTPUT_VALIDATED | WRITE_BACK_COMMITTED | CONFLICT_DETECTED | CONFLICT_RESOLVED",
  "sourceEntity": {
    "entityType": "contract",
    "entityId": "uuid-v4",
    "changeType": "UPDATE",
    "changedFields": ["clause_4_2_text"],
    "changedBy": "human-user-id | ai-agent-id",
    "changedAt": "ISO-8601",
    "sourceVersion": 14
  },
  "aiProcessing": {
    "triggeredAt": "ISO-8601",
    "completedAt": "ISO-8601",
    "agentId": "risk-analysis-agent-v2",
    "outputFields": ["risk_score", "risk_flags"],
    "validationResult": { "schemaPass": true, "businessRulesPass": true, "referentialIntegrityPass": true }
  },
  "targetWrite": {
    "entityId": "uuid-v4",
    "fieldsWritten": ["risk_score", "risk_flags"],
    "targetVersionBefore": 9,
    "targetVersionAfter": 10,
    "conflictDetected": false,
    "conflictResolution": null,
    "committedAt": "ISO-8601"
  },
  "latencyMs": {
    "cdcToAiTrigger": 120,
    "aiProcessing": 4200,
    "validationToWrite": 85,
    "totalSyncLatencyMs": 4405
  }
}

6. Behaviour

6.1 Change Data Capture Triggering AI Re-Processing

When source data changes, the CDC connector emits a change event. The change event router classifies the change:

• Scope: which AI-derived outputs depend on the changed fields?
• Priority: is this a high-priority change (e.g., customer risk classification changed) or a low-priority change (e.g., a contact address updated)?
• Batch eligibility: can this change be batched with other pending changes for the same entity?

High-priority changes are processed immediately. Low-priority changes are batched: the router accumulates changes for the same entity within a configurable window (default: 30 seconds) and triggers a single AI re-processing run covering all accumulated changes. This prevents N AI calls for N rapid successive edits to the same entity.

Dependency tracking. A field dependency map defines which source fields affect which AI output fields. The router uses this map to determine which AI pipelines to trigger. Only the affected pipelines run — not a full re-derivation of all AI outputs.

6.2 Conflict Resolution Strategies

Conflicts arise when both a human and the AI system update the same record between the time the AI read the source data and the time it attempts to write back. Conflict detection uses version timestamps: every record carries a version integer that increments on every write. Before writing, the AI system checks whether the current version in the target store matches the version it read. A mismatch indicates a concurrent write.

Resolution strategies by field type:

Last-write-wins as a fallback. For organisations that cannot afford the human review overhead for shared fields, last-write-wins (by timestamp) is an acceptable fallback with lower quality guarantees. This must be documented as a conscious trade-off, not a default.

Conflict audit record. Every conflict detection and resolution event is written to the sync audit log with: both the AI output and the conflicting human edit, the resolution policy applied, the resolved value, and (for human-review cases) the reviewer's final decision.

6.3 Eventual Consistency Management

The bidirectional sync system is eventually consistent by design. At any point in time, a consumer reading an AI-derived field may be reading a value computed from source data that is N versions old. Consumers must handle this gracefully:

Version timestamps. Every synced record carries sourceVersion (the version of source data the AI processed) and aiOutputVersion (the version of the AI output). Consumers that require strong consistency can check whether sourceVersion matches the current source record version and treat the AI output as stale if it does not.

Staleness metadata. For read-heavy consumers that cannot afford to check version consistency on every read, the sync system can provide a staleness indicator: a field aiOutputAgeSeconds computed from the time of the last successful sync. Consumers can display this to users ("Risk score updated 3 minutes ago") or use it to trigger an explicit refresh.

Stale read handling. Define a stale threshold per field type. For a risk score that changes rarely, a 24-hour stale threshold may be acceptable. For a sentiment score on a live customer interaction, a 5-minute threshold may be required. Consumers that read a field with staleness exceeding the threshold should either trigger a synchronous re-computation or display a "refreshing" indicator.

6.4 Sync Audit Logging

Every sync event — source change detection, AI processing, validation, conflict detection, write-back — is logged to an immutable sync audit trail. The audit trail enables:

• Full reconstruction. Given an entity ID, reconstruct the full chain: which source changes triggered AI re-processing, what the AI produced, whether conflicts were detected, and what was written to the target.
• Data quality investigation. When a target record contains a value that appears incorrect, trace it back through the sync audit trail to identify which AI processing run produced it, what input data it consumed, and whether the validation step passed or was bypassed.
• Regulatory compliance. For regulated data (financial records, clinical data), the audit trail demonstrates that AI-written values were validated before being committed and that human edits were not overwritten without explicit conflict resolution.

6.5 Data Quality Preservation — Validation Before Write-Back

AI-generated outputs must pass three validation layers before being written to a target enterprise data store:

Schema validation. The AI output must conform to the target field's JSON schema or database column type. A risk score must be a float between 0.0 and 1.0, not a string or a number outside range. A classification label must be one of the defined enum values, not a fabricated label.

Business rule validation. The AI output must satisfy the business rules that govern the target field. For example: a risk score cannot decrease by more than 30% in a single re-computation without triggering an alert; a contract status classification cannot transition from "executed" to "draft" regardless of what the AI produces.

Referential integrity validation. If the AI output references other entities (e.g., a risk score that is linked to a specific risk category record), the referenced entities must exist in the target data store. Orphaned foreign key values are rejected.

A validation failure does not corrupt the target store — the AI output is quarantined in a validation failure queue for investigation. The target record retains its previous value. An alert is emitted.

7. Implementation Guide

7.1 Step-by-Step

Step 1 — Map the field dependency graph. Before writing any code, document which source data fields affect which AI output fields. This is the foundation for the change event router's selective triggering logic. Without this map, every source change triggers a full AI re-computation.

Step 2 — Set up CDC. Deploy a CDC connector on your source data store. For Postgres, use logical replication + Debezium. For SQL Server, use CDC tables. For DynamoDB, use DynamoDB Streams. Configure the CDC connector to emit change events to a durable stream (Kafka, Kinesis, SQS).

Step 3 — Build the change event router. Consume the CDC stream, apply the field dependency map to determine which AI pipelines are affected, batch rapid successive changes per entity, and enqueue AI re-processing tasks.

Step 4 — Build the output validator. Implement the three validation layers (schema, business rules, referential integrity) as a composable validation pipeline. Each layer returns: { pass: true } | { pass: false, layer: "schema|business|integrity", reason: "..." }.

Step 5 — Implement conflict detection. On every AI output write attempt, compare the current target version against the version the AI read. If they differ, invoke the conflict resolver with the AI output, the current target value, and the field ownership map.

Step 6 — Build the sync audit logger. Instrument every step of the sync pipeline to emit events to the audit logger. Use a fan-out pattern: the audit logger receives events from all pipeline stages and writes them to the immutable audit store.

Step 7 — Implement staleness indicators. Add aiOutputVersion, sourceVersionAtProcessing, and aiOutputUpdatedAt fields to all synced target records. Build a query helper that computes staleness for a given record.

7.2 Code Skeleton (TypeScript)

interface SyncPipelineConfig {
  entityType: string;
  fieldDependencies: FieldDependencyMap;
  conflictPolicy: ConflictPolicy;
  validationRules: ValidationRule[];
  staleThresholdSeconds: Record<string, number>;
}

class BidirectionalAISyncPipeline {
  async processCDCEvent(event: CDCEvent, config: SyncPipelineConfig): Promise<SyncResult> {
    const syncEventId = crypto.randomUUID();

    // Step 1: Determine affected AI pipelines
    const affectedPipelines = config.fieldDependencies.getAffectedPipelines(event.changedFields);
    if (affectedPipelines.length === 0) {
      return { syncEventId, status: "NO_PIPELINES_AFFECTED" };
    }

    // Step 2: Run AI re-processing
    const aiOutputs = await Promise.all(
      affectedPipelines.map(pipeline => pipeline.run(event.entityId))
    );

    // Step 3: Validate outputs
    for (const output of aiOutputs) {
      const validationResult = await this.validate(output, config.validationRules);
      if (!validationResult.pass) {
        await validationFailureQueue.send({ syncEventId, output, reason: validationResult.reason });
        await auditLog.append({ syncEventId, event: "VALIDATION_FAILED", reason: validationResult.reason });
        return { syncEventId, status: "VALIDATION_FAILED", reason: validationResult.reason };
      }
    }

    // Step 4: Check for conflicts and resolve
    const currentRecord = await targetStore.get(event.entityId);
    const conflict = this.detectConflict(currentRecord, event, aiOutputs);

    if (conflict) {
      const resolution = config.conflictPolicy.resolve(conflict);
      await auditLog.append({ syncEventId, event: "CONFLICT_RESOLVED", resolution });
      if (resolution.action === "HUMAN_REVIEW") {
        await humanReviewQueue.send({ syncEventId, conflict });
        return { syncEventId, status: "PENDING_HUMAN_REVIEW" };
      }
    }

    // Step 5: Write back to target store
    const mergedUpdate = this.buildMergedUpdate(currentRecord, aiOutputs, conflict?.resolution);
    await targetStore.update(event.entityId, {
      ...mergedUpdate,
      aiOutputVersion: mergedUpdate.version + 1,
      sourceVersionAtProcessing: event.sourceVersion,
      aiOutputUpdatedAt: new Date().toISOString()
    });

    await auditLog.append({ syncEventId, event: "WRITE_BACK_COMMITTED", entityId: event.entityId });
    return { syncEventId, status: "SUCCESS" };
  }

  private detectConflict(
    currentRecord: TargetRecord,
    cdcEvent: CDCEvent,
    aiOutputs: AIOutput[]
  ): Conflict | null {
    // Version mismatch = concurrent write occurred
    if (currentRecord.version !== cdcEvent.targetVersionAtReadTime) {
      return {
        aiOutput: aiOutputs,
        humanEdit: currentRecord,
        versionAtRead: cdcEvent.targetVersionAtReadTime,
        versionNow: currentRecord.version
      };
    }
    return null;
  }
}

8. Observability

8.1 Sync Pipeline Metrics

8.2 Sync Lag Dashboard

A dedicated dashboard showing the distribution of aiOutputAgeSeconds across all synced record types. This is the primary indicator of whether the sync system is keeping up with source data change velocity. A widening distribution indicates backpressure in the AI re-processing pipeline.

9. Cost Governance

• Selective triggering. Only run AI re-processing for the pipelines affected by the changed fields (from the dependency map). Do not re-run all pipelines on every change. This is the highest-impact cost control.
• Batching. Batch rapid successive changes for the same entity. A document edited 10 times in 30 seconds requires only one AI re-processing run, not 10.
• Model tiering by priority. Use efficient models for low-priority re-processing (e.g., updating a general sentiment score); use frontier models only for high-priority, high-stakes re-processing (e.g., updating a credit risk score).
• Re-processing budget per entity. Set a maximum re-processing frequency per entity type (e.g., re-process a contract no more than once per hour). Changes within the window are batched and processed together at the next scheduled window. Prevents runaway cost from high-frequency source data changes.

10. Security Considerations

10.1 AI Write Authorisation

The AI sync pipeline must write to target enterprise data stores using a service account with the minimum required write permissions. The service account must not have permission to write to fields classified as human-owned in the conflict policy. Enforce this at the database permission level, not only in application code.

10.2 Audit Log Integrity

The sync audit log is a critical compliance artefact. Apply the same hash-chain integrity mechanism described in EAAPL-MAG006: each record includes the hash of the previous record. This makes retroactive tampering detectable.

10.3 Data Masking in Audit Records

Sync audit records may capture sensitive source data (e.g., a CDC event for a credit record). Apply data masking to PII fields in audit records per your data classification policy. The audit record should capture the fact of the change and the version numbers, not necessarily the full field values.

11. Failure Modes and Mitigations

12. Compliance and Governance

12.1 Data Lineage

The sync audit trail constitutes the data lineage record for all AI-written fields. For any field in the target data store that was written by the AI sync pipeline, the sync audit trail provides the answer to: what source data change triggered this AI output, which AI model processed it, what validation checks passed, and whether any conflict was detected and resolved. This lineage is required for GDPR Article 22 (automated decision-making), financial regulatory data lineage requirements, and clinical data audit requirements.

12.2 GDPR Article 22 — Automated Decision-Making

When AI-written fields are used in decisions that significantly affect individuals (credit decisions, risk classifications, clinical recommendations), the sync audit trail provides the Article 22 explanation basis: the individual can request what data was used to generate the AI output and which AI system produced it.

12.3 Data Quality Management

Integrate the sync validation failure queue with your data quality management process. Validation failures are data quality incidents. They must be triaged, root-caused (is this an AI pipeline regression or a data model change?), and resolved within a defined SLA.

13. Testing Strategy

13.1 Unit Tests

• Field dependency map: given a CDC event changing field X, assert the correct set of AI pipelines is returned.
• Schema validator: given an AI output with an out-of-range risk score, assert validation fails with SCHEMA_INVALID.
• Business rule validator: given a risk score decrease of 50%, assert the business rule fires and the output is rejected.
• Conflict detector: given currentVersion: 10 and versionAtRead: 9, assert a conflict is detected.
• Conflict resolver: for each field type (AI-owned, human-owned, shared), assert the correct resolution policy is applied.

13.2 Integration Tests

• Full sync cycle: emit a CDC event; assert AI re-processing runs; assert validation passes; assert target record is updated with new AI output and incremented version.
• Conflict scenario: between AI read and write, update the target record manually; assert conflict is detected; assert resolution policy is applied; assert audit record is written.
• Validation failure: configure a test AI pipeline that returns an out-of-range value; assert the validation failure queue receives the rejected output and the target record is NOT modified.
• Stale threshold alert: update a source record but do not process the CDC event for 2× the stale threshold; assert a stale record alert is emitted.

13.3 Load Tests

• Simulate 1000 CDC events per minute; assert end-to-end sync latency remains within SLA; assert no events are dropped from the queue.
• Simulate a burst of 100 rapid edits to the same entity within 30 seconds; assert batching fires and only one AI re-processing run is triggered.

13.4 End-to-End Tests

• Update a source contract record in the source system; within the latency SLA, assert the target risk score record is updated; assert the sync audit trail contains a complete record with all five event types (SOURCE_CHANGE_TRIGGERED, AI_OUTPUT_VALIDATED, WRITE_BACK_COMMITTED).

14. Variants and Extensions

14.1 Event-Sourced Sync

Rather than overwriting target records, write each AI-derived output as an immutable event to an event store. The "current" AI output is the result of applying all events for a given entity in sequence. This provides full history of AI output evolution and simplifies rollback — roll back by replaying events up to a prior version.

14.2 Federated Sync

For enterprise architectures with multiple source systems and multiple target stores, deploy one sync pipeline per source-target pair. A sync coordination layer tracks which pipelines have processed which source versions and alerts when pipelines fall out of sync with each other (e.g., the CRM sync processed version 15 but the risk system sync is still on version 12).

14.3 Human-in-the-Loop Write-Back

For the highest-stakes AI-to-enterprise writes (e.g., updating a clinical patient record, updating a credit limit), insert a HITL checkpoint (EAAPL-MAG003) between the output validation and the target write. A human reviewer approves the AI-generated value before it is committed. The sync pipeline pauses at the checkpoint and resumes after approval.

15. Trade-off Analysis

16. Known Implementations

17. Related Patterns

18. References

1. Gartner, "Bidirectional AI Data Integration Patterns," 2025 (ID: G00824512)
2. Debezium: Open Source CDC Platform — debezium.io
3. Fowler, M., "Event Sourcing Pattern" — martinfowler.com/eaaDev/EventSourcing.html
4. Kleppmann, M., "Designing Data-Intensive Applications," O'Reilly, 2017 — Chapter 11: Stream Processing
5. GDPR (Regulation 2016/679), Article 22: Automated Individual Decision-Making
6. AWS Database Blog, "Implementing Change Data Capture with Amazon DMS," 2024
7. Confluent, "Building Real-Time Data Pipelines with Kafka Connect and CDC," 2024
8. NIST AI RMF 1.0, Measure 2.6: Ongoing Data Quality Monitoring
9. ISO 8000: Data Quality — Part 61: Data Quality Management
10. Martin Fowler, "Versioning in an Event Sourced System" — martinfowler.com/articles/eventsourcing-versioning.html