Collaboration API Reference¶

Module: fcc.collaboration

The collaboration module provides human-in-the-loop session management with scoring, approval gates, progress tracking, shared context, and session recording.

stateDiagram-v2
    [*] --> CREATED : create_session()
    CREATED --> ACTIVE : start_session()
    ACTIVE --> ACTIVE : take_turn()
    ACTIVE --> PAUSED : pause_session()
    PAUSED --> ACTIVE : resume_session()
    ACTIVE --> COMPLETED : complete_session()
    CREATED --> ABORTED : abort_session()
    ACTIVE --> ABORTED : abort_session()
    PAUSED --> ABORTED : abort_session()
    COMPLETED --> [*]
    ABORTED --> [*]

Module Structure¶

Module	Description
`fcc.collaboration.models`	11 frozen dataclasses for session data
`fcc.collaboration.engine`	`CollaborationEngine` session lifecycle management
`fcc.collaboration.scoring`	`ScoringEngine` deliverable quality evaluation
`fcc.collaboration.progress`	`ProgressTracker` completion monitoring
`fcc.collaboration.context`	`SharedContext` auditable key-value workspace
`fcc.collaboration.recording`	`SessionRecorder` JSON persistence and replay

CollaborationEngine¶

from fcc.collaboration.engine import CollaborationEngine

Method	Signature	Returns	Description
`create_session()`	`(workflow_id, participants=None, gates=None, handoff_protocol=None)`	CollaborationSession	Create a new session
`start_session()`	`(session_id: str)`	CollaborationSession	CREATED -> ACTIVE
`take_turn()`	`(session_id, participant, turn_type, content, metadata=None)`	CollaborationSession	Record a turn
`pause_session()`	`(session_id: str)`	CollaborationSession	ACTIVE -> PAUSED
`resume_session()`	`(session_id: str)`	CollaborationSession	PAUSED -> ACTIVE
`complete_session()`	`(session_id: str)`	CollaborationSession	ACTIVE -> COMPLETED
`abort_session()`	`(session_id: str)`	CollaborationSession	Any -> ABORTED
`get_session()`	`(session_id: str)`	CollaborationSession	Get current session state
`list_sessions()`	`(status=None)`	list[CollaborationSession]	List all or filtered sessions

Usage¶

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

engine = CollaborationEngine()

session = engine.create_session(
    workflow_id="base_sequence",
    participants=["reviewer", "RC"],
    gates=[ApprovalGate(gate_id="g1", workflow_node_id="n2")],
)

session = engine.start_session(session.session_id)
session = engine.take_turn(
    session.session_id, "RC", TurnType.AGENT,
    content="Research findings...",
)
session = engine.complete_session(session.session_id)

ScoringEngine¶

from fcc.collaboration.scoring import ScoringEngine

Method	Signature	Returns	Description
`score_deliverable()`	`(deliverable_id, scorer, score, rubric_scores=None, justification="")`	QualityScore	Score a deliverable (1-5)
`evaluate_at_gate()`	`(gate, deliverable_id, scorer, score, persona_id=None)`	tuple[ApprovalDecision, QualityScore]	Evaluate against gate threshold
`compute_capability_rating()`	`(persona_id: str)`	AgentCapabilityRating	Compute aggregate rating
`scores`	property	list[QualityScore]	All recorded scores
`decisions`	property	list[tuple[str, ApprovalDecision]]	All gate decisions

Usage¶

from fcc.collaboration.scoring import ScoringEngine
from fcc.collaboration.models import ApprovalGate

scoring = ScoringEngine()
qs = scoring.score_deliverable("d1", "reviewer", 4.2,
    rubric_scores={"completeness": 4.5, "accuracy": 3.9})

gate = ApprovalGate(gate_id="g1", workflow_node_id="n2", required_score=3.5)
decision, score = scoring.evaluate_at_gate(gate, "d1", "reviewer", 4.2)
# decision == ApprovalDecision.APPROVED

ProgressTracker¶

from fcc.collaboration.progress import ProgressTracker

Method	Signature	Returns	Description
`register()`	`(entity_id, entity_type, total_steps)`	ProgressState	Register entity for tracking
`advance()`	`(entity_id, steps=1)`	ProgressState	Advance progress
`complete()`	`(entity_id)`	ProgressState	Mark as completed
`fail()`	`(entity_id)`	ProgressState	Mark as failed
`get()`	`(entity_id)`	ProgressState	Get current state
`all_states()`	`()`	list[ProgressState]	Get all tracked states

Usage¶

from fcc.collaboration.progress import ProgressTracker

tracker = ProgressTracker()
state = tracker.register("s1", "session", total_steps=5)
state = tracker.advance("s1", steps=2)
print(state.percentage)  # 40.0
state = tracker.complete("s1")
print(state.status)  # "completed"

SharedContext¶

from fcc.collaboration.context import SharedContext

Method	Signature	Returns	Description
`get()`	`(key, default=None)`	Any	Get a value
`set()`	`(key, value, actor="system")`	None	Set a value (recorded in history)
`delete()`	`(key, actor="system")`	bool	Remove key; returns True if found
`keys()`	`()`	list[str]	All current keys
`to_dict()`	`()`	dict	Copy of current data
`history`	property	list[dict]	Full audit trail of changes

Usage¶

from fcc.collaboration.context import SharedContext

ctx = SharedContext()
ctx.set("status", "in_progress", actor="RC")
ctx.set("status", "complete", actor="DE")
print(ctx.get("status"))  # "complete"
print(len(ctx.history))   # 2 (both set operations recorded)

SessionRecorder¶

from fcc.collaboration.recording import SessionRecorder

Method	Signature	Returns	Description
`save_json()`	`(session, path)`	None	Save session to JSON file
`load_json()`	`(path)`	CollaborationSession	Load session from JSON file
`replay_session()`	`(session, bus)`	int	Replay session events through bus

Usage¶

from fcc.collaboration.recording import SessionRecorder

SessionRecorder.save_json(session, "output/session.json")
restored = SessionRecorder.load_json("output/session.json")

from fcc.messaging.bus import EventBus
bus = EventBus()
deliveries = SessionRecorder.replay_session(restored, bus)

Data Models¶

All models are frozen dataclasses from fcc.collaboration.models:

Class	Description
`SessionStatus`	Enum: CREATED, ACTIVE, PAUSED, COMPLETED, ABORTED
`TurnType`	Enum: HUMAN, AGENT, SYSTEM
`ApprovalDecision`	Enum: APPROVED, REJECTED, NEEDS_REVISION, DEFERRED
`ApprovalGate`	Checkpoint with required_score and rubric
`QualityScore`	Deliverable quality evaluation
`SessionTurn`	Single turn in a session
`HandoffProtocol`	Turn transition rules
`CollaborationSession`	Full session snapshot
`ProgressState`	Progress tracking state
`AgentCapabilityRating`	Aggregate capability rating
`DeliverableRating`	Per-deliverable rating

All models support to_dict() and from_dict() for serialization.