Champion Team Sprint

Duration: 60 minutes Difficulty: Intermediate Pattern: Hub-and-Spoke

This scenario demonstrates a champion persona orchestrating a research sprint across specialized team members, using the event bus for coordination and quality gates for approval.

Scenario Overview

Problem: A comprehensive taxonomy of research metadata standards needs to be produced within a single sprint. The task requires multiple specialized perspectives coordinated by a champion.

Goal: Execute a champion-led sprint that delegates subtasks to specialists, collects results, synthesizes findings, and produces a quality-gated deliverable.

Persona Team

Persona	ID	Role in Sprint	Category
Research Crafter Champion	RCHM	Sprint lead, delegates and synthesizes	champion
Research Crafter	RC	Primary research and literature review	core
Catalog Indexer Analyst	CIA	Catalog structure and indexing strategy	integration
Semantic Taxonomy Expert	STE	Taxonomy design and classification	integration

Setup

from fcc.personas.registry import PersonaRegistry
from fcc.simulation.engine import SimulationEngine
from fcc.simulation.messages import SimulationMessage
from fcc.messaging.bus import EventBus
from fcc.messaging.events import Event, EventType

registry = PersonaRegistry.from_yaml_directory("src/fcc/data/personas")
bus = EventBus()
engine = SimulationEngine(registry=registry, mode="deterministic")

# Verify champion orchestration relationships
champion = registry.get("RCHM")
print(f"Champion: {champion.name}")
print(f"Champion of: {champion.champion_of}")
print(f"Orchestrates: {list(champion.orchestrates)}")

Phase 1: Champion Kickoff

The champion defines the sprint objective and creates task assignments:

# Champion defines the sprint
sprint_objective = (
    "Produce a comprehensive taxonomy of research metadata standards "
    "covering Dublin Core, DataCite, Schema.org, DCAT, and DDI. "
    "Deliverables: taxonomy hierarchy, standard comparison matrix, "
    "and adoption recommendations."
)

# Champion delegates to each spoke
tasks = {
    "RC": (
        "Conduct a literature review on metadata standards for research data. "
        "Cover Dublin Core, DataCite, Schema.org, DCAT, and DDI. "
        "Focus on adoption rates, interoperability, and community support."
    ),
    "CIA": (
        "Design a catalog indexing strategy for organizing metadata standards. "
        "Propose a hierarchical structure with categories, subcategories, "
        "and cross-references between related standards."
    ),
    "STE": (
        "Create a formal taxonomy for classifying metadata standards. "
        "Include concept hierarchies, broader/narrower relationships, "
        "and SKOS-compatible vocabulary terms."
    ),
}

Phase 2: Parallel Delegation

The champion sends tasks to each team member:

spoke_results = {}

for persona_id, task_description in tasks.items():
    message = SimulationMessage(
        sender="RCHM",
        receiver=persona_id,
        content=task_description,
        phase="find" if persona_id == "RC" else "create",
    )
    result = engine.step(message)
    spoke_results[persona_id] = result

    # Publish delegation event
    bus.publish(Event(
        event_type=EventType.SIMULATION_STEP_COMPLETED,
        source=f"champion_sprint.{persona_id}",
        payload={
            "persona_id": persona_id,
            "task": task_description[:100],
            "output_length": len(result.content),
        },
    ))
    print(f"  {persona_id} completed: {len(result.content)} chars")

Phase 3: Champion Synthesis

The champion collects all spoke outputs and synthesizes them into a unified deliverable:

# Prepare synthesis input
synthesis_input = "Sprint findings from team:\n\n"
for persona_id, result in spoke_results.items():
    persona = registry.get(persona_id)
    synthesis_input += f"## {persona.name} ({persona_id})\n"
    synthesis_input += f"{result.content[:500]}\n\n"

synthesis_message = SimulationMessage(
    sender="orchestrator",
    receiver="RCHM",
    content=(
        f"Synthesize the following team findings into a unified "
        f"taxonomy deliverable:\n\n{synthesis_input}\n\n"
        "Produce: 1) Executive summary, 2) Unified taxonomy hierarchy, "
        "3) Standards comparison matrix, 4) Adoption recommendations."
    ),
    phase="create",
)

synthesis_result = engine.step(synthesis_message)
print(f"Champion synthesis: {len(synthesis_result.content)} chars")

Phase 4: Quality Gate

Apply quality gates to the synthesized deliverable:

from fcc.collaboration.scoring import ScoringEngine

scorer = ScoringEngine()
quality_score = scorer.score_text(synthesis_result.content)
print(f"Quality score: {quality_score:.2f}")

# Quality gate decision
quality_threshold = 0.7
if quality_score >= quality_threshold:
    print("Quality gate PASSED -- deliverable approved")
    gate_status = "approved"
else:
    print("Quality gate FAILED -- needs revision")
    gate_status = "rejected"

# Publish gate event
bus.publish(Event(
    event_type=EventType.COLLABORATION_GATE_DECIDED,
    source="champion_sprint.quality_gate",
    payload={
        "gate": "sprint_quality",
        "score": quality_score,
        "threshold": quality_threshold,
        "status": gate_status,
    },
))

Event Bus Coordination

Monitor the sprint through event subscriptions:

sprint_log = []

def log_event(event: Event) -> None:
    sprint_log.append({
        "type": event.event_type.value,
        "source": event.source,
        "timestamp": event.timestamp,
    })

bus.subscribe(EventType.SIMULATION_STEP_COMPLETED, log_event)
bus.subscribe(EventType.COLLABORATION_GATE_DECIDED, log_event)

# Print the sprint log
print(f"\nSprint event log ({len(sprint_log)} events):")
for entry in sprint_log:
    print(f"  [{entry['timestamp'][:19]}] "
          f"{entry['type']}: {entry['source']}")

Sprint Output Summary

summary = {
    "sprint_objective": sprint_objective[:100],
    "champion": "RCHM",
    "team_size": len(tasks),
    "spoke_outputs": {
        pid: len(r.content) for pid, r in spoke_results.items()
    },
    "synthesis_length": len(synthesis_result.content),
    "quality_score": quality_score,
    "gate_status": gate_status,
}

import json
print(json.dumps(summary, indent=2))

Exercises

1. Add a feedback loop: If the quality gate fails, have the champion send specific revision requests back to the relevant spoke personas.
2. Expand the team: Add a governance persona (GCA) as a final approval step after the quality gate.
3. Track time: Add timestamps to each delegation and calculate per-persona execution time.
4. Cross-reference: Use the CrossReferenceMatrix to verify that the champion has valid orchestration relationships with all spoke personas.

Summary

In this scenario you executed a champion-led research sprint:

• RCHM defined the sprint objective and delegated subtasks
• RC, CIA, and STE each contributed specialized findings
• RCHM synthesized spoke outputs into a unified deliverable
• A quality gate evaluated the final output
• The event bus provided full observability of the sprint

Next Steps

• Governance Audit Flow -- Add compliance auditing
• ML Pipeline Handoff -- Multi-stage pipeline with handoffs