Skip to content

state

code_context_agent.consumer.state

Mutable display state for agent event rendering.

This module provides Pydantic models for tracking the current state of agent execution, used by consumers to render live updates.

ToolCallState

Bases: StrictModel

State for a single tool call.

Attributes:

Name Type Description
tool_call_id str

Unique identifier for the tool call.

tool_name str

Name of the tool being executed.

args_buffer str

Accumulated tool arguments (streaming).

result Any

Tool execution result (when complete).

status str

Current status ("running", "completed", "error").

AgentDisplayState

Bases: StrictModel

Mutable state for agent display rendering.

This class tracks all state needed to render the agent's progress in a terminal or UI. It accumulates streaming text, tracks active and completed tool calls, and maintains error state.

Attributes:

Name Type Description
current_phase str

Description of current analysis phase.

text_buffer str

Accumulated streaming text from agent.

active_message_id str | None

ID of currently streaming message.

active_tool ToolCallState | None

Currently executing tool state (if any).

completed_tools list[ToolCallState]

List of completed tool executions.

state_snapshot dict[str, Any]

Latest state snapshot from agent.

error str | None

Error message if run failed.

completed bool

Whether the run has finished.

thread_id str | None

Current thread identifier.

run_id str | None

Current run identifier.

Example

state = AgentDisplayState() state.text_buffer += "Analyzing repository..." state.active_tool = ToolCallState(tool_call_id="t1", tool_name="rg_search")

clear_text_buffer

clear_text_buffer()

Clear and return the text buffer.

Returns:

Type Description
str

The text buffer contents before clearing.

Source code in src/code_context_agent/consumer/state.py
def clear_text_buffer(self) -> str:
    """Clear and return the text buffer.

    Returns:
        The text buffer contents before clearing.
    """
    text = self.text_buffer
    self.text_buffer = ""
    return text

complete_active_tool

complete_active_tool(result=None)

Mark the active tool as completed and move to history.

Parameters:

Name Type Description Default
result Any

Optional result to store with the tool call.

None
Source code in src/code_context_agent/consumer/state.py
def complete_active_tool(self, result: Any = None) -> None:
    """Mark the active tool as completed and move to history.

    Args:
        result: Optional result to store with the tool call.
    """
    if self.active_tool:
        self.active_tool.status = "completed"
        self.active_tool.result = result
        self.completed_tools.append(self.active_tool)
        self.active_tool = None

get_recent_tools

get_recent_tools(count=5)

Get the most recent completed tools.

Parameters:

Name Type Description Default
count int

Maximum number of tools to return.

5

Returns:

Type Description
list[ToolCallState]

List of recently completed tool states.

Source code in src/code_context_agent/consumer/state.py
def get_recent_tools(self, count: int = 5) -> list[ToolCallState]:
    """Get the most recent completed tools.

    Args:
        count: Maximum number of tools to return.

    Returns:
        List of recently completed tool states.
    """
    return self.completed_tools[-count:]

get_tool_stats

get_tool_stats()

Get tool call counts grouped by prefix.

Groups tools by their name prefix (e.g., lsp_, code_graph_) for display in the TUI dashboard.

Returns:

Type Description
dict[str, int]

Dictionary mapping tool prefixes to call counts.

Source code in src/code_context_agent/consumer/state.py
def get_tool_stats(self) -> dict[str, int]:
    """Get tool call counts grouped by prefix.

    Groups tools by their name prefix (e.g., lsp_*, code_graph_*) for
    display in the TUI dashboard.

    Returns:
        Dictionary mapping tool prefixes to call counts.
    """
    stats: dict[str, int] = {}
    for tool in self.completed_tools:
        # Group by first part of tool name (before underscore)
        parts = tool.tool_name.split("_")
        prefix = parts[0] + "_*" if len(parts) > 1 else tool.tool_name
        stats[prefix] = stats.get(prefix, 0) + 1
    return stats

get_elapsed_seconds

get_elapsed_seconds()

Get elapsed time since run started.

Returns:

Type Description
float

Elapsed seconds, or 0.0 if not started.

Source code in src/code_context_agent/consumer/state.py
def get_elapsed_seconds(self) -> float:
    """Get elapsed time since run started.

    Returns:
        Elapsed seconds, or 0.0 if not started.
    """
    if self.start_time is None:
        return 0.0
    return time.monotonic() - self.start_time

get_tool_elapsed_seconds

get_tool_elapsed_seconds()

Get elapsed time for current tool.

Returns:

Type Description
float

Elapsed seconds for active tool, or 0.0 if no active tool.

Source code in src/code_context_agent/consumer/state.py
def get_tool_elapsed_seconds(self) -> float:
    """Get elapsed time for current tool.

    Returns:
        Elapsed seconds for active tool, or 0.0 if no active tool.
    """
    if self.tool_start_time is None:
        return 0.0
    return time.monotonic() - self.tool_start_time

get_success_count

get_success_count()

Get count of successful tool calls.

Returns:

Type Description
int

Number of completed tools minus errors.

Source code in src/code_context_agent/consumer/state.py
def get_success_count(self) -> int:
    """Get count of successful tool calls.

    Returns:
        Number of completed tools minus errors.
    """
    return len(self.completed_tools) - self.tool_errors

advance_phase

advance_phase(phase)

Advance to a new phase if it's higher than the current one.

Parameters:

Name Type Description Default
phase Any

AnalysisPhase enum value to advance to.

required
Source code in src/code_context_agent/consumer/state.py
def advance_phase(self, phase: Any) -> None:
    """Advance to a new phase if it's higher than the current one.

    Args:
        phase: AnalysisPhase enum value to advance to.
    """
    from .phases import PHASE_DESCRIPTIONS, PHASE_NAMES, PhaseState

    # Don't regress to a lower phase
    if self.phases and phase <= self.phases[-1].phase:
        return

    # Complete the previous phase
    now = time.monotonic()
    if self.phases and not self.phases[-1].is_complete:
        self.phases[-1].completed_at = now

    # Start the new phase
    self.phases.append(
        PhaseState(
            phase=phase,
            name=PHASE_NAMES.get(phase, f"Phase {phase}"),
            description=PHASE_DESCRIPTIONS.get(phase, ""),
            started_at=now,
        ),
    )
    self.current_phase_index = len(self.phases) - 1

add_discovery

add_discovery(event)

Add a discovery event, evicting oldest if at capacity.

Parameters:

Name Type Description Default
event Any

DiscoveryEvent to add.

required
Source code in src/code_context_agent/consumer/state.py
def add_discovery(self, event: Any) -> None:
    """Add a discovery event, evicting oldest if at capacity.

    Args:
        event: DiscoveryEvent to add.
    """
    self.discoveries.append(event)
    # Capped list — evict oldest when over capacity
    while len(self.discoveries) > self.max_discoveries:
        self.discoveries.pop(0)

reset

reset()

Reset state for a new run.

Source code in src/code_context_agent/consumer/state.py
def reset(self) -> None:
    """Reset state for a new run."""
    self.current_phase = ""
    self.text_buffer = ""
    self.active_message_id = None
    self.active_tool = None
    self.completed_tools = []
    self.state_snapshot = {}
    self.error = None
    self.completed = False
    # Reset metrics
    self.start_time = None
    self.turn_count = 0
    self.tool_errors = 0
    self.tool_start_time = None
    # Reset phase tracking (v7)
    self.phases = []
    self.current_phase_index = -1
    self.discoveries = []