Skip to content

Your First Collaboration Session

The FCC collaboration engine enables human-in-the-loop workflows where AI agent personas and human reviewers take turns contributing to a deliverable. This tutorial walks through the full session lifecycle: creation, turn-taking, gate evaluation, handoff detection, and session completion.

Core Concepts

Concept Description
Session A workspace linking a workflow to participants, turns, and approval gates
Turn A single contribution from a human, agent, or system actor
Approval Gate A checkpoint where a deliverable is scored against a rubric
Handoff Protocol Rules governing when an agent should yield to a human reviewer
Shared Context An auditable key-value store shared across all turns

Creating a Session

from fcc.collaboration.engine import CollaborationEngine
from fcc.collaboration.models import ApprovalGate, HandoffProtocol

# Create the engine
engine = CollaborationEngine()

# Define approval gates
gates = [
    ApprovalGate(
        gate_id="gate-research",
        workflow_node_id="node-RC",
        required_score=3.5,
        requires_human=True,
        rubric=(
            "Source attribution completeness",
            "Coverage of required topics",
            "Factual accuracy",
        ),
    ),
    ApprovalGate(
        gate_id="gate-blueprint",
        workflow_node_id="node-BC",
        required_score=4.0,
        requires_human=True,
        rubric=(
            "Architecture diagram clarity",
            "Component specification completeness",
            "Governance compliance",
        ),
    ),
]

# Define handoff rules
handoff = HandoffProtocol(
    max_consecutive_agent_turns=3,
    auto_approve_threshold=4.5,
    escalation_threshold=2.0,
)

# Create the session
session = engine.create_session(
    workflow_id="wf-base-sequence",
    participants=["RC", "BC", "human-reviewer"],
    gates=gates,
    handoff_protocol=handoff,
)

print(f"Session ID: {session.session_id}")
print(f"Status: {session.status.value}")
print(f"Gates: {len(session.gates)}")
print(f"Participants: {list(session.participants)}")

Expected output:

Session ID: a1b2c3d4-...
Status: created
Gates: 2
Participants: ['RC', 'BC', 'human-reviewer']

Starting the Session

Sessions begin in CREATED status and must be explicitly started:

session = engine.start_session(session.session_id)
print(f"Status: {session.status.value}")  # "active"

Session Status Lifecycle

Sessions follow this state machine: CREATED -> ACTIVE -> COMPLETED or ABORTED. Only active sessions accept new turns.

Taking Turns

Agent Turns

Record an agent persona's contribution using agent_turn:

turn1 = engine.agent_turn(
    session_id=session.session_id,
    persona_id="RC",
    content=(
        "## Research Brief: Payment Gateway Architecture\n\n"
        "### Sources Identified\n"
        "- Internal architecture docs (2024-Q3)\n"
        "- API specification v2.1\n"
        "- Production incident reports (last 6 months)\n\n"
        "### Key Findings\n"
        "1. Current gateway handles 50K TPS\n"
        "2. Authentication uses OAuth 2.0 + mTLS\n"
        "3. Rate limiting is per-tenant, not global"
    ),
)

print(f"Turn ID: {turn1.turn_id}")
print(f"Type: {turn1.turn_type.value}")
print(f"Actor: {turn1.actor}")

Human Turns

Record human feedback using take_turn with TurnType.HUMAN:

from fcc.collaboration.models import TurnType, ApprovalDecision

turn2 = engine.take_turn(
    session_id=session.session_id,
    actor="human-reviewer",
    content="Good coverage. Please add the disaster recovery SLA from the SRE runbook.",
    turn_type=TurnType.HUMAN,
)

Turns with Approval

Humans can attach an approval decision and a score to their turn:

turn3 = engine.take_turn(
    session_id=session.session_id,
    actor="human-reviewer",
    content="Research brief now meets all rubric criteria.",
    turn_type=TurnType.HUMAN,
    approval=ApprovalDecision.APPROVED,
    score=4.2,
)
print(f"Approval: {turn3.approval.value}")
print(f"Score: {turn3.score}")

Evaluating Quality Gates

Gates are evaluated using the built-in ScoringEngine. The engine compares a score against the gate's required_score to produce a decision:

decision, quality_score = engine.evaluate_gate(
    session_id=session.session_id,
    gate_id="gate-research",
    scorer="human-reviewer",
    score=4.2,
    persona_id="RC",
)

print(f"Decision: {decision.value}")
print(f"Score: {quality_score.score}")
print(f"Scorer: {quality_score.scorer}")

Expected output:

Decision: approved
Score: 4.2
Scorer: human-reviewer

Decision Logic

  • Score >= required_score -> APPROVED
  • Score >= required_score - 1.0 -> NEEDS_REVISION
  • Score < required_score - 1.0 -> REJECTED

Handoff Detection

The handoff protocol prevents agents from running indefinitely without human review:

# After 3 consecutive agent turns, handoff is triggered
engine.agent_turn(session.session_id, "RC", "Draft v1")
engine.agent_turn(session.session_id, "RC", "Draft v2")
engine.agent_turn(session.session_id, "RC", "Draft v3")

needs_handoff = engine.handoff(session.session_id)
print(f"Handoff needed: {needs_handoff}")  # True

The max_consecutive_agent_turns setting (default 3) controls this threshold.

Using Shared Context

Each session has a SharedContext -- an auditable key-value store that persists across turns:

ctx = engine.get_context(session.session_id)

# Set values with actor attribution
ctx.set("research_status", "complete", actor="RC")
ctx.set("sources_count", 15, actor="RC")

# Read values
print(f"Research status: {ctx.get('research_status')}")
print(f"Sources: {ctx.get('sources_count')}")

# Check what changed and who changed it
for entry in ctx.history:
    print(f"  {entry['actor']} set {entry['key']} = {entry['new_value']}")

Completing a Session

final_session = engine.complete_session(session.session_id)
print(f"Status: {final_session.status.value}")  # "completed"
print(f"Total turns: {len(final_session.turns)}")

Saving and Loading Sessions

Use SessionRecorder to persist sessions as JSON:

from fcc.collaboration.recording import SessionRecorder

# Save
SessionRecorder.save_json(final_session, "session_output.json")

# Load
loaded = SessionRecorder.load_json("session_output.json")
print(f"Loaded session: {loaded.session_id}")
print(f"Turns: {len(loaded.turns)}")

Replaying Sessions Through the Event Bus

Recorded sessions can be replayed as events for analysis or testing:

from fcc.messaging.bus import EventBus
from fcc.collaboration.recording import SessionRecorder

bus = EventBus()
replay_log = []
bus.subscribe(lambda e: replay_log.append(e))

deliveries = SessionRecorder.replay_session(final_session, bus)
print(f"Replayed {len(replay_log)} events ({deliveries} deliveries)")

for event in replay_log:
    print(f"  {event.event_type.value}: {event.payload}")

Tracking Progress

Use ProgressTracker for completion tracking across sessions:

from fcc.collaboration.progress import ProgressTracker

tracker = ProgressTracker()
tracker.register(session.session_id, "session", total_steps=5)

# Advance as work completes
tracker.advance(session.session_id)
tracker.advance(session.session_id, steps=2)

state = tracker.get_state(session.session_id)
print(f"Progress: {state.percentage:.0f}% ({state.completed_steps}/{state.total_steps})")
print(f"Status: {state.status}")

# View all tracked entities
print(tracker.summary())

Next Steps