Handle /context command - show token usage breakdown. Args: args: Command arguments context: Command context Returns: LocalCommandResult with Markdown table of context usage
(args: str, context: CommandContext)
| 328 | |
| 329 | |
| 330 | def context_command_call(args: str, context: CommandContext) -> LocalCommandResult: |
| 331 | """ |
| 332 | Handle /context command - show token usage breakdown. |
| 333 | |
| 334 | Args: |
| 335 | args: Command arguments |
| 336 | context: Command context |
| 337 | |
| 338 | Returns: |
| 339 | LocalCommandResult with Markdown table of context usage |
| 340 | """ |
| 341 | try: |
| 342 | # Get conversation messages in API format |
| 343 | conversation_api: list[dict[str, Any]] = [] |
| 344 | if hasattr(context.conversation, "get_messages"): |
| 345 | conversation_api = context.conversation.get_messages() |
| 346 | elif hasattr(context.conversation, "messages"): |
| 347 | # Fall back for simple mock conversations |
| 348 | for msg in context.conversation.messages: |
| 349 | role = getattr(msg, 'role', 'unknown') |
| 350 | content = getattr(msg, 'content', '') |
| 351 | conversation_api.append({"role": role, "content": content}) |
| 352 | |
| 353 | # Get system prompt from config |
| 354 | system_prompt = context.config.get("system_prompt", "") |
| 355 | |
| 356 | # Get tool schemas from config |
| 357 | tool_schemas = context.config.get("tool_schemas", []) |
| 358 | |
| 359 | # Get MCP tools info from config |
| 360 | mcp_tools = context.config.get("mcp_tools", []) |
| 361 | |
| 362 | # Get custom agents info from config |
| 363 | custom_agents = context.config.get("custom_agents", []) |
| 364 | |
| 365 | # Get CLAUDE.md content |
| 366 | claude_md_content = "" |
| 367 | try: |
| 368 | import asyncio |
| 369 | from ..context_system.claude_md import get_claude_mds, get_memory_files |
| 370 | |
| 371 | async def _load(): |
| 372 | files = await get_memory_files(cwd=str(context.cwd or context.workspace_root)) |
| 373 | return get_claude_mds(files) |
| 374 | |
| 375 | try: |
| 376 | loop = asyncio.get_running_loop() |
| 377 | except RuntimeError: |
| 378 | loop = None |
| 379 | if loop and loop.is_running(): |
| 380 | import concurrent.futures |
| 381 | with concurrent.futures.ThreadPoolExecutor(max_workers=1) as pool: |
| 382 | claude_md_content = pool.submit(asyncio.run, _load()).result(timeout=10) |
| 383 | else: |
| 384 | claude_md_content = asyncio.run(_load()) |
| 385 | except Exception: |
| 386 | pass |
| 387 |
no test coverage detected