Skip to content

Chat API Reference

The chat subsystem provides interactive, multi-turn conversation capabilities for HoloDeck agents. It coordinates message validation, agent execution via the provider-agnostic backend layer, streaming responses, tool execution progress tracking, and session lifecycle management.

Module: holodeck.chat.executor

Orchestrates agent execution for chat sessions using the backend abstraction layer (AgentBackend / AgentSession). Supports both synchronous turn-based and streaming response modes, with lazy backend initialization and optional task-bound session wrapping for HTTP server contexts.

AgentResponse

AgentResponse(content, tool_executions, tokens_used, execution_time) dataclass

Response from agent execution.

Contains the agent's text response, any tool executions performed, token usage tracking, and execution timing information.

AgentExecutor

AgentExecutor(agent_config, backend=None, on_execution_start=None, on_execution_complete=None, release_transport_after_turn=False)

Coordinates agent execution for chat sessions.

Uses the provider-agnostic AgentBackend/AgentSession abstractions to execute user messages and manage conversation history.

Initialize executor with agent configuration.

No I/O is performed during construction — backend and session are lazily created on the first execute_turn() call.

Parameters:

Name Type Description Default
agent_config Agent

Agent configuration with model and instructions.

required
backend AgentBackend | None

Optional pre-initialized backend (bypasses BackendSelector).

None
on_execution_start Callable[[str], None] | None

Optional callback before agent execution.

None
on_execution_complete Callable[[AgentResponse], None] | None

Optional callback after agent execution.

None
release_transport_after_turn bool

If True, wrap the backend session in a _TaskBoundSession actor so the SDK client stays in a single background task. Required for HTTP servers where each request runs in a different async task context.

False
Source code in src/holodeck/chat/executor.py
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
def __init__(
    self,
    agent_config: Agent,
    backend: AgentBackend | None = None,
    on_execution_start: Callable[[str], None] | None = None,
    on_execution_complete: Callable[[AgentResponse], None] | None = None,
    release_transport_after_turn: bool = False,
) -> None:
    """Initialize executor with agent configuration.

    No I/O is performed during construction — backend and session
    are lazily created on the first ``execute_turn()`` call.

    Args:
        agent_config: Agent configuration with model and instructions.
        backend: Optional pre-initialized backend (bypasses BackendSelector).
        on_execution_start: Optional callback before agent execution.
        on_execution_complete: Optional callback after agent execution.
        release_transport_after_turn: If True, wrap the backend session
            in a ``_TaskBoundSession`` actor so the SDK client stays in
            a single background task.  Required for HTTP servers where
            each request runs in a different async task context.
    """
    self.agent_config = agent_config
    self._backend: AgentBackend | None = backend
    self._session: AgentSession | None = None
    self._history: list[dict[str, Any]] = []
    self.on_execution_start = on_execution_start
    self.on_execution_complete = on_execution_complete
    self._use_task_bound_session = release_transport_after_turn

    logger.info(f"AgentExecutor initialized for agent: {agent_config.name}")

tool_event_queue property

The tool event queue from the underlying session, if available.

clear_history() async

Clear conversation history and close the current session.

Resets the agent's chat history to start fresh conversation. The next execute_turn() will create a new session.

Source code in src/holodeck/chat/executor.py
315
316
317
318
319
320
321
322
323
324
325
async def clear_history(self) -> None:
    """Clear conversation history and close the current session.

    Resets the agent's chat history to start fresh conversation.
    The next ``execute_turn()`` will create a new session.
    """
    logger.debug("Clearing chat history and closing session")
    if self._session is not None:
        await self._session.close()
        self._session = None
    self._history = []

execute_turn(message) async

Execute a single turn of agent conversation.

Sends a user message to the agent, captures the response, extracts tool calls, and tracks token usage.

Parameters:

Name Type Description Default
message str

User message to send to the agent.

required

Returns:

Type Description
AgentResponse

AgentResponse with content, tool executions, tokens, and timing.

Raises:

Type Description
RuntimeError

If agent execution fails.

Source code in src/holodeck/chat/executor.py
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
async def execute_turn(self, message: str) -> AgentResponse:
    """Execute a single turn of agent conversation.

    Sends a user message to the agent, captures the response,
    extracts tool calls, and tracks token usage.

    Args:
        message: User message to send to the agent.

    Returns:
        AgentResponse with content, tool executions, tokens, and timing.

    Raises:
        RuntimeError: If agent execution fails.
    """
    start_time = time.time()

    try:
        logger.debug(f"Executing turn for agent: {self.agent_config.name}")

        # Call pre-execution callback
        if self.on_execution_start:
            self.on_execution_start(message)

        # Lazy initialize backend and session
        await self._ensure_backend_and_session()

        # Invoke agent via backend session
        result: ExecutionResult = await self._session.send(message)  # type: ignore[union-attr]
        elapsed = time.time() - start_time

        # Extract content from execution result
        content = result.response

        # Convert tool calls to ToolExecution models
        tool_executions = self._convert_tool_calls(result.tool_calls)

        # Extract token usage (always a TokenUsage, never None)
        tokens_used = result.token_usage

        logger.debug(
            f"Turn executed successfully: content={len(content)} chars, "
            f"tools={len(tool_executions)}, time={elapsed:.2f}s"
        )

        # Track history
        self._history.append({"role": "user", "content": message})
        self._history.append({"role": "assistant", "content": content})

        response = AgentResponse(
            content=content,
            tool_executions=tool_executions,
            tokens_used=tokens_used,
            execution_time=elapsed,
        )

        # Call post-execution callback
        if self.on_execution_complete:
            self.on_execution_complete(response)

        return response

    except BackendSessionError:
        raise
    except BackendInitError as e:
        logger.error(f"Agent execution failed: {e}", exc_info=True)
        raise RuntimeError(f"Agent execution failed: {e}") from e
    except RuntimeError:
        raise
    except Exception as e:
        logger.error(f"Agent execution failed: {e}", exc_info=True)
        raise RuntimeError(f"Agent execution failed: {e}") from e

execute_turn_streaming(message) async

Stream agent response token by token.

Parameters:

Name Type Description Default
message str

User message to send to the agent.

required

Yields:

Type Description
AsyncGenerator[str, None]

Successive string chunks of the agent response.

Raises:

Type Description
RuntimeError

If agent execution fails.

Source code in src/holodeck/chat/executor.py
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
async def execute_turn_streaming(self, message: str) -> AsyncGenerator[str, None]:
    """Stream agent response token by token.

    Args:
        message: User message to send to the agent.

    Yields:
        Successive string chunks of the agent response.

    Raises:
        RuntimeError: If agent execution fails.
    """
    try:
        await self._ensure_backend_and_session()
        collected: list[str] = []
        async for chunk in self._session.send_streaming(message):  # type: ignore[union-attr]
            collected.append(chunk)
            yield chunk
        # Update history after stream completes
        self._history.append({"role": "user", "content": message})
        self._history.append({"role": "assistant", "content": "".join(collected)})
    except BackendSessionError:
        raise
    except BackendInitError as e:
        raise RuntimeError(f"Agent streaming failed: {e}") from e

get_history()

Get current conversation history.

Returns:

Type Description
list[dict[str, Any]]

Serialized conversation history as a list of dicts, or empty list.

Source code in src/holodeck/chat/executor.py
307
308
309
310
311
312
313
def get_history(self) -> list[dict[str, Any]]:
    """Get current conversation history.

    Returns:
        Serialized conversation history as a list of dicts, or empty list.
    """
    return list(self._history)

shutdown() async

Cleanup executor resources.

Called when ending a chat session to release any held resources. Closes the session and tears down the backend.

Source code in src/holodeck/chat/executor.py
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
async def shutdown(self) -> None:
    """Cleanup executor resources.

    Called when ending a chat session to release any held resources.
    Closes the session and tears down the backend.
    """
    try:
        logger.debug("AgentExecutor shutting down")
        if self._session is not None:
            await self._session.close()
            self._session = None
        if self._backend is not None:
            await self._backend.teardown()
            self._backend = None
        logger.debug("AgentExecutor shutdown complete")
    except Exception as e:
        logger.error(f"Error during shutdown: {e}")

Module: holodeck.chat.message

Validates user messages before they reach the agent, enforcing content standards such as empty-message detection, size limits, control-character filtering, and UTF-8 validation.

MessageValidator

MessageValidator(max_length=10000)

Validates user messages before sending to the agent.

Uses ValidationPipeline to enforce content standards including empty message detection, size limits, control character filtering, and UTF-8 validation.

Initialize validator with length constraints.

Parameters:

Name Type Description Default
max_length int

Maximum message length in characters. Defaults to 10,000.

10000
Source code in src/holodeck/chat/message.py
16
17
18
19
20
21
22
def __init__(self, max_length: int = 10_000) -> None:
    """Initialize validator with length constraints.

    Args:
        max_length: Maximum message length in characters. Defaults to 10,000.
    """
    self._pipeline = ValidationPipeline(max_length=max_length)

validate(message)

Validate a message and return validation status.

Parameters:

Name Type Description Default
message str | None

User message to validate (None, empty, or any content).

required

Returns:

Type Description
bool

Tuple of (is_valid: bool, error_message: str | None).

str | None

If valid, error_message is None.

tuple[bool, str | None]

If invalid, error_message describes the validation failure.

Validation checks: - Message is not None or empty - Message does not exceed max_length - Message contains no control characters - Message is valid UTF-8

Source code in src/holodeck/chat/message.py
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
def validate(self, message: str | None) -> tuple[bool, str | None]:
    """Validate a message and return validation status.

    Args:
        message: User message to validate (None, empty, or any content).

    Returns:
        Tuple of (is_valid: bool, error_message: str | None).
        If valid, error_message is None.
        If invalid, error_message describes the validation failure.

    Validation checks:
    - Message is not None or empty
    - Message does not exceed max_length
    - Message contains no control characters
    - Message is valid UTF-8
    """
    return self._pipeline.validate(message)

Module: holodeck.chat.session

Manages chat session lifecycle and state, coordinating between message validation, agent execution, token tracking, and session statistics.

ChatSessionManager

ChatSessionManager(agent_config, config)

Maintains chat session lifecycle and state management.

Coordinates between message validation, agent execution, and session state tracking.

Initialize session manager with configuration.

Parameters:

Name Type Description Default
agent_config Agent

Agent configuration to use for execution.

required
config ChatConfig

Chat runtime configuration.

required
Source code in src/holodeck/chat/session.py
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
def __init__(self, agent_config: Agent, config: ChatConfig) -> None:
    """Initialize session manager with configuration.

    Args:
        agent_config: Agent configuration to use for execution.
        config: Chat runtime configuration.
    """
    self.agent_config = agent_config
    self.config = config
    self.session: ChatSession | None = None
    self._executor: AgentExecutor | None = None
    self._validator = MessageValidator(max_length=10_000)

    # Token tracking
    self.total_tokens = TokenUsage(
        prompt_tokens=0,
        completion_tokens=0,
        total_tokens=0,
    )
    self.session_start_time = datetime.now()

    logger.debug(f"ChatSessionManager initialized for agent: {agent_config.name}")

get_session()

Get current chat session.

Returns:

Type Description
ChatSession | None

ChatSession instance, or None if not started.

Source code in src/holodeck/chat/session.py
154
155
156
157
158
159
160
def get_session(self) -> ChatSession | None:
    """Get current chat session.

    Returns:
        ChatSession instance, or None if not started.
    """
    return self.session

get_session_stats()

Get current session statistics.

Returns:

Type Description
dict[str, Any]

Dict with message_count, total_tokens, and session_duration.

Source code in src/holodeck/chat/session.py
162
163
164
165
166
167
168
169
170
171
172
173
def get_session_stats(self) -> dict[str, Any]:
    """Get current session statistics.

    Returns:
        Dict with message_count, total_tokens, and session_duration.
    """
    session_duration = (datetime.now() - self.session_start_time).total_seconds()
    return {
        "message_count": self.session.message_count if self.session else 0,
        "total_tokens": self.total_tokens,
        "session_duration": session_duration,
    }

process_message(message) async

Process a user message through validation and execution.

Parameters:

Name Type Description Default
message str

User message to process.

required

Returns:

Type Description
AgentResponse

AgentResponse from the agent.

Raises:

Type Description
RuntimeError

If session not started or execution fails.

ValueError

If message validation fails.

Source code in src/holodeck/chat/session.py
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
async def process_message(self, message: str) -> AgentResponse:
    """Process a user message through validation and execution.

    Args:
        message: User message to process.

    Returns:
        AgentResponse from the agent.

    Raises:
        RuntimeError: If session not started or execution fails.
        ValueError: If message validation fails.
    """
    if self.session is None or self._executor is None:
        raise RuntimeError("Session not started. Call start() first.")

    # Validate message
    is_valid, error = self._validator.validate(message)
    if not is_valid:
        raise ValueError(f"Invalid message: {error}")

    # Execute turn
    try:
        logger.debug(f"Processing message ({len(message)} chars)")
        response = await self._executor.execute_turn(message)

        # Increment message count
        self.session.message_count += 1

        # Accumulate token usage
        if response.tokens_used:
            self.total_tokens = TokenUsage(
                prompt_tokens=self.total_tokens.prompt_tokens
                + response.tokens_used.prompt_tokens,
                completion_tokens=self.total_tokens.completion_tokens
                + response.tokens_used.completion_tokens,
                total_tokens=self.total_tokens.total_tokens
                + response.tokens_used.total_tokens,
            )

        logger.debug(
            f"Message processed: count={self.session.message_count}, "
            f"tools={len(response.tool_executions)}"
        )

        return response

    except Exception as e:
        logger.error(f"Message processing failed: {e}", exc_info=True)
        raise RuntimeError(f"Message processing failed: {e}") from e

process_message_streaming(message) async

Stream a user message through validation and agent execution.

Parameters:

Name Type Description Default
message str

User message to process.

required

Yields:

Type Description
AsyncGenerator[str, None]

Successive string chunks of the agent response.

Raises:

Type Description
RuntimeError

If session not started or execution fails.

ValueError

If message validation fails.

Source code in src/holodeck/chat/session.py
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
async def process_message_streaming(
    self, message: str
) -> AsyncGenerator[str, None]:
    """Stream a user message through validation and agent execution.

    Args:
        message: User message to process.

    Yields:
        Successive string chunks of the agent response.

    Raises:
        RuntimeError: If session not started or execution fails.
        ValueError: If message validation fails.
    """
    if self.session is None or self._executor is None:
        raise RuntimeError("Session not started. Call start() first.")

    # Validate message (same as process_message)
    is_valid, error = self._validator.validate(message)
    if not is_valid:
        raise ValueError(f"Invalid message: {error}")

    logger.debug(f"Streaming message ({len(message)} chars)")
    async for chunk in self._executor.execute_turn_streaming(message):
        yield chunk

    # Increment message count after stream completes
    self.session.message_count += 1
    logger.debug(f"Streamed message: count={self.session.message_count}")

should_warn_context_limit()

Check if conversation is approaching context limit.

Returns:

Type Description
bool

True if message count >= 80% of max_messages, False otherwise.

Source code in src/holodeck/chat/session.py
175
176
177
178
179
180
181
182
183
184
185
186
def should_warn_context_limit(self) -> bool:
    """Check if conversation is approaching context limit.

    Returns:
        True if message count >= 80% of max_messages, False otherwise.
    """
    if self.session is None:
        return False

    threshold = int(self.config.max_messages * 0.8)
    should_warn = self.session.message_count >= threshold
    return should_warn

start() async

Start a new chat session.

Initializes the agent executor, creates a chat session, and transitions state to ACTIVE.

Raises:

Type Description
RuntimeError

If session initialization fails.

Source code in src/holodeck/chat/session.py
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
async def start(self) -> None:
    """Start a new chat session.

    Initializes the agent executor, creates a chat session,
    and transitions state to ACTIVE.

    Raises:
        RuntimeError: If session initialization fails.
    """
    logger.info(f"Starting chat session for agent: {self.agent_config.name}")

    # Constructor does no I/O — errors surface on first turn (lazy-init)
    self._executor = AgentExecutor(self.agent_config)

    # Create chat session with empty history
    self.session = ChatSession(
        agent_config=self.agent_config,
        history=[],
        state=SessionState.ACTIVE,
    )

    logger.info(f"Chat session started: session_id={self.session.session_id}")

terminate() async

Terminate the chat session.

Cleans up resources and transitions state to TERMINATED.

Source code in src/holodeck/chat/session.py
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
async def terminate(self) -> None:
    """Terminate the chat session.

    Cleans up resources and transitions state to TERMINATED.
    """
    try:
        if self.session is not None:
            self.session.state = SessionState.TERMINATED
            logger.info(
                f"Chat session terminated: session_id={self.session.session_id}"
            )

        if self._executor is not None:
            await self._executor.shutdown()

    except Exception as e:
        logger.error(f"Error during session termination: {e}", exc_info=True)

Module: holodeck.chat.streaming

Streams tool execution events to callers in real time, allowing UIs to display progress as tools start, run, and complete (or fail).

ToolExecutionStream

ToolExecutionStream(verbose=False)

Streams tool execution events to the caller.

Emits ToolEvent instances as a tool executes, allowing callers to display real-time progress. Supports both standard and verbose modes.

Initialize the stream with verbosity preference.

Parameters:

Name Type Description Default
verbose bool

If True, include detailed execution data (parameters, results). If False, emit minimal data (tool name, status, timing).

False
Source code in src/holodeck/chat/streaming.py
22
23
24
25
26
27
28
29
def __init__(self, verbose: bool = False) -> None:
    """Initialize the stream with verbosity preference.

    Args:
        verbose: If True, include detailed execution data (parameters, results).
                If False, emit minimal data (tool name, status, timing).
    """
    self.verbose = verbose

stream_execution(tool_call) async

Stream execution events for a tool call.

Simulates the execution lifecycle by emitting events: 1. STARTED - Tool execution begins 2. PROGRESS - (optional, for long operations) 3. COMPLETED or FAILED - Execution finished

Parameters:

Name Type Description Default
tool_call ToolExecution

Tool execution with status and result data.

required

Yields:

Type Description
AsyncIterator[ToolEvent]

ToolEvent instances representing execution progression.

Source code in src/holodeck/chat/streaming.py
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
async def stream_execution(
    self, tool_call: ToolExecution
) -> AsyncIterator[ToolEvent]:
    """Stream execution events for a tool call.

    Simulates the execution lifecycle by emitting events:
    1. STARTED - Tool execution begins
    2. PROGRESS - (optional, for long operations)
    3. COMPLETED or FAILED - Execution finished

    Args:
        tool_call: Tool execution with status and result data.

    Yields:
        ToolEvent instances representing execution progression.
    """
    # Emit STARTED event
    yield ToolEvent(
        event_type=ToolEventType.STARTED,
        tool_name=tool_call.tool_name,
        timestamp=datetime.utcnow(),
        data={"parameters": tool_call.parameters} if self.verbose else {},
    )

    # Emit COMPLETED or FAILED based on status
    if tool_call.status == ToolStatus.SUCCESS:
        yield ToolEvent(
            event_type=ToolEventType.COMPLETED,
            tool_name=tool_call.tool_name,
            timestamp=datetime.utcnow(),
            data={
                "result": tool_call.result if self.verbose else None,
                "execution_time": tool_call.execution_time or 0,
            },
        )
    elif tool_call.status == ToolStatus.FAILED:
        yield ToolEvent(
            event_type=ToolEventType.FAILED,
            tool_name=tool_call.tool_name,
            timestamp=datetime.utcnow(),
            data={
                "error": tool_call.error_message or "Unknown error",
            },
        )
    else:
        # For PENDING or RUNNING, emit as PROGRESS
        yield ToolEvent(
            event_type=ToolEventType.PROGRESS,
            tool_name=tool_call.tool_name,
            timestamp=datetime.utcnow(),
            data={"status": tool_call.status.value},
        )

ToolEvent and ToolEventType

ToolEvent and ToolEventType are re-exported from holodeck.models.tool_event. See the Models API Reference for full documentation.

Module: holodeck.chat.progress

Tracks and displays chat session progress with animated spinners and adaptive status output (inline for default mode, rich panel for verbose mode).

ChatProgressIndicator

ChatProgressIndicator(max_messages, quiet, verbose)

Bases: SpinnerMixin

Track and display chat session progress with spinner and status information.

Provides animated spinner during agent execution and adaptive status display (minimal in default mode, rich in verbose mode). Tracks message count, tokens, session time, response timing, and tool executions.

Inherits spinner animation from SpinnerMixin.

Initialize progress indicator.

Parameters:

Name Type Description Default
max_messages int

Maximum messages for session before warning.

required
quiet bool

Suppress status display (spinner still shows).

required
verbose bool

Show rich status panel instead of inline status.

required
Source code in src/holodeck/chat/progress.py
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
def __init__(self, max_messages: int, quiet: bool, verbose: bool) -> None:
    """Initialize progress indicator.

    Args:
        max_messages: Maximum messages for session before warning.
        quiet: Suppress status display (spinner still shows).
        verbose: Show rich status panel instead of inline status.
    """
    self.max_messages = max_messages
    self.quiet = quiet
    self.verbose = verbose

    # State tracking
    self.current_messages = 0
    self.total_tokens = TokenUsage.zero()
    self.last_response_time: float | None = None
    self.session_start = datetime.now()
    self._spinner_index = 0

    # Tool execution tracking
    self.last_tool_executions: list[ToolExecution] = []
    self.total_tool_calls: int = 0

get_spinner_char()

Get current spinner character and advance rotation.

Returns:

Type Description
str

Current spinner character from the braille sequence.

Source code in src/holodeck/lib/ui/spinner.py
36
37
38
39
40
41
42
43
44
def get_spinner_char(self) -> str:
    """Get current spinner character and advance rotation.

    Returns:
        Current spinner character from the braille sequence.
    """
    char = self.SPINNER_CHARS[self._spinner_index % len(self.SPINNER_CHARS)]
    self._spinner_index += 1
    return char

get_spinner_line()

Get current spinner animation frame.

Returns:

Type Description
str

Animated spinner text, or empty string if not TTY.

Source code in src/holodeck/chat/progress.py
46
47
48
49
50
51
52
53
54
55
56
def get_spinner_line(self) -> str:
    """Get current spinner animation frame.

    Returns:
        Animated spinner text, or empty string if not TTY.
    """
    if not is_tty():
        return ""

    spinner_char = self.get_spinner_char()
    return f"{spinner_char} Thinking..."

get_status_inline()

Get minimal inline status for default mode.

Format: [messages_current/messages_max | execution_time]

Returns:

Type Description
str

Inline status string.

Source code in src/holodeck/chat/progress.py
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
def get_status_inline(self) -> str:
    """Get minimal inline status for default mode.

    Format: [messages_current/messages_max | execution_time]

    Returns:
        Inline status string.
    """
    status_parts = []

    # Message count
    status_parts.append(f"{self.current_messages}/{self.max_messages}")

    # Execution time
    if self.last_response_time is not None:
        time_str = f"{self.last_response_time:.1f}s"
        status_parts.append(time_str)

    return f"[{' | '.join(status_parts)}]" if status_parts else ""

get_status_panel()

Get rich status panel for verbose mode.

Returns:

Type Description
str

Multi-line status panel string.

Source code in src/holodeck/chat/progress.py
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
def get_status_panel(self) -> str:
    """Get rich status panel for verbose mode.

    Returns:
        Multi-line status panel string.
    """
    lines = []
    content_width = 39  # Width of content area (excluding "│ " and " │")

    # Top border
    lines.append("╭─── Chat Status ─────────────────────────╮")

    # Session time
    session_duration = (datetime.now() - self.session_start).total_seconds()
    hours = int(session_duration // 3600)
    minutes = int((session_duration % 3600) // 60)
    seconds = int(session_duration % 60)
    time_str = f"{hours:02d}:{minutes:02d}:{seconds:02d}"
    content = f"Session Time: {time_str}"
    lines.append(f"│ {content:<{content_width}} │")

    # Message count with percentage
    percentage = int((self.current_messages / self.max_messages) * 100)
    msg_str = f"{self.current_messages} / {self.max_messages} ({percentage}%)"
    content = f"Messages: {msg_str}"
    lines.append(f"│ {content:<{content_width}} │")

    # Token usage
    total_str = f"{self.total_tokens.total_tokens:,}"
    content = f"Total Tokens: {total_str}"
    lines.append(f"│ {content:<{content_width}} │")

    # Token breakdown - prompt
    prompt_str = f"{self.total_tokens.prompt_tokens:,}"
    content = f"  ├─ Prompt: {prompt_str}"
    lines.append(f"│ {content:<{content_width}} │")

    # Token breakdown - completion
    completion_str = f"{self.total_tokens.completion_tokens:,}"
    content = f"  └─ Completion: {completion_str}"
    lines.append(f"│ {content:<{content_width}} │")

    # Last response time
    if self.last_response_time is not None:
        time_str = f"{self.last_response_time:.1f}s"
        content = f"Last Response: {time_str}"
        lines.append(f"│ {content:<{content_width}} │")

    # Tool execution summary
    content = f"Tool Calls (Total): {self.total_tool_calls}"
    lines.append(f"│ {content:<{content_width}} │")

    # Last tool executions (if any)
    if self.last_tool_executions:
        lines.append(f"│ {'Last Tools Called:':<{content_width}} │")
        for tool_exec in self.last_tool_executions:
            # Get status indicator
            status_icon = self._get_status_icon(tool_exec.status)
            # Truncate tool name if too long
            tool_name = tool_exec.tool_name
            max_name_len = content_width - 6  # Account for "  └─ " and icon
            if len(tool_name) > max_name_len:
                tool_name = tool_name[: max_name_len - 3] + "..."
            content = f"  └─ {status_icon} {tool_name}"
            lines.append(f"│ {content:<{content_width}} │")

    # Bottom border
    lines.append("╰─────────────────────────────────────────╯")

    return "\n".join(lines)

update(response)

Update progress after agent response.

Parameters:

Name Type Description Default
response Any

AgentResponse object with execution_time, tokens_used, and tool_executions.

required
Source code in src/holodeck/chat/progress.py
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
def update(self, response: Any) -> None:
    """Update progress after agent response.

    Args:
        response: AgentResponse object with execution_time, tokens_used,
            and tool_executions.
    """
    # Update message count
    self.current_messages += 1

    # Update execution time
    if hasattr(response, "execution_time"):
        self.last_response_time = response.execution_time

    # Accumulate token usage
    if hasattr(response, "tokens_used") and response.tokens_used:
        self.total_tokens = self.total_tokens + response.tokens_used

    # Track tool executions
    if hasattr(response, "tool_executions") and response.tool_executions:
        self.last_tool_executions = response.tool_executions
        self.total_tool_calls += len(response.tool_executions)
    else:
        self.last_tool_executions = []