Bidirectional Federation Patterns

Duration: 80 minutes Level: Advanced Module: fcc.federation.bidirectional Pillar: v1.5.0 Pillar E

This tutorial teaches you how to synchronize entity changes across two or more federated namespaces using BidirectionalSync. You will learn the three conflict policies, the conflict-to-resolution flow, how the merge queue handles manual-only cases, and two worked patterns (federated commerce and federated research repos) that show the class in realistic topologies.

The unidirectional problem

v1.3.x federation gave FCC a read-only view onto partner namespaces. FCC could resolve a Research Crafter in project A to a Research Analyst in project B, but every change still had to flow one direction. That was fine for a distillation topology (FCC as the downstream consumer) and it broke as soon as any constellation vertical wanted to propose a vocabulary term back upstream.

v1.5.0 Pillar E adds the reverse channel. BidirectionalSync holds one ChangeTracker per namespace, materializes each namespace's state into a shared {(namespace, entity_id): change} dict, and reconciles conflicts via a per-namespace ConflictPolicy. The class is pure in-memory with no external I/O — you can embed it inside a FederationRegistry, wire it to an EventBus subscriber, or drive it directly.

The three policies

ConflictPolicy (from fcc.federation.namespaces) is an Enum with three values:

Policy	Resolution when both sides edit an entity
LAST_WRITE_WINS	Compare ISO-8601 timestamps on the two changes; the newer one wins. Ties favor the source side.
DOMAIN_OWNER_AUTHORITY	Look up the target namespace's domain_owner; the owner's change wins. Falls through to LWW when no owner is configured.
MANUAL_MERGE	Queue the conflict on the merge_queue and defer — no automatic write.

Each NamespaceConfig carries its own conflict_policy, so different namespaces can use different rules inside the same federation. When a sync crosses a boundary, the target namespace's policy governs the conflict.

The Resolution enum

BidirectionalSync.reconcile returns a Resolution describing the outcome:

• ACCEPTED_SOURCE — source change won, target state updated
• ACCEPTED_TARGET — target change was newer / authoritative, source state was reconciled
• QUEUED_FOR_MERGE — MANUAL_MERGE policy; the Conflict is in the queue
• SKIPPED — currently only returned internally as a defensive no-op

Minimal sync walkthrough

Two namespaces, one entity edited on both sides, LWW policy on the target.

from datetime import datetime, timezone

from fcc.federation.bidirectional import BidirectionalSync
from fcc.federation.change_tracking import ModelChange
from fcc.federation.namespaces import (
    ConflictPolicy, NamespaceConfig, NamespaceRegistry,
)

# 1. Registry with two namespaces, one LWW, one owner-authoritative
ns = NamespaceRegistry()
ns.register(NamespaceConfig(
    namespace="fcc",
    prefix="fcc",
    base_uri="https://fcc.example.org/ontology/",
    conflict_policy=ConflictPolicy.LAST_WRITE_WINS,
))
ns.register(NamespaceConfig(
    namespace="partner",
    prefix="pp",
    base_uri="https://partner.example.org/ontology/",
    conflict_policy=ConflictPolicy.LAST_WRITE_WINS,
))

# 2. Sync coordinator
sync = BidirectionalSync(ns)

# 3. Seed both sides with a known state so the conflict is visible
now = datetime.now(timezone.utc).isoformat()
earlier = "2026-04-20T10:00:00+00:00"

sync.seed("fcc", ModelChange(
    entity_id="RC",
    change_type="modified",
    namespace="fcc",
    old_value={"v": 0},
    new_value={"v": 2},
    timestamp=now,            # fcc edit is NEWER
))
sync.seed("partner", ModelChange(
    entity_id="RC",
    change_type="modified",
    namespace="partner",
    old_value={"v": 0},
    new_value={"v": 1},
    timestamp=earlier,        # partner edit is OLDER
))

# 4. Stage an outbound change on fcc's tracker
sync.tracker_for("fcc").track_change(ModelChange(
    entity_id="RC",
    change_type="modified",
    namespace="fcc",
    old_value={"v": 2},
    new_value={"v": 3},
    timestamp=now,
))

# 5. Propagate fcc -> partner
report = sync.sync_to(target_namespace="partner", source_namespace="fcc")
print(f"Applied:   {report.applied}")
print(f"Conflicts: {report.conflicts}")
print(f"Resolved:  {report.resolved}")
print(f"Lag ms:    {report.sync_lag_ms:.2f}")
print(f"Resolutions: {report.resolutions}")

The sync_to method replays every outbound change on fcc's tracker against partner's materialized state. The target's policy decides each conflict. The report returned is a SyncReport — a frozen dataclass you can JSON-serialize and ship into your audit trail.

The merge queue for MANUAL_MERGE

When the target policy is MANUAL_MERGE, conflicts are queued rather than auto-resolved. The caller is expected to review each conflict, pick a winner, and call resolve_manual to apply it.

from fcc.federation.bidirectional import BidirectionalSync
from fcc.federation.change_tracking import ModelChange
from fcc.federation.namespaces import (
    ConflictPolicy, NamespaceConfig, NamespaceRegistry,
)

ns = NamespaceRegistry()
ns.register(NamespaceConfig(
    namespace="fcc", prefix="fcc",
    base_uri="https://fcc.example.org/",
    conflict_policy=ConflictPolicy.MANUAL_MERGE,
))
ns.register(NamespaceConfig(
    namespace="partner", prefix="pp",
    base_uri="https://partner.example.org/",
    conflict_policy=ConflictPolicy.MANUAL_MERGE,
))

sync = BidirectionalSync(ns)

# Pre-existing state on fcc, outbound change on partner
sync.seed("fcc", ModelChange(
    entity_id="DGS", change_type="modified",
    namespace="fcc", new_value={"version": "1.0"},
))
sync.tracker_for("partner").track_change(ModelChange(
    entity_id="DGS", change_type="modified",
    namespace="partner", new_value={"version": "1.1-edge"},
))

report = sync.sync_to(target_namespace="fcc", source_namespace="partner")
print(f"Queued: {report.queued}")   # 1

# Inspect the merge queue
for conflict in sync.merge_queue():
    print("Conflict on", conflict.entity_id)
    print("  source:", conflict.source_change.new_value)
    print("  target:", conflict.target_change.new_value)

# Operator decides — apply a hand-picked winner
winner = ModelChange(
    entity_id="DGS", change_type="modified",
    namespace="fcc", new_value={"version": "1.1-stable"},
)
sync.resolve_manual(entity_id="DGS", winner=winner)
assert sync.state_for("fcc", "DGS").new_value == {"version": "1.1-stable"}

resolve_manual applies the chosen winner to both namespaces' materialized state and removes the conflict from the queue. If the entity is not in the queue, it returns False so the caller can detect the race.

DOMAIN_OWNER_AUTHORITY — structural authority

Use this policy when one namespace is the authoritative owner of an entity class and the other is a consumer. Changes from the owner override changes from the consumer; the reverse is a conflict that falls through to LWW.

from fcc.federation.namespaces import (
    ConflictPolicy, NamespaceConfig, NamespaceRegistry,
)

ns = NamespaceRegistry()
ns.register(NamespaceConfig(
    namespace="fcc", prefix="fcc",
    base_uri="https://fcc.example.org/",
    conflict_policy=ConflictPolicy.DOMAIN_OWNER_AUTHORITY,
    domain_owner="fcc",   # fcc owns its own persona catalog
))
ns.register(NamespaceConfig(
    namespace="partner", prefix="pp",
    base_uri="https://partner.example.org/",
    conflict_policy=ConflictPolicy.DOMAIN_OWNER_AUTHORITY,
    domain_owner="fcc",   # partner treats fcc as authoritative
))

When partner proposes a change that conflicts with fcc's state, reconcile reads the target namespace's domain_owner:

• owner == source_namespace → ACCEPTED_SOURCE (authoritative push)
• owner == target_namespace → ACCEPTED_TARGET (partner tried to override owner; rejected)
• owner unset or mismatched → falls through to LWW

This is the right pattern for reference vocabulary management: the reference owner pushes canonical changes; consumers may propose, but only the owner's changes take effect automatically. Propose-from-consumer becomes a change that must round-trip through the owner.

Pattern 1: federated commerce — shared product catalog

Two projects, commerce-a and commerce-b, share a product catalog. Pricing changes happen on both sides; structural changes (new SKUs, category splits) are owned by commerce-a.

from fcc.federation.bidirectional import BidirectionalSync
from fcc.federation.change_tracking import ChangeTracker, ModelChange
from fcc.federation.namespaces import (
    ConflictPolicy, NamespaceConfig, NamespaceRegistry,
)

ns = NamespaceRegistry()
ns.register(NamespaceConfig(
    namespace="commerce-a", prefix="ca",
    base_uri="https://a.example.org/",
    conflict_policy=ConflictPolicy.DOMAIN_OWNER_AUTHORITY,
    domain_owner="commerce-a",
))
ns.register(NamespaceConfig(
    namespace="commerce-b", prefix="cb",
    base_uri="https://b.example.org/",
    conflict_policy=ConflictPolicy.LAST_WRITE_WINS,   # pricing -> LWW
))

sync = BidirectionalSync(ns)

# A launches a new SKU (structural)
sync.tracker_for("commerce-a").track_change(ModelChange(
    entity_id="SKU-42",
    change_type="added",
    namespace="commerce-a",
    new_value={"name": "Widget", "price_cents": 999, "category": "gadgets"},
))

# Push to B — accepted because A owns the catalog structure
r1 = sync.sync_to("commerce-b", source_namespace="commerce-a")
assert r1.applied == 1

# B adjusts price locally (non-structural)
sync.tracker_for("commerce-b").track_change(ModelChange(
    entity_id="SKU-42",
    change_type="modified",
    namespace="commerce-b",
    new_value={"name": "Widget", "price_cents": 899, "category": "gadgets"},
))

# Pull B's pricing change into A — LWW wins on target=commerce-a? NO —
# commerce-a's policy is DOMAIN_OWNER_AUTHORITY with owner=commerce-a, so
# B's change is REJECTED back to commerce-a's state.
r2 = sync.sync_to("commerce-a", source_namespace="commerce-b")
print(r2.resolutions)   # {"SKU-42": Resolution.ACCEPTED_TARGET}

The asymmetric policy choice keeps catalog structure canonical on A while allowing B to drift on pricing — until someone explicitly reconciles.

Pattern 2: federated research repos

Two research teams run independent persona catalogs that share a common vocabulary. Either side may propose a new term; conflicts go to human review.

from fcc.federation.bidirectional import BidirectionalSync
from fcc.federation.change_tracking import ModelChange
from fcc.federation.namespaces import (
    ConflictPolicy, NamespaceConfig, NamespaceRegistry,
)

ns = NamespaceRegistry()
for label in ("team-east", "team-west"):
    ns.register(NamespaceConfig(
        namespace=label, prefix=label[:4],
        base_uri=f"https://{label}.example.org/",
        conflict_policy=ConflictPolicy.MANUAL_MERGE,
    ))

sync = BidirectionalSync(ns)

# Both teams independently discover the same concept and label it differently
sync.tracker_for("team-east").track_change(ModelChange(
    entity_id="CONCEPT-42",
    change_type="added",
    namespace="team-east",
    new_value={"term": "RetrievalAugmentedAnswering"},
))
sync.tracker_for("team-west").track_change(ModelChange(
    entity_id="CONCEPT-42",
    change_type="added",
    namespace="team-west",
    new_value={"term": "GroundedResponseSynthesis"},
))

# East pushes first; concept lands on west (no conflict yet)
sync.sync_to("team-west", source_namespace="team-east")
# West's outbound change collides with east's state on sync back
report = sync.sync_to("team-east", source_namespace="team-west")
print(f"Queued on east: {report.queued}")  # 1

# Human broker picks the winning term
broker_pick = ModelChange(
    entity_id="CONCEPT-42",
    change_type="modified",
    namespace="agreed",
    new_value={"term": "RetrievalAugmentedAnswering",
               "aliases": ["GroundedResponseSynthesis"]},
)
sync.resolve_manual("CONCEPT-42", broker_pick)

The MANUAL_MERGE policy never auto-resolves — the operator is always in the loop for federated research vocabularies where accidental overrides are costly.

Inspecting federation health

The health() helper gives a per-namespace snapshot you can surface in a dashboard:

snapshot = sync.health()
for row in snapshot["namespaces"]:
    lag = row["sync_lag"]
    lag_s = "never" if lag is None else f"{lag:.1f}s"
    print(f"  {row['namespace']:12}  policy={row['policy']:25}  "
          f"lag={lag_s:10}  conflicts={row['conflicts']}")
print(f"Pending conflicts: {snapshot['pending_conflicts']}")

• sync_lag is seconds since the last sync touched the namespace; None means the namespace has never participated in a sync.
• conflicts per namespace counts queue entries where the namespace is on either side.
• tracked_namespaces is the count of ChangeTrackers instantiated so far.

Push a snapshot to your observability lane on a schedule (every 30 s is a good starting point) so operators can see lag growing before it becomes a problem.

Integrating with EventBus

BidirectionalSync does not subscribe to the event bus itself — that would couple it to a specific bus implementation. Instead, wire it up via a subscriber plugin:

from fcc.messaging.bus import EventBus
from fcc.messaging.events import EventType
from fcc.federation.change_tracking import ModelChange

def on_model_change(event) -> None:
    change = ModelChange(**event.payload["change"])
    sync.tracker_for(event.payload["namespace"]).track_change(change)
    # Optionally kick a sync immediately
    for target in ("partner",):
        sync.sync_to(target, source_namespace=event.payload["namespace"])

bus = EventBus()
bus.subscribe(EventType.MODEL_CHANGED, on_model_change)

Keep the subscriber idempotent — duplicate events (e.g. from event replay) should result in duplicate track_change calls that wash out on the next sync_to because the resulting outbound is the same.

Timestamp discipline

LWW depends on accurate timestamps. _parse_ts uses datetime.fromisoformat and falls back to Unix epoch on malformed input — an epoch timestamp always loses, which is a safety net, not a feature. Your ingestion layer should:

• Attach timestamp=datetime.now(timezone.utc).isoformat() on every ModelChange at the point of edit.
• Refuse to accept changes with future timestamps (clock-skew attacks).
• Log when timestamp is missing so upstream clients can fix the bug.

Troubleshooting

All conflicts go to ACCEPTED_SOURCE. Either the source really is newer, or the target's timestamps are malformed and parsing to epoch. Print _parse_ts(target_change.timestamp) for a suspect change.

sync_to raises ValueError: namespace not registered. The namespace must be in the NamespaceRegistry passed to BidirectionalSync. A common mistake is registering one side and forgetting the other.

Merge queue grows without bound under MANUAL_MERGE. Alarm on sync.health()["pending_conflicts"] > threshold. Build a UI that lets operators resolve conflicts in bulk.

Outbound queue is double-consumed. The _run_sync method drains src_tracker._outbound after each call. If you call sync_to twice in a row the second call finds an empty queue — by design. Stage a fresh outbound change each cycle.

Summary

In this tutorial you learned how to:

• Build a BidirectionalSync over a NamespaceRegistry with per-namespace ConflictPolicy choices
• Run sync_to / sync_from passes and interpret SyncReport.resolutions
• Use MANUAL_MERGE + merge_queue + resolve_manual for human-in-the-loop conflict resolution
• Apply DOMAIN_OWNER_AUTHORITY for structurally asymmetric federations
• Seed initial state with seed() and inspect health with health()
• Wire the sync to an EventBus subscriber for reactive propagation

Next steps

• Combine with FederationRegistry (see the Federation tutorial) for the full namespace + entity-resolver + bidirectional-sync surface
• Emit ModelChange events from a ChangeTracker onto your event bus so sync_to runs on every edit
• Publish sync.health() to your monitoring lane on a schedule
• Add a merge-queue dashboard (see the CLI fcc.dashboard.federation module)

Related tutorials

• Federation — one-directional resolver + namespaces
• Knowledge Graphs — the data model underneath
• GraphRAG Deep Dive — reads from federated KGs
• Messaging Patterns — wiring subscribers

Related references

• ADR-013 Bidirectional federation conflict policy — docs/decisions/ADR-013_bidirectional_federation_conflict_policy.md
• Six Pillars overview — publications/chapters/arch/six-pillars.qmd
• Bidirectional federation chapter stub — publications/chapters/arch/bidirectional-federation.qmd
• Source — src/fcc/federation/bidirectional.py, src/fcc/federation/namespaces.py
• Sequence diagram — docs/architecture/sequence-diagrams/bidirectional-federation-sync.md