primus.console.chat_runtime module

class primus.console.chat_runtime.ConsoleMessage(id: 'str', account_id: 'str', session_id: 'str', procedure_id: 'str', content: 'str', role: 'str', message_type: 'str', human_interaction: 'str', response_target: 'str', response_status: 'str', created_at: 'str', selected_model: 'Optional[str]' = None, tool_access_mode: 'Optional[str]' = None, private: 'bool' = False, privacy_owner_user_id: 'Optional[str]' = None, privacy_span_id: 'Optional[str]' = None, client_send_started_at: 'Optional[str]' = None, metadata: 'Optional[Dict[str, Any]]' = None, session_scope: 'Optional[Dict[str, Any]]' = None)

Bases: object

__init__(id: str, account_id: str, session_id: str, procedure_id: str, content: str, role: str, message_type: str, human_interaction: str, response_target: str, response_status: str, created_at: str, selected_model: str | None = None, tool_access_mode: str | None = None, private: bool = False, privacy_owner_user_id: str | None = None, privacy_span_id: str | None = None, client_send_started_at: str | None = None, metadata: Dict[str, Any] | None = None, session_scope: Dict[str, Any] | None = None) → None

account_id: str

client_send_started_at: str | None = None

content: str

created_at: str

human_interaction: str

id: str

message_type: str

metadata: Dict[str, Any] | None = None

privacy_owner_user_id: str | None = None

privacy_span_id: str | None = None

private: bool = False

procedure_id: str

response_status: str

response_target: str

role: str

selected_model: str | None = None

session_id: str

session_scope: Dict[str, Any] | None = None

tool_access_mode: str | None = None

primus.console.chat_runtime.build_response_owner(target: str, *, request_id: str | None = None) → str

primus.console.chat_runtime.claim_message(client: PrimusDashboardClient, message: ConsoleMessage, *, expected_target: str, owner: str, latency_trace: Dict[str, Any] | None = None) → bool

primus.console.chat_runtime.console_async_tasks_available() → bool

Whether this Console worker can dispatch work to a durable async consumer.

A missing flag preserves the production behavior. Sandboxes without the TaskDispatcher/Celery stack set it to false so the agent can offer a synchronous alternative instead of creating a task that can never run.

primus.console.chat_runtime.fetch_assistant_chat_messages_between(client: PrimusDashboardClient, *, session_id: str, start_created_at: str, end_created_at: str, limit: int = 3) → List[str]

primus.console.chat_runtime.fetch_chat_session(client: PrimusDashboardClient, session_id: str) → Dict[str, Any] | None

primus.console.chat_runtime.fetch_first_assistant_chunk_details(client: PrimusDashboardClient, session_id: str, *, after_iso: str | None, max_items: int = 300) → Dict[str, Any] | None

primus.console.chat_runtime.fetch_first_assistant_chunk_timestamp(client: PrimusDashboardClient, session_id: str, *, after_iso: str | None, max_items: int = 300) → str | None

primus.console.chat_runtime.fetch_latest_report_task(client: PrimusDashboardClient, session_id: str) → Dict[str, str] | None

Read the durable report task captured on the most recent tool message.

primus.console.chat_runtime.fetch_latest_score_edit_audit(client: PrimusDashboardClient, session_id: str, *, max_items: int = 200) → Dict[str, Any] | None

Find the most recent persisted Console score-edit audit for this session.

This deliberately reads message metadata rather than chat prose: later model responses can be wrong, while the audit is written by the score-edit path.

primus.console.chat_runtime.fetch_message(client: PrimusDashboardClient, message_id: str) → ConsoleMessage | None

primus.console.chat_runtime.fetch_recent_user_chat_turns(client: PrimusDashboardClient, session_id: str, *, max_turns: int = 3) → List[Dict[str, str]]

primus.console.chat_runtime.fetch_session_history(client: PrimusDashboardClient, session_id: str, *, limit: int = 6, include_private_for_owner_user_id: str | None = None) → List[Dict[str, str]]

primus.console.chat_runtime.mark_message_completed(client: PrimusDashboardClient, message_id: str, *, created_at: str | None, completed_at: str | None = None, metadata: Dict[str, Any] | None = None) → None

primus.console.chat_runtime.mark_message_failed(client: PrimusDashboardClient, message_id: str, error: Exception, *, created_at: str | None, completed_at: str | None = None, metadata: Dict[str, Any] | None = None) → None

primus.console.chat_runtime.maybe_auto_title_session(client: PrimusDashboardClient, *, message: ConsoleMessage) → None

primus.console.chat_runtime.normalize_response_target(value: str | None) → str

primus.console.chat_runtime.parse_chat_message(raw: Dict[str, Any]) → ConsoleMessage | None

primus.console.chat_runtime.process_console_message(client: PrimusDashboardClient, raw_message: Dict[str, Any], *, expected_target: str, owner: str, poll_context: Dict[str, Any] | None = None) → bool

primus.console.chat_runtime.process_pending_local_messages(client: PrimusDashboardClient, *, response_target: str, owner: str, limit: int = 5) → int

primus.console.chat_runtime.resolve_console_tool_access_mode(client: PrimusDashboardClient, message: ConsoleMessage, *, session_scope: Dict[str, Any] | None = None) → str

primus.console.chat_runtime.run_console_chat_response(client: PrimusDashboardClient, message: ConsoleMessage, *, owner: str, latency_trace: Dict[str, Any] | None = None) → Dict[str, Any]

async primus.console.chat_runtime.run_console_chat_response_async(client: PrimusDashboardClient, message: ConsoleMessage, *, owner: str, latency_trace: Dict[str, Any] | None = None) → Dict[str, Any]

primus.console.chat_runtime.should_handle_message(message: ConsoleMessage, expected_target: str) → bool

primus.console.chat_runtime.update_chat_session_title(client: PrimusDashboardClient, *, session_id: str, title: str, turn: int, trigger_message_id: str | None, existing_metadata: Dict[str, Any] | None) → None

primus.console.chat_runtime.utc_now() → str

primus.console.chat_runtime.warm_console_runtime(client: PrimusDashboardClient) → None