Spaces:
Sleeping
Sleeping
File size: 4,227 Bytes
2415446 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 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 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 | """CLI event handling for a single queued node (transcript + session + errors)."""
from collections.abc import Awaitable, Callable
from typing import Any
from loguru import logger
from free_claude_code.core.trace import trace_event
from .cli_event_constants import TRANSCRIPT_EVENT_TYPES, get_status_for_event
from .managed_protocols import ManagedClaudeSessionManagerProtocol
from .safe_diagnostics import text_len_hint
from .transcript import TranscriptBuffer
from .trees import NodeClaim
RecordSession = Callable[[str], Awaitable[None]]
CompleteClaim = Callable[[str | None], Awaitable[None]]
FailClaim = Callable[[str, str], Awaitable[None]]
async def handle_session_info_event(
event_data: dict[str, Any],
claim: NodeClaim,
captured_session_id: str | None,
temp_session_id: str | None,
*,
cli_manager: ManagedClaudeSessionManagerProtocol,
record_session: RecordSession,
) -> tuple[str | None, str | None]:
"""Handle session_info event; return updated (captured_session_id, temp_session_id)."""
if event_data.get("type") != "session_info":
return captured_session_id, temp_session_id
real_session_id = event_data.get("session_id")
if not real_session_id or not temp_session_id:
return captured_session_id, temp_session_id
registered = await cli_manager.register_real_session_id(
temp_session_id,
real_session_id,
)
if not registered:
raise RuntimeError("Managed Claude session registration failed.")
trace_event(
stage="claude_cli",
event="claude_cli.session.registered",
source="claude_cli",
node_id=claim.node.node_id,
temp_session_id=temp_session_id,
real_session_id=real_session_id,
tree_root_id=claim.identity.root_id,
)
await record_session(real_session_id)
return real_session_id, None
async def process_parsed_cli_event(
parsed: dict[str, Any],
transcript: TranscriptBuffer,
update_ui: Callable[..., Awaitable[None]],
last_status: str | None,
had_transcript_events: bool,
claim: NodeClaim,
captured_session_id: str | None,
*,
format_status: Callable[..., str],
complete_claim: CompleteClaim,
fail_claim: FailClaim,
log_messaging_error_details: bool = False,
) -> tuple[str | None, bool]:
"""Process a single parsed CLI event. Returns (last_status, had_transcript_events)."""
ptype = parsed.get("type") or ""
if ptype in TRANSCRIPT_EVENT_TYPES:
transcript.apply(parsed)
had_transcript_events = True
status = get_status_for_event(ptype, parsed, format_status)
if status is not None:
await update_ui(status)
last_status = status
elif ptype == "block_stop":
await update_ui(last_status, force=True)
elif ptype == "complete":
if parsed.get("status") != "success":
return last_status, had_transcript_events
if not had_transcript_events:
transcript.apply({"type": "text_chunk", "text": "Done."})
trace_event(
stage="claude_cli",
event="turn.completed",
source="cli_event",
node_id=claim.node.node_id,
claude_session_id=captured_session_id,
)
await update_ui(format_status("✅", "Complete"), force=True)
await complete_claim(captured_session_id)
elif ptype == "error":
error_msg = parsed.get("message", "Unknown error")
em = error_msg if isinstance(error_msg, str) else str(error_msg)
trace_event(
stage="claude_cli",
event="turn.failed",
source="cli_event",
node_id=claim.node.node_id,
claude_session_id=captured_session_id,
cli_error_message=em,
)
if log_messaging_error_details:
logger.error("HANDLER: Error event received: {}", error_msg)
else:
logger.error(
"HANDLER: Error event received: message_chars={}",
text_len_hint(em),
)
await update_ui(format_status("❌", "Error"), force=True)
await fail_claim(em, "Parent task failed")
return last_status, had_transcript_events
|