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
)

heard_prefixes class-attribute instance-attribute 

heard_prefixes: dict[str, tuple[str, int, 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
)

messages class-attribute instance-attribute 

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

pending_item_ids class-attribute instance-attribute 

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

pending_truncations_ms class-attribute instance-attribute 

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

retired_audio_item_ids class-attribute instance-attribute 

retired_audio_item_ids: set[str] = field(
    default_factory=set
)

revision class-attribute instance-attribute 

revision: int = 0

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

config_revision class-attribute instance-attribute 

config_revision: 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]]

history_revision property 

history_revision: int

Monotonic version of committed history, including edits and playback cuts.

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_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_input_bytes property 

pending_input_bytes: int

pending_input_turns property 

pending_input_turns: int

perf class-attribute instance-attribute 

perf: SessionPerfLedger | None = field(
    default=None, repr=False
)

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

apply_heard_text 

apply_heard_text(
    response_id: str,
    *,
    text: str,
    audio_end_ms: int,
    char_end: int,
) -> bool

Validate an async result against the current reply and permanent cutoff.

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_pending_input 

cancel_pending_input() -> dict[str, int]

Drop processed input reservations, not appends still queued in the mailbox.

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

finalize_response_playback 

finalize_response_playback(
    response_id: str | None,
    *,
    require_audio_complete: bool = False,
) -> bool

Reconcile history and release finished, acknowledged or truncated output.

ACK and output completion may arrive in either order. Both paths use this check; neither requires another client ACK to finish cleanup. Calibration retains incomplete evidence in its bounded cache. Returns whether answer text was committed, for plugin settlement.

has_assistant_response_item 

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

has_audio_only_response 

has_audio_only_response(response_id: str) -> bool

history_audio_cutoff 

history_audio_cutoff(response_id: str) -> int

history_calibration_retired 

history_calibration_retired(response_id: str) -> bool

history_item_has_audio 

history_item_has_audio(item_id: str) -> bool

Whether this item's response produced audio, including after cleanup.

A transcript or an empty cursor does not establish audio ownership. Retired provenance follows the stored item's lifetime, not an ACK's.

history_snapshot 

history_snapshot() -> tuple[
    tuple[dict[str, object], ...], tuple[str | None, ...]
]

Snapshot committed messages and their aligned, optional item identities.

input_audio_part 

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

Return the native audio part whose identity owns this item's ASR.

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_requests 

release_all_requests() -> list[str]

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

release_buffered_input_bytes 

release_buffered_input_bytes() -> None

release_fence 

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

release_finished_drain_response 

release_finished_drain_response(
    response_id: str | None,
    *,
    require_audio_complete: bool = False,
) -> bool

Finish a natural TTS drain, retaining any audio still awaiting ACK.

release_input_bytes 

release_input_bytes(
    size: int, *, queued: bool = False
) -> 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.

Fence-scoped cancellation 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_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, queued: bool = False
) -> 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

restore_config 

restore_config(previous: DuplexSessionConfig) -> None

Undo a replace_config whose candidate never finished adopting.

The forward replace already assigned the candidate, bumped config_generation, and may have counted its seed text. Putting the previous object back through replace_config would count that seed again and advance the generation a second time.

resume_lease 

resume_lease(
    *,
    expected_lease_generation: int,
    control_id: str | None = None,
) -> int

retire_history_calibration 

retire_history_calibration(response_id: str) -> None

Release retired evidence while keeping the committed turn immutable.

Only the stored prefix and its evidence cutoff survive with the history item. Without the original recording, an earlier truncate must clear that prefix rather than estimate a new alignment or restore full text.

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

Apply this item's heard prefix; a hard cut without audio is a no-op.

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.

update_input_transcript 

update_input_transcript(
    item_id: str,
    transcript: str,
    *,
    audio_part: dict[str, object] | None,
) -> bool

Attach ASR text only to the audio part captured for this request.

Public item IDs can be deleted and reused. A fresh native placeholder has the same value, but must not receive the old audio's transcript.

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

input_audio_part class-attribute instance-attribute 

input_audio_part: dict[str, object] | None = field(
    default=None, repr=False
)

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

queued_input_bytes class-attribute instance-attribute 

queued_input_bytes: 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_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
)