Federation

Duration: 90 minutes Level: Advanced Module: fcc.federation

This tutorial teaches you how to federate FCC knowledge across multiple projects. You will learn to resolve entities across vocabulary namespaces, register and manage project namespaces, track changes for audit compliance, and coordinate everything through the FederationRegistry.

Prerequisites

• Completed beginner/intermediate tutorials
• Familiarity with PersonaRegistry and knowledge graphs
• Understanding of the FCC object model (fcc.objectmodel)

Why Federation?

The FCC ecosystem spans 11 projects, each with its own vocabulary, personas, and domain concepts. Federation enables:

• Entity resolution -- mapping a "Research Crafter" in one project to its equivalent in another
• Namespace management -- organizing entities under canonical prefixes and URIs
• Change tracking -- auditing how federated entities evolve over time
• Cross-project assessment -- comparing object model maturity across projects

EntityResolver

The EntityResolver translates local entity identifiers across namespaces using vocabulary mappings:

from fcc.federation.entities import EntityResolver, FederatedEntity
from fcc.objectmodel.mapping import VocabularyMapping

# Create vocabulary mappings between projects
mapping = VocabularyMapping(
    source_id="RC",
    source_name="Research Crafter",
    source_vocabulary="fcc",
    target_id="researcher",
    target_name="Research Analyst",
    target_vocabulary="partner_project",
    similarity_score=0.85,
)

# Set up the resolver
resolver = EntityResolver(mappings=[mapping])
print(f"Registered mappings: {resolver.mapping_count()}")  # 1

# Resolve a single entity
result = resolver.resolve("RC", "fcc", "partner_project")
if result is not None:
    print(f"Resolved: {result.canonical_id}")     # partner_project:researcher
    print(f"Confidence: {result.confidence}")      # 0.85
    print(f"Source vocab: {result.source_vocabulary}")
    print(f"Properties: {result.properties}")

Batch Resolution

Resolve multiple entities at once. Unresolved entities are returned with resolved=False:

# Add more mappings
resolver.add_mapping(VocabularyMapping(
    source_id="BC",
    source_name="Blueprint Crafter",
    source_vocabulary="fcc",
    target_id="architect",
    target_name="Solution Architect",
    target_vocabulary="partner_project",
    similarity_score=0.78,
))

# Batch resolve
entities = resolver.resolve_batch(
    ids=[("RC", "fcc"), ("BC", "fcc"), ("DE", "fcc")],
    target_namespace="partner_project",
)
for entity in entities:
    status = "resolved" if entity.resolved else "unresolved"
    print(f"  {entity.canonical_id}: {status} "
          f"(confidence={entity.confidence:.2f})")
# partner_project:researcher: resolved (confidence=0.85)
# partner_project:architect: resolved (confidence=0.78)
# fcc:DE: unresolved (confidence=0.00)

NamespaceRegistry

The NamespaceRegistry manages namespace configurations for all federated projects. Each namespace has a canonical label, prefix, base URI, and version:

from fcc.federation.namespaces import NamespaceRegistry, NamespaceConfig

# Create a registry manually
ns_registry = NamespaceRegistry()

# Register namespaces
ns_registry.register(NamespaceConfig(
    namespace="fcc",
    prefix="fcc",
    base_uri="https://fcc.example.org/ontology/",
    version="0.8.0",
    description="FCC Agent Team Framework",
))
ns_registry.register(NamespaceConfig(
    namespace="partner_project",
    prefix="pp",
    base_uri="https://partner.example.org/ontology/",
    version="1.2.0",
    description="Partner project knowledge base",
))

print(f"Namespaces registered: {ns_registry.count()}")  # 2

# Look up a namespace
fcc_config = ns_registry.get("fcc")
print(f"FCC base URI: {fcc_config.base_uri}")

# Resolve a URI
uri = ns_registry.resolve_uri("fcc", "RC")
print(f"Resolved URI: {uri}")
# https://fcc.example.org/ontology/RC

# Validate namespace existence
print(ns_registry.validate("fcc"))              # True
print(ns_registry.validate("unknown_project"))  # False

# List all namespaces
for ns_config in ns_registry.all_namespaces():
    print(f"  {ns_config.prefix}: {ns_config.description}")

Loading from YAML

Load the default registry containing all 11 ecosystem namespaces:

# Load the default 11-ecosystem registry from package data
default_registry = NamespaceRegistry.default_registry()
print(f"Default namespaces: {default_registry.count()}")

for ns in default_registry.all_namespaces():
    print(f"  {ns.namespace}: {ns.base_uri}")

ChangeTracker

The ChangeTracker records model changes and produces auditable change sets:

from fcc.federation.change_tracking import (
    ChangeTracker, ModelChange, ChangeSet
)

tracker = ChangeTracker()

# Track individual changes
tracker.track_change(ModelChange(
    entity_id="RC",
    change_type="modified",
    namespace="fcc",
    old_value={"role_title": "Research Crafter"},
    new_value={"role_title": "Research Crafter (v2)"},
))
tracker.track_change(ModelChange(
    entity_id="NEW_PERSONA",
    change_type="added",
    namespace="fcc",
    new_value={"name": "New Specialist", "category": "integration"},
))

# Create a change set from pending changes
changeset = tracker.create_changeset(
    description="Sprint 42 persona updates",
    source_namespace="fcc",
)
print(f"Changes: {len(changeset.changes)}")
print(f"Added: {changeset.added_count}")
print(f"Modified: {changeset.modified_count}")
print(f"Removed: {changeset.removed_count}")

Diffing Facades

Compare two ModelFacade instances to automatically generate a change set:

from fcc.objectmodel.examples import ExampleFacade

# Create two facade snapshots representing different versions
old_facade = ExampleFacade(name="v0.7.0")
new_facade = ExampleFacade(name="v0.8.0")

diff = tracker.diff_facades(old_facade, new_facade)
print(f"Diff changes: {len(diff.changes)}")
print(f"Description: {diff.description}")

Reviewing History

history = tracker.history()
print(f"Total change sets: {len(history)}")

for entry in history:
    cs = entry.change_set
    print(f"  [{cs.timestamp[:19]}] {cs.description}: "
          f"{len(cs.changes)} changes "
          f"(applied={entry.applied})")

FederationRegistry

The FederationRegistry is the top-level coordinator that aggregates namespace management, entity resolution, and change tracking:

from fcc.federation.registry import FederationRegistry
from fcc.federation.namespaces import NamespaceConfig

# Create the central registry
federation = FederationRegistry()

# Register projects
federation.add_project(
    namespace="fcc",
    namespace_config=NamespaceConfig(
        namespace="fcc",
        prefix="fcc",
        base_uri="https://fcc.example.org/ontology/",
        description="FCC Agent Team Framework",
    ),
)
federation.add_project(
    namespace="partner",
    namespace_config=NamespaceConfig(
        namespace="partner",
        prefix="pp",
        base_uri="https://partner.example.org/ontology/",
        description="Partner knowledge base",
    ),
)

print(f"Projects: {federation.project_count()}")
print(f"All projects: {federation.all_projects()}")

Cross-Project Entity Resolution

Add vocabulary mappings to the internal resolver, then resolve entities across all registered projects:

# Add mappings via the resolver
federation.entity_resolver.add_mapping(VocabularyMapping(
    source_id="RC",
    source_name="Research Crafter",
    source_vocabulary="fcc",
    target_id="researcher",
    target_name="Research Analyst",
    target_vocabulary="partner",
    similarity_score=0.85,
))

# Resolve across all projects
results = federation.resolve_across_projects("RC", "fcc")
for entity in results:
    print(f"  Resolved to {entity.canonical_id} "
          f"(confidence={entity.confidence:.2f})")

Health Check

Inspect the health of all federated projects:

health = federation.health_check()
print(f"Project count: {health['project_count']}")
print(f"Namespace count: {health['namespace_count']}")
print(f"Mapping count: {health['mapping_count']}")
print(f"Change history: {health['change_history_count']}")

for project in health["projects"]:
    print(f"  {project['namespace']}: "
          f"facade={'yes' if project['has_facade'] else 'no'}, "
          f"namespace={'yes' if project['has_namespace_config'] else 'no'}")

Accessing Sub-Components

The FederationRegistry exposes its sub-components for direct access:

# Access the namespace registry
ns_reg = federation.namespace_registry
print(f"Namespaces: {ns_reg.count()}")

# Access the entity resolver
resolver = federation.entity_resolver
print(f"Mappings: {resolver.mapping_count()}")

# Access the change tracker
tracker = federation.change_tracker
print(f"History entries: {len(tracker.history())}")

Cross-Project Assessment

Combine federation with the object model to assess maturity across projects:

from fcc.objectmodel.federation import assess_cross_project

# If you have facades for multiple projects:
# assessment = assess_cross_project([facade_fcc, facade_partner])
# print(f"Cross-project score: {assessment.overall_score}")

Complete Federation Workflow

Here is an end-to-end example that ties all federation components together:

from fcc.federation.registry import FederationRegistry
from fcc.federation.namespaces import NamespaceConfig
from fcc.federation.change_tracking import ModelChange
from fcc.objectmodel.mapping import VocabularyMapping

# 1. Set up federation
fed = FederationRegistry()

# 2. Register projects with namespace configs
for ns, desc in [("fcc", "FCC Framework"), ("paom", "PAOM Project"),
                 ("analytics", "Analytics Platform")]:
    fed.add_project(ns, namespace_config=NamespaceConfig(
        namespace=ns, prefix=ns[:3],
        base_uri=f"https://{ns}.example.org/ontology/",
        description=desc,
    ))

# 3. Register vocabulary mappings
fed.entity_resolver.add_mapping(VocabularyMapping(
    source_id="DGS", source_name="Data Governance Steward",
    source_vocabulary="fcc",
    target_id="data_steward", target_name="Data Steward",
    target_vocabulary="paom", similarity_score=0.90,
))

# 4. Resolve entities
resolved = fed.resolve_across_projects("DGS", "fcc")
for r in resolved:
    print(f"  DGS -> {r.canonical_id} ({r.confidence:.0%})")

# 5. Track changes
fed.change_tracker.track_change(ModelChange(
    entity_id="DGS", change_type="modified", namespace="fcc",
    old_value={"version": "0.7.0"},
    new_value={"version": "0.8.0"},
))
cs = fed.change_tracker.create_changeset("Version bump", "fcc")
print(f"Change set: {len(cs.changes)} changes")

# 6. Health check
health = fed.health_check()
print(f"Federation healthy: {health['project_count']} projects, "
      f"{health['mapping_count']} mappings")

Summary

In this tutorial you learned how to:

• Resolve entities across vocabulary namespaces with EntityResolver
• Manage project namespaces with NamespaceRegistry (including the default 11-ecosystem registry)
• Track and audit model changes with ChangeTracker and ChangeSet
• Coordinate federation through the FederationRegistry
• Run health checks and cross-project entity resolution

Next Steps

• Knowledge Graphs -- Build federated knowledge graphs
• Persona Composition Patterns -- Federated team patterns