Skip to content

Chapter 7: Collaboration Sessions

Learning Objectives

By the end of this chapter you will be able to:

  1. Create and manage collaboration sessions programmatically.
  2. Define approval gates at strategic points in a workflow.
  3. Implement scoring rubrics for deliverable evaluation.
  4. Record sessions and replay them for debugging and audit.
  5. Use shared context to pass information between FCC cycles.

The state diagram below traces a collaboration session from creation through gate-based review loops to a saved recording, showing where human feedback re-enters the Active state.

stateDiagram-v2
    [*] --> Created: create_session(config)
    Created --> Active: engine traverses graph
    Active --> Active: agent turn
    Active --> GateReview: approval gate reached
    GateReview --> Active: approved (score >= threshold)
    GateReview --> Revision: needs_revision
    Revision --> Active: human feedback applied
    Active --> Completed: all gates passed
    Completed --> Saved: SessionRecorder.save()
    Saved --> [*]

Design review gates sparingly: one gate per high-stakes Critique node is usually enough, since every additional gate adds latency and human-review overhead to the session.

Creating a Session

A collaboration session wraps a simulation run with human-in-the-loop capabilities. While a plain simulation runs to completion without stopping, a collaboration session can pause at approval gates, accept human input, and resume.

from fcc.collaboration.engine import CollaborationEngine
from fcc.collaboration.models import SessionConfig
from fcc.scenarios.loader import ScenarioLoader

loader = ScenarioLoader()
scenario = loader.load("competitive_analysis")

config = SessionConfig(
    scenario=scenario,
    workflow="extended_sequence",
    mode="mock",
    approval_gates=["critique_design", "final_review"],
    max_iterations=3,
)

engine = CollaborationEngine()
session = engine.create_session(config)

print(f"Session ID: {session.id}")
print(f"Status: {session.status}")  # "active"

The approval_gates parameter lists the node IDs where the session should pause for human review. You can place gates at any Critique node.

Running Until a Gate

Instead of running the entire workflow at once, advance the session until the next approval gate:

# Run until the first gate
session = engine.advance(session)
print(f"Status: {session.status}")  # "paused"
print(f"Paused at: {session.current_gate}")  # "critique_design"

# Inspect the current deliverable
deliverable = session.current_deliverable
print(deliverable.output[:500])
print(f"Quality scores: {deliverable.scores}")

The session pauses at critique_design and presents the current deliverable -- the output of the Create node that precedes the gate. The human reviewer can now inspect the deliverable, its quality scores, and the trace of how it was produced.

Providing Human Input

The human reviewer has three options at an approval gate:

Option 1: Approve

session = engine.approve(session, rating=4, feedback="Good analysis, proceed.")

The session records the approval, the rating (1--5), and the feedback, then resumes the workflow from the next node after the gate.

Option 2: Reject with Feedback

session = engine.reject(
    session,
    rating=2,
    feedback="Missing competitive pricing data. Please re-research.",
)

The session records the rejection, triggers the feedback edge from the gate's Critique node back to the appropriate Find or Create node, and re-runs that portion of the workflow. The feedback is injected into the re-run as additional context, so the persona knows what to fix.

Option 3: Modify Directly

session = engine.modify(
    session,
    modified_output="[Human-edited version of the deliverable]",
    rating=3,
    feedback="I corrected the pricing section manually.",
)

The session records the human modification, replaces the current deliverable with the modified version, and resumes the workflow. This option is for cases where it is faster for the human to fix the issue directly rather than describing it to the AI.

Scoring Rubrics

The scoring engine evaluates deliverables at each gate using configurable rubrics. A rubric defines the dimensions on which a deliverable is scored:

from fcc.collaboration.scoring import ScoringEngine, Rubric, Criterion

rubric = Rubric(
    name="design_review",
    criteria=[
        Criterion(
            name="completeness",
            description="All requirements are addressed",
            weight=0.3,
        ),
        Criterion(
            name="technical_accuracy",
            description="Technical details are correct",
            weight=0.3,
        ),
        Criterion(
            name="clarity",
            description="Document is clear and well-structured",
            weight=0.2,
        ),
        Criterion(
            name="actionability",
            description="Recommendations are actionable",
            weight=0.2,
        ),
    ],
)

scoring_engine = ScoringEngine()
score_result = scoring_engine.evaluate(
    artifact=deliverable.output,
    rubric=rubric,
    mode="mock",
)

print(f"Overall score: {score_result.overall:.2f}")
for criterion_score in score_result.criteria_scores:
    print(f"  {criterion_score.name}: {criterion_score.score:.2f}")

Rubrics can be defined in YAML for reuse across sessions:

rubrics:
  design_review:
    criteria:
      - name: completeness
        description: "All requirements are addressed"
        weight: 0.3
      - name: technical_accuracy
        description: "Technical details are correct"
        weight: 0.3
      - name: clarity
        description: "Document is clear and well-structured"
        weight: 0.2
      - name: actionability
        description: "Recommendations are actionable"
        weight: 0.2

Session Recording

Every session is automatically recorded. The recorder captures:

  • All turns (persona activations and outputs).
  • All human inputs (approvals, rejections, modifications).
  • All scores (automated and human).
  • All events emitted during the session.
from fcc.collaboration.recording import SessionRecorder

recorder = SessionRecorder()

# Save the session
recorder.save(session, "sessions/analysis_session.json")

# Load and inspect a saved session
loaded = recorder.load("sessions/analysis_session.json")
print(f"Turns: {len(loaded.turns)}")
print(f"Human inputs: {len(loaded.human_inputs)}")
print(f"Final score: {loaded.final_score:.2f}")

Replay

Replay re-executes the session step by step, emitting events to the event bus:

recorder.replay("sessions/analysis_session.json")

During replay, you can attach subscribers to inspect events, verify that dashboards update correctly, or generate reports. Replay is deterministic -- the same session always produces the same events.

Shared Context

The shared context (src/fcc/collaboration/context.py) provides an auditable workspace that all participants (personas and humans) can read from and write to:

from fcc.collaboration.context import SharedContext

context = SharedContext(session_id=session.id)

# Write a finding
context.set("market_size", "4.2B USD", source="research_analyst")

# Read a finding
value = context.get("market_size")
print(value)  # "4.2B USD"

# Audit trail
history = context.history("market_size")
for entry in history:
    print(f"  {entry.timestamp}: {entry.value} (set by {entry.source})")

Every read and write is logged with a timestamp and source identifier. This audit trail is critical for understanding how shared information evolved during a multi-cycle session.

Cross-Cycle Context

When a session spans multiple FCC cycles (common with the extended and complete workflow graphs), the shared context persists across cycles. Findings from the first Find phase are available to the second and third Find phases, avoiding redundant research.

The context also supports namespaced keys to prevent collisions:

context.set("cycle1.requirements", "...", source="domain_expert")
context.set("cycle2.requirements", "...", source="domain_expert")

Progress Tracking

The progress tracker provides real-time visibility into session completion:

from fcc.collaboration.progress import ProgressTracker

tracker = ProgressTracker(session)

print(f"Nodes completed: {tracker.nodes_completed}/{tracker.nodes_total}")
print(f"Gates passed: {tracker.gates_passed}/{tracker.gates_total}")
print(f"Estimated remaining: {tracker.estimated_remaining_ms}ms")
print(f"Completion: {tracker.completion_pct:.0f}%")

Progress is also published to the event bus, enabling real-time dashboard updates.

The Collaboration CLI

The FCC CLI provides a terminal interface for collaboration sessions:

# Start an interactive session
fcc collab start --scenario competitive_analysis --workflow extended_sequence

# List active sessions
fcc collab list

# Resume a paused session
fcc collab resume <session-id>

# View session history
fcc collab history <session-id>

The CLI dashboard shows the current session state, pending approval gates, scores, and progress in an ASCII-formatted terminal display.

Key Takeaways

  • Collaboration sessions wrap simulations with human-in-the-loop approval gates.
  • Humans can approve, reject with feedback, or modify deliverables directly.
  • Scoring rubrics evaluate deliverables across configurable dimensions.
  • Session recording captures all turns, inputs, scores, and events for audit and replay.
  • Shared context provides an auditable workspace that persists across FCC cycles.
  • Progress tracking and CLI dashboards provide real-time visibility.

Cross-References

← Chapter 6: Event Bus and Observability | Next: Chapter 8 -- Production Deployment →