Log user message to console and UI. Used by message API and queue processing.
(
context: "AgentContext",
message: str,
attachment_paths: list[str],
message_id: str | None = None,
source: str = "",
)
| 112 | |
| 113 | |
| 114 | def log_user_message( |
| 115 | context: "AgentContext", |
| 116 | message: str, |
| 117 | attachment_paths: list[str], |
| 118 | message_id: str | None = None, |
| 119 | source: str = "", |
| 120 | ): |
| 121 | """Log user message to console and UI. Used by message API and queue processing.""" |
| 122 | # Prepare attachment filenames for logging |
| 123 | attachment_filenames = ( |
| 124 | [os.path.basename(path) for path in attachment_paths] |
| 125 | if attachment_paths |
| 126 | else [] |
| 127 | ) |
| 128 | |
| 129 | # Print to console |
| 130 | label = f"User message{source}:" |
| 131 | PrintStyle( |
| 132 | background_color="#6C3483", font_color="white", bold=True, padding=True |
| 133 | ).print(label) |
| 134 | PrintStyle(font_color="white", padding=False).print(f"> {message}") |
| 135 | if attachment_filenames: |
| 136 | PrintStyle(font_color="white", padding=False).print("Attachments:") |
| 137 | for filename in attachment_filenames: |
| 138 | PrintStyle(font_color="white", padding=False).print(f"- {filename}") |
| 139 | |
| 140 | # Log to UI |
| 141 | context.log.log( |
| 142 | type="user", |
| 143 | heading="", |
| 144 | content=message, |
| 145 | kvps={"attachments": attachment_filenames}, |
| 146 | id=message_id, |
| 147 | ) |
| 148 | |
| 149 | |
| 150 | def send_message(context: "AgentContext", item: dict, source: str = " (from queue)"): |
no test coverage detected