intellicrack.core.orchestrator

Main AI agent orchestrator for Intellicrack.

This module provides the central orchestration layer that coordinates between the user, LLM providers, and tool bridges to execute reverse engineering workflows.

OPERATOR_CANCELLED_ERROR: Final[str] = 'Cancelled by the operator': The error a tool call the operator cancelled ends with.

class OrchestratorConfig[source]

Bases: object

Configuration for the orchestrator.

Variables:

confirmation_level (ConfirmationLevel) – When to ask for user confirmation.
max_iterations (int) – Maximum tool call iterations per request.
timeout_seconds (int) – Timeout for LLM requests.
temperature (float) – LLM temperature setting.
max_tokens (int) – Maximum tokens in LLM response.
stream_responses (bool) – Whether to stream LLM responses.
stream_mode (Literal['auto', 'always', 'never']) – Streaming mode (“auto”, “always”, “never”).
tool_choice (ToolChoice | None) – How the model should select tools.
thinking (ThinkingConfig | None) – Extended thinking configuration.
cache (CacheConfig | None) – Prompt caching configuration.
context_window_override (int | None) – Optional explicit context window size (in tokens) used when the provider cannot report one for the active model. When None the provider is required to return a context window; otherwise trimming is skipped.
enable_dynamic_loading (bool) – When True (the default), the agent loop advertises only the always-on core tool set plus the tools.search meta-tool, growing the active set on demand as the model discovers more via search. When False, every tool function in the registry is advertised on every iteration (the legacy behaviour), which is useful for tests that must exercise both modes.
core_tools (frozenset[str]) – Canonical dotted tool-function names that are always advertised, independent of what the model has discovered via tools.search. Empty by default: with no core set, the model must search before it can call anything but the meta-tool.
search_result_limit (int) – Default maximum number of matches tools.search returns per call when the caller does not specify its own limit argument.

confirmation_level: ConfirmationLevel = 'destructive'

max_iterations: int = 20

timeout_seconds: int = 120

temperature: float = 0.7

max_tokens: int = 4096

stream_responses: bool = True

stream_mode: Literal['auto', 'always', 'never'] = 'auto'

tool_choice: ToolChoice | None = None

thinking: ThinkingConfig | None = None

cache: CacheConfig | None = None

context_window_override: int | None = None

enable_dynamic_loading: bool = True

core_tools: frozenset[str]

search_result_limit: int = 10

__init__(confirmation_level=ConfirmationLevel.DESTRUCTIVE, max_iterations=20, timeout_seconds=120, temperature=0.7, max_tokens=4096, stream_responses=True, stream_mode='auto', tool_choice=None, thinking=None, cache=None, context_window_override=None, enable_dynamic_loading=True, core_tools=<factory>, search_result_limit=10)

Parameters:

confirmation_level (ConfirmationLevel)
max_iterations (int)
timeout_seconds (int)
temperature (float)
max_tokens (int)
stream_responses (bool)
stream_mode (Literal['auto', 'always', 'never'])
tool_choice (ToolChoice | None)
thinking (ThinkingConfig | None)
cache (CacheConfig | None)
context_window_override (int | None)
enable_dynamic_loading (bool)
core_tools (frozenset[str])
search_result_limit (int)

Return type:

None

class PendingConfirmation[source]

Bases: object

A tool call waiting for user confirmation.

Hashing falls back to id()-based identity (eq=False) so instances can live inside the orchestrator’s pending-confirmation set even though ToolCall is not hashable. Identity semantics are correct here because each pending confirmation is a unique instance.

Variables:

call (ToolCall) – The tool call awaiting confirmation.
future (Future[bool]) – Future that resolves to True when the user approves the call, False when they decline, and is cancelled (raising asyncio.CancelledError to the awaiter) when the orchestrator shuts down or a cancellation is signalled while a confirmation is pending.

call: ToolCall

future: Future[bool]

__init__(call, future)

Parameters:

call (ToolCall)
future (Future[bool])

Return type:

None

DestructiveClassification

Classification of a tool call’s effect on external state.

destructive operations modify external state (memory, files, processes, sandboxes) and require confirmation when the orchestrator is configured for ConfirmationLevel.DESTRUCTIVE. read_only operations only inspect state and never need confirmation. unknown indicates the bridge is not recognised; the orchestrator treats unknown operations as destructive to fail safe.

alias of Literal[‘destructive’, ‘read_only’, ‘unknown’]

BRIDGE_DESTRUCTIVE_METHODS: dict[ToolName, frozenset[str]] = {ToolName.CUTTER: frozenset({'add_comment', 'add_flag', 'add_zignature', 'analyze', 'execute_command', 'load_binary', 'open_project', 'patch_bytes', 'rename_function', 'save_project', 'set_function_signature'}), ToolName.FRIDA: frozenset({'allocate_memory', 'allocate_string', 'attach', 'attach_by_name', 'call_function', 'call_system_function', 'cancel', 'cloak_add_range', 'cloak_add_thread', 'cloak_remove_range', 'cloak_remove_thread', 'connect_device', 'create_cancellable', 'create_cmodule', 'detach', 'disable_child_gating', 'disable_crash_reporting', 'enable_child_gating', 'enable_crash_reporting', 'eternalize_script', 'execute_persistent_script', 'execute_script', 'file_write_target', 'flush_interceptor', 'hook_function', 'inject_library_blob', 'inject_library_file', 'intercept_return', 'java_deoptimize', 'java_hook_method', 'kernel_alloc', 'kernel_protect', 'kernel_write', 'load_module', 'monitor_path', 'objc_hook_method', 'patch_code', 'post_message', 'protect_memory', 'remove_hook', 'replace_function', 'resume', 'resume_child', 'revert_hook', 'rpc_call', 'set_exception_handler', 'socket_connect', 'socket_listen', 'spawn', 'sqlite_exec', 'sqlite_open', 'stalker_add_call_probe', 'stalker_follow', 'stalker_remove_call_probe', 'stalker_unfollow', 'stop_monitor', 'unload_all_scripts', 'unload_script', 'write_code', 'write_memory'}), ToolName.GHIDRA: frozenset({'add_bookmark', 'add_comment', 'add_external_function', 'add_external_reference', 'add_label', 'add_reference', 'add_thunk', 'analyze', 'apply_structure_at', 'configure_analysis', 'create_bookmark', 'create_data', 'create_data_type', 'create_equate', 'create_function', 'create_memory_block', 'create_namespace', 'create_overlay_space', 'define_structure', 'delete_function', 'delete_reference', 'edit_function_signature', 'execute_script', 'execute_script_with_params', 'import_debug_info', 'load_binary', 'redo', 'remove_bookmark', 'remove_external_reference', 'remove_label', 'remove_thunk', 'rename_function', 'set_color', 'set_data_type', 'set_decompiler_options', 'set_function_variable_type', 'set_label', 'set_program_metadata', 'start_headless', 'undo', 'write_bytes'}), ToolName.HEX_EDITOR: frozenset({'add_bookmark', 'add_highlight_rule', 'apply_arithmetic_to_selection', 'apply_pipeline', 'apply_transform', 'auto_detect_va_mappings', 'close_file', 'copy_block', 'delete_bytes', 'export_annotated_html', 'export_annotated_pdf', 'export_patches', 'export_patches_bps', 'export_patches_ups', 'fill_block', 'generate_structure_bookmarks', 'goto_offset', 'import_patches', 'import_patches_bps', 'import_patches_ups', 'insert_bytes', 'move_block', 'open_file', 'open_process_memory', 'redo', 'register_template', 'remove_bookmark', 'remove_highlight_rule', 'remove_template', 'remove_va_mapping', 'repair_pe_checksum', 'replace_bytes', 'run_python_script', 'save', 'save_as', 'save_to_sandbox', 'select_range', 'set_alignment_grid', 'set_bit', 'set_chunk_size', 'set_color_mode', 'set_display_mode', 'set_memory_budget', 'set_va_base', 'snap_to_alignment', 'swap_blocks', 'test_in_sandbox', 'toggle_bit', 'undo', 'write_bytes'}), ToolName.PROCESS: frozenset({'adjust_token_privilege', 'allocate', 'close', 'create_section', 'device_close', 'device_ioctl', 'device_open', 'free', 'inject_dll', 'map_section', 'open', 'pipe_close', 'pipe_connect', 'pipe_read', 'pipe_write', 'protect', 'resume', 'set_thread_context', 'suspend', 'terminate', 'unmap_section', 'write_memory'}), ToolName.SANDBOX: frozenset({'anti_evasion', 'cont', 'copy_from', 'copy_to', 'create', 'destroy', 'execute', 'extract_dropped_files', 'memory_dump', 'pcap_start', 'pcap_stop', 'run_binary', 'screenshot', 'set_vnc_password', 'snapshot_create', 'snapshot_delete', 'snapshot_restore', 'stop_pcap'}), ToolName.TOOLS: frozenset({}), ToolName.X64DBG: frozenset({'add_watch', 'adjust_privilege', 'allocate_memory', 'animate_start', 'animate_stop', 'assemble_at', 'attach', 'break_on_tls_callbacks', 'clear_database', 'close_handle', 'configure_breakpoint', 'detach', 'disable_breakpoint', 'dump_memory_to_file', 'enable_breakpoint', 'execute_til_return', 'export_patches', 'free_memory', 'goto_address', 'load', 'load_database', 'nop_range', 'patch_anti_debug', 'patch_instruction', 'pause', 'plugin_load', 'plugin_unload', 'reconstruct_imports', 'remove_breakpoint', 'remove_watch', 'remove_watchpoint', 'restore_patch', 'resume_thread', 'run', 'run_command', 'run_to', 'save_database', 'script_abort', 'script_cmd', 'script_load', 'script_run', 'set_breakpoint', 'set_breakpoint_on_api', 'set_comment', 'set_dll_breakpoint', 'set_exception_config', 'set_ip', 'set_label', 'set_logging_breakpoint', 'set_register', 'set_thread_name', 'set_watchpoint', 'skip_instruction', 'spawn', 'step_count', 'step_into', 'step_out', 'step_over', 'stop', 'suspend_thread', 'switch_thread', 'trace_into', 'trace_over', 'trace_start', 'trace_stop', 'write_memory', 'yara_scan'})}

Per-bridge whitelist of method names that mutate external state.

Each entry maps a ToolName to the exact method-name leaves (the part after the "<tool>." prefix) that the orchestrator must classify as destructive. Method names not present in the relevant set are read-only. Bridges absent from this map default to unknown classification, which the orchestrator treats as destructive so that newly added bridges fail safe until their methods are catalogued here.

class McpClassifier[source]

Bases: Protocol

Answers whether one canonical MCP tool call is read-only.

Implemented by McpToolSource, which only answers True for a tool on a server the operator has marked trusted.

owns_namespace(namespace)[source]

Report whether a tool namespace belongs to an MCP server.

Parameters:: namespace (str) – The namespace half of a canonical tool name.
Returns:: True when this source owns the namespace.
Return type:: bool

is_read_only(canonical_name)[source]

Report whether a canonical MCP tool call mutates nothing.

Parameters:: canonical_name (str) – mcp-<serverId>.<toolName>.
Returns:: True only when the call is known to be read-only.
Return type:: bool

catalog_lines()[source]

Render this source’s prompt section.

The source renders its own section because it is the thing that knows which servers exist, how healthy they are, and which parts of the text came from a server and therefore have to be fenced before they reach the model. The orchestrator only splices the result in.

Returns:: Prompt lines, with every server-supplied fragment already sanitized and fenced.
Return type:: list[str]

__init__(*args, **kwargs)

set_mcp_classifier(classifier)[source]

Install the classifier consulted for MCP tool calls.

Parameters:: classifier (McpClassifier | None) – The classifier to install, or None to remove the current one and return every MCP call to the destructive default.
Return type:: None

classify_tool_call(call)[source]

Classify a tool call as destructive, read_only or unknown.

Performs an exact lookup against BRIDGE_DESTRUCTIVE_METHODS. The bridge name is resolved from ToolCall.tool_name, falling back to the prefix of a "tool.method" style function_name when the field is empty. Method names are looked up by their leaf (the segment after the ".").

A namespace claimed by an installed external tool source – an MCP server – is answered by that source instead of falling through the ToolName lookup to unknown. It still answers destructive for everything except a trusted server’s explicitly read-only tool, so the deny-by-default posture is unchanged and simply becomes answerable.

Parameters:: call (ToolCall) – The ToolCall to classify.
Returns:: "destructive" when the bridge is known and the method is in its destructive set; "read_only" when the bridge is known and the method is not in its destructive set; "unknown" when the bridge name cannot be resolved to a registered ToolName.
Return type:: DestructiveClassification

class OrchestratorStats[source]

Bases: object

Statistics for orchestrator operations.

Variables:

total_requests (int) – Total user requests processed.
total_tool_calls (int) – Total tool calls executed.
successful_tool_calls (int) – Successful tool call count.
failed_tool_calls (int) – Failed tool call count.
total_tokens_used (int) – Approximate tokens used (heuristic estimate).
provider_prompt_tokens (int) – Real prompt-token totals reported by providers via LLMProviderBase.get_pending_usage().
provider_completion_tokens (int) – Real completion-token totals reported by providers via LLMProviderBase.get_pending_usage().
provider_total_tokens (int) – Real combined-token totals reported by providers via LLMProviderBase.get_pending_usage().
thinking_blocks_collected (int) – Number of extended-thinking blocks captured via LLMProviderBase.get_pending_thinking().
average_response_time_ms (float) – Average response time.

total_requests: int = 0

total_tool_calls: int = 0

successful_tool_calls: int = 0

failed_tool_calls: int = 0

total_tokens_used: int = 0

provider_prompt_tokens: int = 0

provider_completion_tokens: int = 0

provider_total_tokens: int = 0

thinking_blocks_collected: int = 0

average_response_time_ms: float = 0.0

record_response_time(time_ms)[source]

Record a response time and update rolling average.

Maintains a bounded window of the last 1000 response times to prevent unbounded memory growth.

Parameters:: time_ms (float) – Response time in milliseconds.
Return type:: None

to_dict()[source]

Convert statistics to dictionary for reporting.

Returns:: Dictionary containing all statistics.
Return type:: dict[str, Any]

__init__(total_requests=0, total_tool_calls=0, successful_tool_calls=0, failed_tool_calls=0, total_tokens_used=0, provider_prompt_tokens=0, provider_completion_tokens=0, provider_total_tokens=0, thinking_blocks_collected=0, average_response_time_ms=0.0, _response_times=<factory>)

Parameters:

total_requests (int)
total_tool_calls (int)
successful_tool_calls (int)
failed_tool_calls (int)
total_tokens_used (int)
provider_prompt_tokens (int)
provider_completion_tokens (int)
provider_total_tokens (int)
thinking_blocks_collected (int)
average_response_time_ms (float)
_response_times (deque[float])

Return type:

None

class Orchestrator[source]

Bases: object

Main AI agent orchestrator.

Manages the conversation loop between the user, LLM, and tools. Coordinates tool execution and handles confirmations.

Variables:: DESTRUCTIVE_PATTERNS (tuple[str, ...]) – Legacy substring tuple kept for callers that iterate the public attribute. The orchestrator itself classifies tool calls via classify_tool_call(), which performs exact method-name lookup against BRIDGE_DESTRUCTIVE_METHODS instead of substring matching.

DESTRUCTIVE_PATTERNS: tuple[str, ...] = ('write', 'patch', 'modify', 'delete', 'remove', 'set_', 'assemble', 'inject', 'intercept_return', 'hook', 'replace', 'overwrite')

__init__(provider_registry, tool_registry, session_manager, config=None)[source]

Initialize the orchestrator with registries and session manager.

Parameters:

provider_registry (ProviderRegistry) – Registry of LLM providers used for routing chat requests.
tool_registry (ToolRegistry) – Registry of tool bridges available for execution.
session_manager (SessionManager) – Session state manager that persists conversation state.
config (OrchestratorConfig | None) – Optional configuration override; defaults to OrchestratorConfig().

Return type:

None

tool_calls: The tool calls running now; the operator can cancel one and see each one’s progress.

property tool_registry: ToolRegistry

The tool registry backing this orchestrator.

Returns:: The registry of initialized tool bridges.
Return type:: ToolRegistry

property mcp_source: McpClassifier | None

The installed external tool source, when one is wired up.

Returns:: The source, or None.
Return type:: McpClassifier | None

set_mcp_tool_source(source)[source]

Wire up the Model Context Protocol tool source.

Installed at runtime rather than taken in the constructor so intellicrack.core.orchestrator never imports intellicrack.mcp and the two layers stay independent. Without one, an MCP namespace resolves to unknown and is confirmed as destructive, which is the correct default.

Parameters:: source (McpClassifier | None) – The source to install, or None to remove it.
Return type:: None

property state: Literal['idle', 'processing', 'waiting_confirmation', 'cancelled']

The current orchestrator state.

Returns:: Current state.
Return type:: OrchestratorState

property current_session: Session | None

The current session.

Returns:: Current session or None.
Return type:: Session | None

property stats: OrchestratorStats

The orchestrator statistics.

Returns:: Statistics instance.
Return type:: OrchestratorStats

property provider_registry: ProviderRegistry

The provider registry.

Returns:: The provider registry instance.
Return type:: ProviderRegistry

property pending_confirmation: PendingConfirmation | None

The most-recently registered pending confirmation, if any.

Returns:: The latest entry registered via _request_confirmation(), or None when no confirmation is currently outstanding.
Return type:: PendingConfirmation | None

property pending_confirmations: frozenset[PendingConfirmation]

A snapshot of every outstanding confirmation.

Returns:: An immutable snapshot of the pending-confirmation set. Mutating the orchestrator after the call will not affect the returned snapshot.
Return type:: frozenset[PendingConfirmation]

property shutdown_called: bool

Whether shutdown() has been invoked at least once.

Returns:: True once shutdown() has run; False before.
Return type:: bool

property shutdown_complete: bool

Whether the orchestrator’s shutdown event has fired.

Returns:: True when shutdown() has marked the internal shutdown event; otherwise False.
Return type:: bool

static is_destructive_operation(call)[source]

Determine whether a tool call requires destructive-op confirmation.

Uses the explicit per-bridge classifier classify_tool_call(), which does exact method-name lookup against BRIDGE_DESTRUCTIVE_METHODS. destructive and unknown classifications both require confirmation - unknown bridges fail safe so newly added integrations cannot bypass confirmation by virtue of not being catalogued. read_only operations skip confirmation.

Parameters:: call (ToolCall) – The tool call to evaluate.
Returns:: True when the call is classified as destructive or unknown; False only when the call is explicitly classified as read-only.
Return type:: bool

async request_confirmation(call)[source]

Request user confirmation for a tool call via the public API.

Parameters:: call (ToolCall) – The tool call requiring confirmation.
Returns:: The confirmation outcome (True confirmed, False declined, cancelled, or no callback registered).
Return type:: bool

set_script_manager(manager)[source]

Set the script manager for recording tool execution results.

Parameters:: manager (ScriptManager) – The ScriptManager instance.
Return type:: None

tag_current_session(tag)[source]

Add a tag to the current session.

CLI-friendly companion to the tag-chips widget in the session manager dialog. Delegates to Session.add_tag(), which normalises whitespace and rejects empty tags.

Parameters:: tag (str) – Non-empty tag string to add.
Returns:: True if the tag was newly added, False if it was already present on the session.
Return type:: bool
Raises:: RuntimeError – If no session is currently active.

untag_current_session(tag)[source]

Remove a tag from the current session.

CLI-friendly companion to the tag-chips widget in the session manager dialog. Delegates to Session.remove_tag().

Parameters:: tag (str) – Tag string to remove. Leading/trailing whitespace is stripped to match the normalisation performed by Session.add_tag().
Returns:: True if the tag was removed, False if it was not present on the session.
Return type:: bool
Raises:: RuntimeError – If no session is currently active.

async start_session(provider, model, binary_path=None, name=None, description=None)[source]

Start a new session.

Parameters:

provider (str) – Instance id of the LLM provider to use. Case is normalized, so a display-cased id still resolves.
model (str) – Model ID to use.
binary_path (Path | None) – Optional binary to load.
name (str | None) – Optional human-readable session name recorded on the new Session.
description (str | None) – Optional free-form description persisted as Session.notes.

Returns:

New session instance.

Return type:

Session

Raises:

ValueError – If the provider id is malformed or the provider is not available.

async load_session(session_id)[source]

Load an existing session and make it the current session.

Delegates to SessionManager.load() so the manager’s _current pointer is updated and the auto-save background task is started for this session, ensuring later edits persist without requiring an explicit save_session call.

Parameters:: session_id (str) – ID of session to load.
Returns:: Loaded session.
Return type:: Session
Raises:: ValueError – If session not found.

async process_user_input(text)[source]

Process user input and generate response.

This is the main agent loop:

Send pre-flight user message + history to the LLM with tool definitions; the user message is not persisted to the session until the loop completes successfully.
If LLM returns tool calls, execute them.
Send tool results back to LLM.
Repeat until LLM returns final text response.
On successful completion, append the user message and any assistant / tool messages emitted during the loop to the session and persist.
On cancellation or unhandled exception the session is not modified, leaving the persisted state consistent with the last successful turn.

Propagates asyncio.TimeoutError from _run_user_turn() when the agent loop does not complete within self._config.timeout_seconds; the in-memory turn is rolled back before the error surfaces here.

Parameters:

text (str) – User’s natural language input.

Raises:

RuntimeError – If no active session.
CancelledError – If the request is cancelled. The cancellation is re-raised after the in-memory turn rollback so callers can differentiate cancellation from completion.

Return type:

None

build_system_prompt()[source]

Render the current system prompt.

Public seam over _generate_system_prompt() for callers (UI, tests, embedded clients) that need to inspect what the orchestrator will send to the LLM without going through process_user_input.

Returns:

System prompt for the active session, or an empty string: when no session is active.

Return type:

str

static estimate_tokens(text, tokenizer=None)[source]

Count tokens in text using the model’s own tiktoken encoding.

Public entry point that delegates to _estimate_tokens(). The encoding comes from the model’s capability record rather than from the provider’s identity, because the two do not correlate once a provider can be any endpoint at all: an Anthropic-compatible gateway serving Llama is not cl100k. A model that states no tokenizer falls back to o200k_base, which overcounts against most real tokenizers rather than undercounting, avoiding the runaway prompt-size failures the original len // 4 heuristic produced on token-dense payloads (code, hex dumps, table output).

Parameters:

text (str) – Text to count tokens for.
tokenizer (str | None) – tiktoken encoding name from the model’s capability record, or None to use the default encoding.

Returns:

Token count for text.

Return type:

int

static trim_messages_to_context_window(messages, context_window, *, tokenizer=None, tool_overhead_tokens=0)[source]

Remove oldest non-system messages until within context budget.

Keeps 85% of the context window as the token budget to leave headroom for the response, then subtracts whatever the advertised tools will occupy in the same request. Token counting uses _message_tokens(), which measures tool calls and tool results as well as message content.

context_window=None is treated as a hard error rather than a silent passthrough so callers cannot accidentally send unbounded history to a provider that does not report a window. Use the per-provider helper _trim_messages_for_provider() from the agent loop, which always passes a resolved value.

Parameters:

messages (list[Message]) – List of messages to trim. Mutated in place.
context_window (int | None) – Maximum context window in tokens. None raises ToolError instead of skipping trimming.
tokenizer (str | None) – tiktoken encoding name from the model’s capability record, used for token counting.
tool_overhead_tokens (int) – Tokens the advertised tool definitions will occupy in the same request, subtracted from the budget.

Returns:

Trimmed list of messages.

Return type:

list[Message]

Raises:

ToolError – If context_window is None, or if the advertised tools leave no room for any conversation at all.

confirm_pending(*, confirmed)[source]

Confirm or decline the most recently registered pending operation.

Resolves the latest entry registered in _pending_confirmations. If the future was already cancelled (for example, by cancel() or shutdown()), this is a no-op rather than raising asyncio.InvalidStateError.

Parameters:: confirmed (bool) – True to confirm the operation, False to decline.
Return type:: None

async cancel()[source]

Cancel the current operation and marshal pending confirmations.

Sets the cancel event, requests provider-side cancellation, cancels any tool call still running, then cancels every outstanding confirmation future tracked in _pending_confirmations. future.cancel() propagates an asyncio.CancelledError to any awaiting _request_confirmation(), which translates it back into a False return so callers do not see leaked exceptions and do not hang waiting for user input that will never arrive.

Return type:: None

async add_binary(path, *, run_bridge_analysis=True)[source]

Add a binary to the current session.

Parameters:

path (Path) – Path to the binary.
run_bridge_analysis (bool) – Whether to run bridge analysis automatically.

Returns:

Binary information.

Return type:

BinaryInfo

Raises:

RuntimeError – If no active session.

async reanalyze_bridge_analysis(binary_name=None)[source]

Re-run bridge analysis on the active or specified binary.

Parameters:: binary_name (str | None) – Optional binary name; uses active binary if not specified.
Returns:: Refreshed results or None on failure.
Return type:: BridgeAnalysisSummary | None

set_bridge_analysis_callback(callback)[source]

Set callback for bridge analysis completion.

Parameters:: callback (Callable[[BridgeAnalysisSummary], None] | None) – Function to call with analysis results.
Return type:: None

async set_active_binary(index)[source]

Set the active binary by index.

Parameters:

index (int) – Index of binary to activate.

Raises:

RuntimeError – If no active session.
IndexError – If index out of range.

Return type:

None

async add_patch(patch)[source]

Add a patch to the current session.

Parameters:: patch (PatchInfo) – Patch information.
Raises:: RuntimeError – If no active session.
Return type:: None

set_message_callback(callback)[source]

Set callback for new messages.

Parameters:: callback (Callable[[Message], None]) – Function to call with each new message.
Return type:: None

set_tool_call_callback(callback)[source]

Set callback for tool calls.

Parameters:: callback (Callable[[ToolCall], None]) – Function to call when tool is called.
Return type:: None

set_tool_result_callback(callback)[source]

Set callback for tool results.

Parameters:: callback (Callable[[ToolResult], None]) – Function to call when tool returns result.
Return type:: None

set_stream_callback(callback)[source]

Set callback for streaming response chunks.

Parameters:: callback (Callable[[str], None]) – Function to call with each text chunk.
Return type:: None

set_confirmation_callback(callback)[source]

Set synchronous callback for confirmation requests.

Parameters:: callback (Callable[[ToolCall], bool]) – Function to call for confirmation, returns True to proceed.
Return type:: None

set_async_confirmation_callback(callback)[source]

Set async callback for confirmation requests.

Parameters:: callback (Callable[[ToolCall], Future[bool]]) – Function returning a Future that resolves to True/False.
Return type:: None

async get_tool_status()[source]

Get status of all tools.

Returns:: List of tool status dictionaries.
Return type:: list[dict[str, Any]]

get_available_tool_names()[source]

Get names of all available tools.

Returns:: List of available tool name strings.
Return type:: list[str]

get_current_bridge_analysis(binary_name)[source]

Get cached bridge analysis for a binary.

Parameters:: binary_name (str) – Name of the binary.
Returns:: Cached results if available, None otherwise.
Return type:: BridgeAnalysisSummary | None

get_typed_bridge(tool_name)[source]

Get a typed bridge instance by tool name.

Uses the ToolRegistry’s typed getters for safe bridge access.

Parameters:: tool_name (str) – Name of the tool bridge to retrieve.
Returns:: Typed bridge instance or None if not available.
Return type:: object | None

async initialize_tool(tool_name)[source]

Initialize a specific tool.

Parameters:: tool_name (str | ToolName) – Name of the tool to initialize.
Returns:: True if initialization succeeded.
Return type:: bool

async save_session()[source]

Save the current session.

Delegates to the session manager to persist the current session state.

Return type:: None

set_confirmation_level(level)[source]

Set the confirmation level for tool calls.

Parameters:: level (ConfirmationLevel) – The desired confirmation level.
Return type:: None

async get_system_status()[source]

Get comprehensive system status report.

Returns:: Dictionary containing session, metrics, and tool status.
Return type:: dict[str, Any]

configure_hooks(on_bridge_analysis=None, on_confirmation=None)[source]

Configure event hooks.

Parameters:

on_bridge_analysis (Callable[[BridgeAnalysisSummary], None] | None) – Callback for bridge analysis completion.
on_confirmation (Callable[[ToolCall], bool] | None) – Callback for confirmation requests.

Return type:

None

async refresh_session_state()[source]

Refresh cached bridge analysis for the active session.

Re-runs reanalyze_bridge_analysis() against the current session’s active binary so stale bridge results are regenerated. Does not reload the session from disk or refresh any other session state.

Return type:: None

async register_manual_patch(address, original_bytes, new_bytes, description)[source]

Register a manually applied patch.

Parameters:

address (int) – Patch address.
original_bytes (bytes) – Original bytes.
new_bytes (bytes) – New bytes.
description (str) – Description.

Return type:

None

resolve_confirmation(*, approved)[source]

Resolve any pending confirmation request.

Parameters:: approved (bool) – Whether to approve the request.
Return type:: None

async activate_binary_by_name(name)[source]

Activate a binary by name.

Parameters:

name (str) – Name of binary to activate.

Raises:

ValueError – If binary not found.
RuntimeError – If no active session.

Return type:

None

async shutdown()[source]

Shutdown the orchestrator and cleanup resources.

Each teardown step is guarded by except Exception so one failing stage cannot leave other resources dangling. BaseException subclasses (notably asyncio.CancelledError and KeyboardInterrupt) are intentionally not caught and propagate through the finally clause that clears final state. All caught exceptions are collected and, once every stage has had a chance to run, bundled into a single ExceptionGroup so callers learn about every teardown failure.

Before any other teardown work, all in-flight confirmation futures are marshalled via _marshal_pending_confirmations() so awaiters do not leak coroutines or hang waiting for input that will never arrive.

Raises:: ExceptionGroup – When one or more teardown steps raised non-cancellation exceptions, all collected failures are bundled into an ExceptionGroup.
Return type:: None

extract_imports(binary)[source]

Extract import metadata from a parsed lief binary.

Implements full coverage for the three formats Intellicrack ingests:

PE - walks every ImportEntry from every imported DLL.
ELF - enumerates every imported dynamic symbol (object and function), not just PLT/GOT relocations. This pulls in data symbols (__environ, stdout) and lazy-bound functions that the previous pltgot_relocations scan missed entirely on stripped or BIND_NOW-linked binaries.
Mach-O - walks imported_symbols so dyld-resolved imports (e.g. _printf) appear in the import list. The enclosing dylib is resolved through the symbol’s library attribute when ordinal-based two-level lookups expose it.

Parameters:: binary (object) – A parsed lief.PE.Binary, lief.ELF.Binary, or lief.MachO.Binary instance.
Returns:: Imported entries with the full (dll, function, ordinal, address) tuple populated as far as the format reveals.
Return type:: list[ImportInfo]

extract_exports(binary)[source]

Extract export metadata from a parsed lief binary.

Implements full coverage for the three formats Intellicrack ingests:

PE - walks every ExportEntry from the export directory.
ELF - walks every dynamic symbol marked as exported.
Mach-O - walks exported_symbols, which surfaces both classic __TEXT symbol-table exports and LC_DYLD_EXPORTS_TRIE / LC_DYLD_INFO trie-encoded exports that newer macOS dylibs publish.

Parameters:: binary (object) – A parsed lief.PE.Binary, lief.ELF.Binary, or lief.MachO.Binary instance.
Returns:: Exported entries with (name, ordinal, address) populated. Ordinals are 0 for ELF / Mach-O because those formats do not assign export ordinals.
Return type:: list[ExportInfo]