Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 4 additions & 0 deletions reai_toolkit/app/app.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@
)
from reai_toolkit.app.services.analysis_sync.analysis_sync import AnalysisSyncService
from reai_toolkit.app.services.auth.auth_service import AuthService
from reai_toolkit.app.services.chat.chat_service import ChatService
from reai_toolkit.app.services.existing_analyses.existing_analyses_service import (
ExistingAnalysesService,
)
Expand Down Expand Up @@ -64,4 +65,7 @@ def __init__(self, ida_version: str = "UNKNOWN", plugin_version: str = "UNKNOWN"
self.similarity_service = SimilarityService(
netstore_service=self.netstore_service, sdk_config=sdk_config
)
self.chat_service = ChatService(
netstore_service=self.netstore_service, sdk_config=sdk_config
)

93 changes: 93 additions & 0 deletions reai_toolkit/app/components/tabs/chat_render.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,93 @@
"""Pure presentation helpers for the Agent Chat panel.

Turns a :class:`ChatState` into a markdown transcript. No Qt / IDA imports, so it
is unit-testable without a GUI (Qt only imports in the GUI version of IDA).
"""

from __future__ import annotations

from typing import Optional

from reai_toolkit.app.services.chat.schema import ChatState, ToolConfirmation


def title_case(name: str) -> str:
pretty = (name or "").replace("_", " ").strip()
return pretty.title() if pretty else (name or "tool")


def tool_marker(is_error: bool, status: str) -> str:
if is_error:
return "✗"
return "✓" if status == "finished" else "…"


JUMP_SCHEME = "ida://jump/"


def jump_href(ea: int) -> str:
return f"{JUMP_SCHEME}{ea}"


def parse_jump_href(url: str) -> Optional[int]:
if not url.startswith(JUMP_SCHEME):
return None
try:
return int(url[len(JUMP_SCHEME):])
except (ValueError, TypeError):
return None


def _function_links(functions) -> str:
parts = [f"[{f.name}]({jump_href(f.ea)})" for f in functions if f.name]
return "↪ " + " · ".join(parts) if parts else ""


def render_transcript_markdown(state: ChatState) -> str:
"""Build a single markdown document from the chat items."""
parts: list[str] = []
for item in state.items:
kind = item.kind
if kind == "user-message":
parts.append(f"**You:** {item.content}")
elif kind == "assistant-message":
text = item.content or ""
if item.is_streaming:
text = f"{text} ▍"
parts.append(text if text.strip() else "_…_")
elif kind == "tool-call":
parts.append(f"`{tool_marker(item.is_error, item.status)} {title_case(item.name)}`")
if item.functions:
links = _function_links(item.functions)
if links:
parts.append(links)
elif kind == "step":
if item.status == "running":
parts.append(f"_{item.step_name}…_")
elif kind == "tool-confirmation":
tool = title_case(item.tool_name)
if item.status == "pending":
parts.append(f"> ⚠ **Approval needed** — `{tool}`")
elif item.status == "approved":
parts.append(f"> ✓ Approved — `{tool}`")
else:
parts.append(f"> ✗ Rejected — `{tool}`")
elif kind == "context-compacted":
parts.append("_— context compacted —_")

if state.run_status == "running":
last = state.items[-1] if state.items else None
if last is None or last.kind == "user-message":
parts.append("_Thinking…_")

if state.run_status == "error" and state.run_error is not None:
parts.append(f"> ⚠ **Error:** {state.run_error.message}")

return "\n\n".join(parts)


def find_pending_confirmation(state: ChatState) -> Optional[ToolConfirmation]:
for item in reversed(state.items):
if isinstance(item, ToolConfirmation) and item.status == "pending":
return item
return None
Loading
Loading