Skip to content

symfonic.services.conversation.capability

capability

The conversation capability: identity plus history, as one plan value.

Small on purpose. The capability's job is to answer "for this session, which thread key and which history policy govern?" and to answer it as a value the compiler can bind into a plan. Everything heavier — sessions, transcripts, checkpoints — is a service this capability does not own and does not reach into, which is what keeps conversation policy out of orchestration.

ConversationCapability 

ConversationCapability(*, history: HistoryStrategy | None = None, clock: object | None = None)

Resolves a session identity and a history policy into one plan.

Source code in src/symfonic/services/conversation/capability.py 
53
54
55
56
57
def __init__(
    self, *, history: HistoryStrategy | None = None, clock: object | None = None
) -> None:
    self._history = history
    self._clock = clock if clock is not None else _SystemClock()

plan_for 

plan_for(identity: SessionIdentity) -> ConversationPlan

Deterministic for a given identity: no clock, no counters, no ids.

Reproducibility is the property that lets a parity harness compare the legacy and migrated paths turn for turn.

Source code in src/symfonic/services/conversation/capability.py 
59
60
61
62
63
64
65
66
67
68
69
def plan_for(self, identity: SessionIdentity) -> ConversationPlan:
    """Deterministic for a given identity: no clock, no counters, no ids.

    Reproducibility is the property that lets a parity harness compare the
    legacy and migrated paths turn for turn.
    """
    return ConversationPlan(
        thread_id=identity.thread_id,
        identity=identity,
        history=resolve_history(self._history),
    )

ConversationPlan dataclass 

ConversationPlan(thread_id: str, identity: SessionIdentity, history: HistoryDirective)

What one session's conversation looks like before any turn runs.

to_legacy_overrides 

to_legacy_overrides() -> dict[str, Any]

Config overrides plus the thread key, for the legacy adapter.

Source code in src/symfonic/services/conversation/capability.py 
34
35
36
37
38
def to_legacy_overrides(self) -> dict[str, Any]:
    """Config overrides plus the thread key, for the legacy adapter."""
    overrides = self.history.to_legacy_overrides()
    overrides["_thread_id"] = self.thread_id
    return overrides