Skip to content

vllm_omni.engine.duplex.session.engine_session

The one duplex session state, owned by the engine-side DuplexSessionRunner.

DuplexEngineSession is the single session object: the input / response / playback / conversation ledgers, the lease, the identity fence, the stage request resources, the append sequencing, the model plugin's per-session state (model_state) and the Realtime projection state (projector). There is no cross-boundary fence protocol; session.fence is derived from the session's own epoch/turn and accepted_fence is the monotonic high-water mark of fences accepted for stage requests (sync_fence() publishes the current identity there).

RESPONSE_REQUEST_MEASUREMENT_ORIGIN module-attribute 

RESPONSE_REQUEST_MEASUREMENT_ORIGIN: dict[str, str] = {
    "ttft": "accepted native-append start to first non-empty text output; pending turns keep the latest append before first output, so overlapping user speech shortens TTFT",
    "ttfp": "accepted native-append start to first audio output; pending turns keep the latest append before first output, so overlapping user speech shortens TTFP",
}

ConversationHistory dataclass

assistant_response_snapshots class-attribute instance-attribute 

assistant_response_snapshots: dict[
    str,
    tuple[
        dict[str, object],
        tuple[DuplexAssistantAudioTextMark, ...],
        int,
    ],
] = field(default_factory=dict)

hard_truncations_ms class-attribute instance-attribute 

hard_truncations_ms: dict[str, int] = field(
    default_factory=dict
)

history_item_placeholders class-attribute instance-attribute 

history_item_placeholders: dict[str, dict[str, object]] = (
    field(default_factory=dict)
)

item_audio_text_marks class-attribute instance-attribute 

item_audio_text_marks: dict[
    str, list[DuplexAssistantAudioTextMark]
] = field(default_factory=dict)

item_ids class-attribute instance-attribute 

item_ids: dict[str, dict[str, object]] = field(
    default_factory=dict
)

last_assistant_audio_text_marks class-attribute instance-attribute 

last_assistant_audio_text_marks: list[
    DuplexAssistantAudioTextMark
] = field(default_factory=list)

last_assistant_full_message class-attribute instance-attribute 

last_assistant_full_message: dict[str, object] | None = None

messages class-attribute instance-attribute 

messages: list[dict[str, object]] = field(
    default_factory=list
)

pending_item_audio_text_marks class-attribute instance-attribute 

pending_item_audio_text_marks: dict[
    str, list[DuplexAssistantAudioTextMark]
] = field(default_factory=dict)

pending_item_ids class-attribute instance-attribute 

pending_item_ids: dict[str, dict[str, object]] = field(
    default_factory=dict
)

pending_item_input_commit_seqs class-attribute instance-attribute 

pending_item_input_commit_seqs: dict[str, int] = field(
    default_factory=dict
)

pending_truncations_ms class-attribute instance-attribute 

pending_truncations_ms: dict[str, int] = field(
    default_factory=dict
)

unanswered_user_items class-attribute instance-attribute 

unanswered_user_items: int = 0

DuplexAppendReservation dataclass

base_append_turn_key instance-attribute 

base_append_turn_key: tuple[int, int, int] | None

base_fence instance-attribute 

base_fence: DuplexFence

base_input_seq instance-attribute 

base_input_seq: int

base_input_turn_seq instance-attribute 

base_input_turn_seq: int

fence instance-attribute 

fence: DuplexFence

update instance-attribute 

update: DuplexInputAppend

DuplexEngineSession dataclass

The one session state (see module docstring).

Owned and mutated only by its DuplexSessionRunner on the orchestrator loop.

accepted_fence class-attribute instance-attribute 

accepted_fence: DuplexFence = field(
    default=None, repr=False
)

active_request_id property 

active_request_id: str | None

active_response_id property 

active_response_id: str | None

active_response_turn_id property 

active_response_turn_id: int | None

assistant_audio_text_marks property 

assistant_audio_text_marks: tuple[
    DuplexAssistantAudioTextMark, ...
]

assistant_text_buffer property 

assistant_text_buffer: tuple[str, ...]

capabilities class-attribute instance-attribute 

capabilities: DuplexCapabilities = field(
    default_factory=DuplexCapabilities
)

config instance-attribute 

config: DuplexSessionConfig

config_generation class-attribute instance-attribute 

config_generation: int = 0

created_monotonic class-attribute instance-attribute 

created_monotonic: float = field(
    default_factory=time.monotonic
)

epoch class-attribute instance-attribute 

epoch: int = 0

fence property 

fence: DuplexFence

history property 

history: tuple[dict[str, object], ...]

history_item_ids property 

history_item_ids: Mapping[str, dict[str, object]]

input_commit_seq property 

input_commit_seq: int

input_seq class-attribute instance-attribute 

input_seq: int = 0

input_turn_seq class-attribute instance-attribute 

input_turn_seq: int = 0

last_assistant_audio_text_marks property 

last_assistant_audio_text_marks: tuple[
    DuplexAssistantAudioTextMark, ...
]

last_assistant_full_message property 

last_assistant_full_message: dict[str, object] | None

last_response_id property 

last_response_id: str | None

lease class-attribute instance-attribute 

lease: DuplexLeaseState = field(
    default_factory=_default_lease, repr=False
)

lease_generation property 

lease_generation: int

log_stats class-attribute instance-attribute 

log_stats: bool = False

model_state class-attribute instance-attribute 

model_state: DuplexModelSessionState | None = field(
    default=None, repr=False
)

num_stages class-attribute instance-attribute 

num_stages: int = 1

overlap_speech_ms property 

overlap_speech_ms: int

pending_history_item_ids property 

pending_history_item_ids: Mapping[str, dict[str, object]]

pending_history_truncations_ms property 

pending_history_truncations_ms: Mapping[str, int]

pending_input_bytes property 

pending_input_bytes: int

pending_input_turns property 

pending_input_turns: int

playback property 

playback: DuplexPlaybackView

projector class-attribute instance-attribute 

projector: RealtimeProjectionState | None = field(
    default=None, repr=False
)

request_resources class-attribute instance-attribute 

request_resources: dict[
    tuple[int, str], DuplexRequestResource
] = field(default_factory=dict, repr=False)

response_config property 

response_config: DuplexSessionConfig

Return the immutable-for-this-lifecycle response configuration.

config remains the session defaults. A response takes a deep snapshot when it begins so response.create overrides and concurrent session updates cannot mutate each other's ownership domains.

runtime_config property 

runtime_config: Mapping[str, object]

session_id instance-attribute 

session_id: str

state class-attribute instance-attribute 

state: DuplexSessionState = DuplexSessionState.OPEN

turn_id class-attribute instance-attribute 

turn_id: int = 0

turn_state class-attribute instance-attribute 

turn_state: DuplexTurnState = DuplexTurnState.IDLE

accept_fence 

accept_fence(fence: DuplexFence) -> None

accumulate_overlap_speech 

accumulate_overlap_speech(duration_ms: int) -> int

accumulate_response_stage_metrics 

accumulate_response_stage_metrics(
    stage_metrics: Mapping[Any, Any] | None,
) -> dict[str, dict[str, object]]

acknowledge_playback 

acknowledge_playback(
    played_ms: int,
    committed_ms: int | None = None,
    *,
    response_id: str | None = None,
) -> DuplexPlaybackView

active_response_accepts_model_turn 

active_response_accepts_model_turn(
    turn_id: int | None,
) -> bool

append_assistant_text 

append_assistant_text(text: str) -> None

append_draining_assistant_text 

append_draining_assistant_text(
    response_id: str, text: str
) -> None

Append text onto a snapshotted draining response, not the active one.

append_history_message 

append_history_message(message: dict[str, object]) -> None

as_public_dict 

as_public_dict() -> dict[str, object]

assistant_transcript 

assistant_transcript(response_id: str | None = None) -> str

Joined assistant text for the active response, or a draining snapshot.

barge_in 

barge_in() -> int

begin_close 

begin_close(*, reason: str) -> bool

Make close irreversible while retaining stage resources for cleanup retry.

begin_lease_operation 

begin_lease_operation(
    fence: DuplexFence, operation_id: str
) -> None

begin_response 

begin_response(*, turn_id: int | None = None) -> str

bind_draining_request 

bind_draining_request(
    request_id: str, response_id: str
) -> None

Map a still-playing draining-stage request onto the response that owns it.

bind_request 

bind_request(request_id: str | None) -> None

bind_response_turn 

bind_response_turn(turn_id: int | None) -> None

bind_stage_request 

bind_stage_request(
    stage_id: int, request_id: str, *, fence: DuplexFence
) -> None

cancel_fence 

cancel_fence(
    cancelled_fence: DuplexFence, next_fence: DuplexFence
) -> list[str]

cancel_pending_input 

cancel_pending_input() -> dict[str, int]

Drop every pending reservation; the PCM buffer itself is model state cleared by the runner.

clear_draining_for_response 

clear_draining_for_response(
    response_id: str | None,
) -> None

Drop draining bindings owned by response_id; leave other responses intact.

clear_draining_requests 

clear_draining_requests() -> None

clear_playback_cursor 

clear_playback_cursor() -> None

clear_request 

clear_request(
    expected_request_id: str | None = None,
) -> bool

close 

close() -> None

commit_append 

commit_append(
    reservation: DuplexAppendReservation,
) -> DuplexInputAppend

commit_audio_input 

commit_audio_input(
    *,
    transcript: str | None = None,
    turn_id: int | None = None,
) -> DuplexCommittedInput

complete_model_turn 

complete_model_turn(turn_id: int) -> None

Advance the model-owned output identity after its terminal signal.

delete_history_item 

delete_history_item(item_id: str) -> bool

detach_lease 

detach_lease() -> None

discard_response_options 

discard_response_options() -> None

draining_request_ids 

draining_request_ids() -> list[str]

Request ids whose TTS is still running under a previous response.

end_lease_operation 

end_lease_operation(operation_id: str) -> None

end_response 

end_response(
    *,
    commit_text: bool = True,
    playback_commit_policy: str | None = None,
    preserve_request: bool = False,
) -> dict[str, object] | None

has_assistant_response_item 

has_assistant_response_item(
    response_id: str, item_id: str
) -> bool

is_draining_request 

is_draining_request(request_id: str | None) -> bool

mark_audio_sent 

mark_audio_sent(
    duration_ms: int | None = None,
    *,
    text_chars: int | None = None,
    audio_text_marks: list[dict[str, object]] | None = None,
    text_requires_complete_audio: bool = False,
    audio_complete: bool = False,
    response_id: str | None = None,
) -> None

mark_closing 

mark_closing() -> None

mark_model_turn_request_started 

mark_model_turn_request_started(
    turn_id: int, started_at_s: float
) -> None

Record the native request start that can own one model turn.

Pending turns keep the latest accepted append. After begin_response the first accepted append wins; later appends must not rebind.

mark_response_first_outputs 

mark_response_first_outputs(
    *, observed_at_s: float, has_text: bool, has_audio: bool
) -> dict[str, object]

Return server-monotonic TTF metrics newly observed for the active response.

mark_user_input_activity 

mark_user_input_activity() -> None

notify_new_user_item 

notify_new_user_item() -> None

Record a user item that a later response.create may answer.

observe_stage_request_stats 

observe_stage_request_stats(
    stage_id: int, metrics: StageRequestStats
) -> None

Feed one engine StageRequestStats into the response's logger table.

playback_ack_is_too_late 

playback_ack_is_too_late(
    response_id: str, item_id: str
) -> bool

playback_for_response 

playback_for_response(
    response_id: str | None = None,
) -> DuplexPlaybackView

pop_draining_request 

pop_draining_request(request_id: str | None) -> str | None

prepare_append 

prepare_append(
    fence: DuplexFence,
) -> DuplexAppendReservation

prepare_cancel_fence 

prepare_cancel_fence(
    cancelled_fence: DuplexFence, next_fence: DuplexFence
) -> list[str]

Advance the cancellation fence without dropping cleanup records.

register_history_item 

register_history_item(
    item_id: str | None, message: dict[str, object] | None
) -> None

release_all_input_bytes 

release_all_input_bytes() -> None

release_all_requests 

release_all_requests() -> list[str]

Drop every reserved/submitted stage request; returns their ids for stage cleanup.

release_fence 

release_fence(fence: DuplexFence) -> list[str]

release_finished_drain_response 

release_finished_drain_response(
    response_id: str | None,
) -> None

Drop unused books for a response whose TTS just finished.

Overlap leaves the next turn active, so this must not call end_response. Audio that was already sent stays ACK-admissible: the client may playback.ack after the drain response.done. A response that sent no audio has nothing to acknowledge, so its snapshot, playback cursor, and unused history placeholder are dropped. The live response is left untouched.

release_input_bytes 

release_input_bytes(size: int) -> None

release_pending_turn 

release_pending_turn() -> None

release_resources_for_request_ids 

release_resources_for_request_ids(
    request_ids: Iterable[str],
) -> list[str]

Drop bindings for these request ids, whichever fence they sit on.

cancel_fence only releases the fence being cancelled. Overlapped draining output stages belong to an older turn fence and would otherwise stay until the session closes.

release_response_history_snapshot 

release_response_history_snapshot(
    response_id: str | None,
) -> None

Release a final response snapshot after its playback fully commits.

release_response_playback 

release_response_playback(response_id: str | None) -> None

replace_config 

replace_config(config: DuplexSessionConfig) -> None

replace_response_stage_metric_snapshots 

replace_response_stage_metric_snapshots(
    stage_metrics: Mapping[Any, Any] | None,
) -> dict[str, dict[str, object]]

Merge cumulative chat snapshots by replacing each stage's latest value.

replace_runtime_config 

replace_runtime_config(
    runtime_config: Mapping[str, object],
) -> None

reserve_history_item 

reserve_history_item(item_id: str) -> None

Reserve the current history position for a response-owned item.

reserve_input_bytes 

reserve_input_bytes(size: int, *, limit: int) -> bool

reserve_pending_turn 

reserve_pending_turn(*, limit: int) -> bool

reserve_response_options 

reserve_response_options(
    options: ResponseCreateOptions,
) -> None

reserve_stage_request 

reserve_stage_request(
    stage_id: int, request_id: str, *, fence: DuplexFence
) -> None

reset_overlap_speech 

reset_overlap_speech() -> int

reset_unanswered_user_items 

reset_unanswered_user_items() -> None

A turn has started: the pending items are now its input.

resource_request_ids 

resource_request_ids(
    fence: DuplexFence | None = None,
    *,
    submitted: bool | None = None,
) -> list[str]

response_has_draining_request 

response_has_draining_request(
    response_id: str | None,
) -> bool

response_id_for_request 

response_id_for_request(
    request_id: str | None,
) -> str | None

resume_lease 

resume_lease(*, expected_lease_generation: int) -> int

signal_turn 

signal_turn(
    event_type: str,
    payload: Mapping[str, object] | None = None,
) -> TurnEvent

Apply one external turn signal and return the typed turn.event.

snapshot_active_response_for_drain 

snapshot_active_response_for_drain() -> None

Keep the active response ACK-admissible after a later response starts.

Overlapped commit calls begin_response while prior TTS is still draining. That clears the live text buffer and playback cursor. end_response is the normal snapshot, but it would also close the response. Copy the same snapshot (and reserve the history slot) first so a playback ACK for the draining response is not playback_item_not_found.

stage_request_submitted 

stage_request_submitted(
    stage_id: int, request_id: str
) -> bool

stash_stage_metrics 

stash_stage_metrics(
    stage_metrics: Mapping[Any, Any] | None,
) -> None

Hold a stage snapshot until a response exists to attribute it to.

A stage that hands its output to the next stage instead of the client (stage 0 feeding the TTS stage) reports its token metrics before the response carrying those tokens is created. Dropping them would leave every response without an engine-side token count; holding them keeps the count whole, at the cost of attributing a turn the model never spoke to the next response it does speak.

sync_fence 

sync_fence() -> DuplexFence

Publish the current (epoch, turn_id) identity as the accepted fence.

touch_lease 

touch_lease(activity: DuplexLeaseActivity) -> None

transition_session 

transition_session(state: DuplexSessionState) -> None

transition_turn 

transition_turn(state: DuplexTurnState) -> None

truncate_history_item 

truncate_history_item(
    item_id: str,
    *,
    audio_end_ms: int,
    playback: DuplexPlaybackCursor
    | DuplexPlaybackView
    | None = None,
    hard: bool = False,
) -> bool

truncate_playback_commit 

truncate_playback_commit(
    committed_ms: int, *, response_id: str | None = None
) -> DuplexPlaybackView

unanswered_user_items 

unanswered_user_items() -> int

How many user items are waiting for a response.

DuplexFenceMismatchError

Bases: RuntimeError

actual instance-attribute 

actual = actual

expected instance-attribute 

expected = expected

DuplexInputAppend dataclass

seq instance-attribute 

seq: int

turn_id instance-attribute 

turn_id: int

turn_seq instance-attribute 

turn_seq: int

DuplexRequestResource dataclass

fence instance-attribute 

fence: DuplexFence

request_id instance-attribute 

request_id: str

stage_id instance-attribute 

stage_id: int

submitted class-attribute instance-attribute 

submitted: bool = False

InputBufferState dataclass

commit_seq class-attribute instance-attribute 

commit_seq: int = 0

overlap_speech_ms class-attribute instance-attribute 

overlap_speech_ms: int = 0

pending_turns class-attribute instance-attribute 

pending_turns: int = 0

reserved_input_bytes class-attribute instance-attribute 

reserved_input_bytes: int = 0

PlaybackLedger dataclass

by_response class-attribute instance-attribute 

by_response: dict[str, DuplexPlaybackCursor] = field(
    default_factory=dict
)

current class-attribute instance-attribute 

current: DuplexPlaybackCursor = field(
    default_factory=DuplexPlaybackCursor
)

ResponseState dataclass

active_config class-attribute instance-attribute 

active_config: DuplexSessionConfig | None = None

active_options class-attribute instance-attribute 

active_options: ResponseCreateOptions | None = None

active_request_id class-attribute instance-attribute 

active_request_id: str | None = None

active_response_awaits_input_commit class-attribute instance-attribute 

active_response_awaits_input_commit: bool = False

active_response_id class-attribute instance-attribute 

active_response_id: str | None = None

active_response_input_commit_seq class-attribute instance-attribute 

active_response_input_commit_seq: int | None = None

active_response_request_started_at_s class-attribute instance-attribute 

active_response_request_started_at_s: float | None = None

active_response_ttfp_ms class-attribute instance-attribute 

active_response_ttfp_ms: float | None = None

active_response_ttft_ms class-attribute instance-attribute 

active_response_ttft_ms: float | None = None

active_response_turn_id class-attribute instance-attribute 

active_response_turn_id: int | None = None

assistant_audio_text_marks class-attribute instance-attribute 

assistant_audio_text_marks: list[
    DuplexAssistantAudioTextMark
] = field(default_factory=list)

assistant_text_buffer class-attribute instance-attribute 

assistant_text_buffer: list[str] = field(
    default_factory=list
)

draining_response_by_request class-attribute instance-attribute 

draining_response_by_request: dict[str, str] = field(
    default_factory=dict
)

last_response_id class-attribute instance-attribute 

last_response_id: str | None = None

pending_options class-attribute instance-attribute 

pending_options: ResponseCreateOptions | None = None

request_started_at_s_by_turn class-attribute instance-attribute 

request_started_at_s_by_turn: dict[int, float] = field(
    default_factory=dict
)

stage_metric_tpot_weight class-attribute instance-attribute 

stage_metric_tpot_weight: dict[str, int] = field(
    default_factory=dict
)

stage_metric_tpot_weighted_ms class-attribute instance-attribute 

stage_metric_tpot_weighted_ms: dict[str, float] = field(
    default_factory=dict
)

stage_metrics class-attribute instance-attribute 

stage_metrics: dict[str, dict[str, object]] = field(
    default_factory=dict
)