Function bodies 215 total

class BootstrapGraph:
    stages: tuple[str, ...]

    def as_markdown(self) -> str:
        lines = ['# Bootstrap Graph', '']
        lines.extend(f'- {stage}' for stage in self.stages)
        return '\n'.join(lines)

    def as_markdown(self) -> str:
        lines = ['# Bootstrap Graph', '']
        lines.extend(f'- {stage}' for stage in self.stages)
        return '\n'.join(lines)

def build_bootstrap_graph() -> BootstrapGraph:
    return BootstrapGraph(
        stages=(
            'top-level prefetch side effects',
            'warning handler and environment guards',
            'CLI parser and pre-action trust gate',
            'setup() + commands/agents parallel load',
            'deferred init after trust',
            'mode routing: local / remote / ssh / teleport / direct-connect / deep-link',
            'query engine submit loop',
        )
    )

class CommandGraph:
    builtins: tuple[PortingModule, ...]
    plugin_like: tuple[PortingModule, ...]
    skill_like: tuple[PortingModule, ...]

    def flattened(self) -> tuple[PortingModule, ...]:
        return self.builtins + self.plugin_like + self.skill_like

    def as_markdown(self) -> str:
        lines = [
            '# Command Graph',
            '',
            f'Builtins: {len(self.builtins)}',
            f'Plugin-like commands: {len(self.plugin_like)}',
            f'Skill-like commands: {len(self.skill_like)}',
        ]
        return '\n'.join(lines)

    def as_markdown(self) -> str:
        lines = [
            '# Command Graph',
            '',
            f'Builtins: {len(self.builtins)}',
            f'Plugin-like commands: {len(self.plugin_like)}',
            f'Skill-like commands: {len(self.skill_like)}',
        ]
        return '\n'.join(lines)

def build_command_graph() -> CommandGraph:
    commands = get_commands()
    builtins = tuple(module for module in commands if 'plugin' not in module.source_hint.lower() and 'skills' not in module.source_hint.lower())
    plugin_like = tuple(module for module in commands if 'plugin' in module.source_hint.lower())
    skill_like = tuple(module for module in commands if 'skills' in module.source_hint.lower())
    return CommandGraph(builtins=builtins, plugin_like=plugin_like, skill_like=skill_like)

class CommandExecution:
    name: str
    source_hint: str
    prompt: str
    handled: bool
    message: str

def load_command_snapshot() -> tuple[PortingModule, ...]:
    raw_entries = json.loads(SNAPSHOT_PATH.read_text())
    return tuple(
        PortingModule(
            name=entry['name'],
            responsibility=entry['responsibility'],
            source_hint=entry['source_hint'],
            status='mirrored',
        )
        for entry in raw_entries
    )

def get_command(name: str) -> PortingModule | None:
    needle = name.lower()
    for module in PORTED_COMMANDS:
        if module.name.lower() == needle:
            return module
    return None

def get_commands(cwd: str | None = None, include_plugin_commands: bool = True, include_skill_commands: bool = True) -> tuple[PortingModule, ...]:
    commands = list(PORTED_COMMANDS)
    if not include_plugin_commands:
        commands = [module for module in commands if 'plugin' not in module.source_hint.lower()]
    if not include_skill_commands:
        commands = [module for module in commands if 'skills' not in module.source_hint.lower()]
    return tuple(commands)

def find_commands(query: str, limit: int = 20) -> list[PortingModule]:
    needle = query.lower()
    matches = [module for module in PORTED_COMMANDS if needle in module.name.lower() or needle in module.source_hint.lower()]
    return matches[:limit]

def execute_command(name: str, prompt: str = '') -> CommandExecution:
    module = get_command(name)
    if module is None:
        return CommandExecution(name=name, source_hint='', prompt=prompt, handled=False, message=f'Unknown mirrored command: {name}')
    action = f"Mirrored command '{module.name}' from {module.source_hint} would handle prompt {prompt!r}."
    return CommandExecution(name=module.name, source_hint=module.source_hint, prompt=prompt, handled=True, message=action)

def render_command_index(limit: int = 20, query: str | None = None) -> str:
    modules = find_commands(query, limit) if query else list(PORTED_COMMANDS[:limit])
    lines = [f'Command entries: {len(PORTED_COMMANDS)}', '']
    if query:
        lines.append(f'Filtered by: {query}')
        lines.append('')
    lines.extend(f'- {module.name} — {module.source_hint}' for module in modules)
    return '\n'.join(lines)

class PortContext:
    source_root: Path
    tests_root: Path
    assets_root: Path
    archive_root: Path
    python_file_count: int
    test_file_count: int
    asset_file_count: int
    archive_available: bool

def build_port_context(base: Path | None = None) -> PortContext:
    root = base or Path(__file__).resolve().parent.parent
    source_root = root / 'src'
    tests_root = root / 'tests'
    assets_root = root / 'assets'
    archive_root = root / 'archive' / 'claude_code_ts_snapshot' / 'src'
    return PortContext(
        source_root=source_root,
        tests_root=tests_root,
        assets_root=assets_root,
        archive_root=archive_root,
        python_file_count=sum(1 for path in source_root.rglob('*.py') if path.is_file()),
        test_file_count=sum(1 for path in tests_root.rglob('*.py') if path.is_file()),
        asset_file_count=sum(1 for path in assets_root.rglob('*') if path.is_file()),
        archive_available=archive_root.exists(),
    )

def render_context(context: PortContext) -> str:
    return '\n'.join([
        f'Source root: {context.source_root}',
        f'Test root: {context.tests_root}',
        f'Assets root: {context.assets_root}',
        f'Archive root: {context.archive_root}',
        f'Python files: {context.python_file_count}',
        f'Test files: {context.test_file_count}',
        f'Assets: {context.asset_file_count}',
        f'Archive available: {context.archive_available}',
    ])

def apply_cost_hook(tracker: CostTracker, label: str, units: int) -> CostTracker:
    tracker.record(label, units)
    return tracker

class CostTracker:
    total_units: int = 0
    events: list[str] = field(default_factory=list)

    def record(self, label: str, units: int) -> None:
        self.total_units += units
        self.events.append(f'{label}:{units}')

    def record(self, label: str, units: int) -> None:
        self.total_units += units
        self.events.append(f'{label}:{units}')

class DeferredInitResult:
    trusted: bool
    plugin_init: bool
    skill_init: bool
    mcp_prefetch: bool
    session_hooks: bool

    def as_lines(self) -> tuple[str, ...]:
        return (
            f'- plugin_init={self.plugin_init}',
            f'- skill_init={self.skill_init}',
            f'- mcp_prefetch={self.mcp_prefetch}',
            f'- session_hooks={self.session_hooks}',
        )

    def as_lines(self) -> tuple[str, ...]:
        return (
            f'- plugin_init={self.plugin_init}',
            f'- skill_init={self.skill_init}',
            f'- mcp_prefetch={self.mcp_prefetch}',
            f'- session_hooks={self.session_hooks}',
        )

def run_deferred_init(trusted: bool) -> DeferredInitResult:
    enabled = bool(trusted)
    return DeferredInitResult(
        trusted=trusted,
        plugin_init=enabled,
        skill_init=enabled,
        mcp_prefetch=enabled,
        session_hooks=enabled,
    )

class DialogLauncher:
    name: str
    description: str

class DirectModeReport:
    mode: str
    target: str
    active: bool

    def as_text(self) -> str:
        return f'mode={self.mode}\ntarget={self.target}\nactive={self.active}'

class MirroredCommand:
    name: str
    source_hint: str

    def execute(self, prompt: str) -> str:
        return execute_command(self.name, prompt).message

class MirroredTool:
    name: str
    source_hint: str

    def execute(self, payload: str) -> str:
        return execute_tool(self.name, payload).message

class ExecutionRegistry:
    commands: tuple[MirroredCommand, ...]
    tools: tuple[MirroredTool, ...]

    def command(self, name: str) -> MirroredCommand | None:
        lowered = name.lower()
        for command in self.commands:
            if command.name.lower() == lowered:
                return command
        return None

    def tool(self, name: str) -> MirroredTool | None:
        lowered = name.lower()
        for tool in self.tools:
            if tool.name.lower() == lowered:
                return tool
        return None

    def command(self, name: str) -> MirroredCommand | None:
        lowered = name.lower()
        for command in self.commands:
            if command.name.lower() == lowered:
                return command
        return None

    def tool(self, name: str) -> MirroredTool | None:
        lowered = name.lower()
        for tool in self.tools:
            if tool.name.lower() == lowered:
                return tool
        return None

def build_execution_registry() -> ExecutionRegistry:
    return ExecutionRegistry(
        commands=tuple(MirroredCommand(module.name, module.source_hint) for module in PORTED_COMMANDS),
        tools=tuple(MirroredTool(module.name, module.source_hint) for module in PORTED_TOOLS),
    )

class HistoryEvent:
    title: str
    detail: str

class HistoryLog:
    events: list[HistoryEvent] = field(default_factory=list)

    def add(self, title: str, detail: str) -> None:
        self.events.append(HistoryEvent(title=title, detail=detail))

    def as_markdown(self) -> str:
        lines = ['# Session History', '']
        lines.extend(f'- {event.title}: {event.detail}' for event in self.events)
        return '\n'.join(lines)

    def as_markdown(self) -> str:
        lines = ['# Session History', '']
        lines.extend(f'- {event.title}: {event.detail}' for event in self.events)
        return '\n'.join(lines)

def build_hook_registry(settings_path: Path | None = None) -> HookRegistry:
    """Build a HookRegistry from a settings JSON file.

    Expected format in settings JSON:
    {
      "hooks": {
        "pre_tool_execution": [
          {"name": "audit_log", "command": "echo $HOOK_DATA >> /tmp/audit.log"}
        ],
        "post_tool_execution": [...],
        ...
      }
    }
    """
    registry = HookRegistry()

    if settings_path is None:
        candidates = [
            Path.home() / '.claude' / 'settings.json',
            Path.cwd() / '.claude' / 'settings.json',
        ]
        for candidate in candidates:
            if candidate.exists():
                settings_path = candidate
                break

    if settings_path is None or not settings_path.exists():
        return registry

    try:
        data = json.loads(settings_path.read_text())
    except (json.JSONDecodeError, OSError):
        return registry

    hooks_config = data.get('hooks', {})

    for event 

class HookRegistry:
    """Registry of lifecycle hooks. Handlers fire in registration order."""

    def __init__(self) -> None:
        self._hooks: dict[HookEvent, list[tuple[str, HookHandler]]] = {
            event: [] for event in HookEvent
        }

    def register(self, event: HookEvent, name: str, handler: HookHandler) -> None:
        """Register a hook handler for a lifecycle event."""
        self._hooks[event].append((name, handler))

    def fire(self, event: HookEvent, context: HookContext) -> HookContext:
        """Fire all handlers for an event. Stops if context.cancelled becomes True."""
        for name, handler in self._hooks[event]:
            try:
                context = handler(context)
            except Exception as e:
                # Hooks should not crash the harness
                context.data.setdefault('hook_errors', [])
                context.data['hook_errors'].append(f'{name}: {e}')
            if context.cancelled:
                break
      

    def __init__(self) -> None:
        self._hooks: dict[HookEvent, list[tuple[str, HookHandler]]] = {
            event: [] for event in HookEvent
        }

    def register(self, event: HookEvent, name: str, handler: HookHandler) -> None:
        """Register a hook handler for a lifecycle event."""
        self._hooks[event].append((name, handler))

    def fire(self, event: HookEvent, context: HookContext) -> HookContext:
        """Fire all handlers for an event. Stops if context.cancelled becomes True."""
        for name, handler in self._hooks[event]:
            try:
                context = handler(context)
            except Exception as e:
                # Hooks should not crash the harness
                context.data.setdefault('hook_errors', [])
                context.data['hook_errors'].append(f'{name}: {e}')
            if context.cancelled:
                break
        return context

class ShellHookHandler:
    """Hook handler that runs a shell command, passing context via env vars."""

    def __init__(self, command: str, timeout: int = 10) -> None:
        self.command = command
        self.timeout = timeout

    def __call__(self, context: HookContext) -> HookContext:
        env = os.environ.copy()
        env['HOOK_EVENT'] = context.event.value
        env['HOOK_DATA'] = json.dumps(context.data, default=str)

        try:
            result = subprocess.run(
                self.command,
                shell=True,
                capture_output=True,
                text=True,
                timeout=self.timeout,
                env=env,
            )
            if result.stdout.strip():
                context.data['hook_output'] = result.stdout.strip()
        except subprocess.TimeoutExpired:
            context.data.setdefault('hook_errors', [])
            context.data['hook_errors'].append(f'Shell hook timed out: {self.command}')
        except Excep

    def __init__(self, command: str, timeout: int = 10) -> None:
        self.command = command
        self.timeout = timeout

    def __call__(self, context: HookContext) -> HookContext:
        env = os.environ.copy()
        env['HOOK_EVENT'] = context.event.value
        env['HOOK_DATA'] = json.dumps(context.data, default=str)

        try:
            result = subprocess.run(
                self.command,
                shell=True,
                capture_output=True,
                text=True,
                timeout=self.timeout,
                env=env,
            )
            if result.stdout.strip():
                context.data['hook_output'] = result.stdout.strip()
        except subprocess.TimeoutExpired:
            context.data.setdefault('hook_errors', [])
            context.data['hook_errors'].append(f'Shell hook timed out: {self.command}')
        except Exception as e:
            context.data.setdefault('hook_errors', [])
            context.data['hook_errors'].append(f'Shell hook error: {e}')

        return context

class HookEvent(str, Enum):
    """Lifecycle events that hooks can subscribe to."""
    PRE_TOOL_EXECUTION = 'pre_tool_execution'
    POST_TOOL_EXECUTION = 'post_tool_execution'
    SESSION_START = 'session_start'
    SESSION_END = 'session_end'
    TURN_START = 'turn_start'
    TURN_END = 'turn_end'
    PRE_COMPACTION = 'pre_compaction'

class HookContext:
    """Mutable context passed through hook handlers.

    Hooks can modify `data` to transform tool inputs/outputs.
    Setting `cancelled=True` in a PreToolExecution hook blocks the tool.
    """
    event: HookEvent
    data: dict[str, object] = field(default_factory=dict)
    cancelled: bool = False
    cancel_reason: str = ''

    def cancel(self, reason: str = '') -> None:
        self.cancelled = True
        self.cancel_reason = reason

    def cancel(self, reason: str = '') -> None:
        self.cancelled = True
        self.cancel_reason = reason

def render_markdown_panel(text: str) -> str:
    border = '=' * 40
    return f"{border}\n{text}\n{border}"

class AgentTool(BaseTool):
    """Spawns a sub-conversation with a separate turn loop and tool set."""

    def __init__(
        self,
        llm_client,  # BaseLLMClient — not typed to avoid circular import
        tool_registry_factory: Callable[[], dict[str, BaseTool]],
        hook_registry=None,  # HookRegistry | None
        max_turns: int = 15,
        max_budget_tokens: int = 50_000,
    ) -> None:
        self._llm = llm_client
        self._tool_factory = tool_registry_factory
        self._hooks = hook_registry
        self._max_turns = max_turns
        self._max_budget = max_budget_tokens

    def definition(self) -> ToolDefinition:
        return ToolDefinition(
            name='Agent',
            description=(
                'Launch a sub-agent to handle a complex, multi-step task autonomously. '
                'The sub-agent gets access to Read, Write, Edit, Bash, Grep, Glob tools '
                'but cannot spawn further agents. Use for tasks that require multi

    def __init__(
        self,
        llm_client,  # BaseLLMClient — not typed to avoid circular import
        tool_registry_factory: Callable[[], dict[str, BaseTool]],
        hook_registry=None,  # HookRegistry | None
        max_turns: int = 15,
        max_budget_tokens: int = 50_000,
    ) -> None:
        self._llm = llm_client
        self._tool_factory = tool_registry_factory
        self._hooks = hook_registry
        self._max_turns = max_turns
        self._max_budget = max_budget_tokens

    def definition(self) -> ToolDefinition:
        return ToolDefinition(
            name='Agent',
            description=(
                'Launch a sub-agent to handle a complex, multi-step task autonomously. '
                'The sub-agent gets access to Read, Write, Edit, Bash, Grep, Glob tools '
                'but cannot spawn further agents. Use for tasks that require multiple '
                'tool calls or deep investigation.'
            ),
            input_schema={
                'type': 'object',
                'properties': {
                    'prompt': {
                        'type': 'string',
                        'description': 'The task for the sub-agent to perform',
                    },
                },
                'required': ['prompt'],
            },
        )

    def execute(self, input: dict[str, object]) -> str:
        # Import here to avoid circular dependency
        from ..turn_loop import TurnLoopRunner, TurnLoopConfig

        prompt = str(input.get('prompt', ''))
        if not prompt:
            return 'Error: prompt is required'

        # Build sub-agent tools: everything EXCEPT Agent (no recursion)
        all_tools = self._tool_factory()
        sub_tools = {name: tool for name, tool in all_tools.items() if name != 'Agent'}

        config = TurnLoopConfig(
            max_turns=self._max_turns,
            max_budget_tokens=self._max_budget,
            system_prompt=(
                'You are a sub-agent handling a specific task. '
                'Use the available tools to complete the task thoroughly, '
                'then provide a clear summary of what you found or did.'
            ),
        )

        runner = TurnLoopRunner(
            llm=self._llm,
            tools=sub_tools,
            config=config,
      

    def _compress_result(self, result) -> str:
        """Compress the sub-agent's result into a concise string."""
        lines = [result.final_output]
        lines.append('')
        lines.append(
            f'[Sub-agent: {result.turns_used} turns, '
            f'{result.total_usage.total_tokens} tokens, '
            f'stop={result.stop_reason}]'
        )
        if result.tool_calls_made:
            lines.append(f'[Tools used: {", ".join(result.tool_calls_made)}]')
        return '\n'.join(lines)