trading_agent.backends.claude_pre_compact_hook

A PreCompact hook that records each time Claude Code compacts a desk's session.

trading-agent normally compacts Claude's session itself, between turns at 70% of the window, after a memory-flush turn. Claude Code's own automatic compaction is the safety net for one very long turn. When it fires, this hook publishes the event and tells the turn runner, so the runner can send the instruction files again on the next turn, because a compacted session may have lost updated instructions that were delivered inside earlier messages.

Typical usage example:

hook = ClaudePreCompactHook('main', 'claude', event_bus, runner.note_compaction) hooks = {'PreCompact': [claude_agent_sdk.HookMatcher(hooks=[hook])]}

ClaudePreCompactHook

ClaudePreCompactHook(desk_name, backend_name, event_bus, on_compaction)

Tells the runner about every compaction, and publishes the automatic ones.

A manual compaction is always one the turn runner started with /compact, and the runner publishes a fuller event for it itself, so the hook does not publish a second one.

Creates the hook.

Parameters:

Name	Type	Description	Default
desk_name	str	The desk, used to tag the event.	required
backend_name	str	The backend's name in trading_agent.toml.	required
event_bus	EventBus	Where the event is published.	required
on_compaction	Callable[[], None]	Called once per compaction, so the runner can resend its instructions.	required

Source code in src/trading_agent/backends/claude_pre_compact_hook.py

def __init__(
    self,
    desk_name: str,
    backend_name: str,
    event_bus: event_bus_module.EventBus,
    on_compaction: Callable[[], None],
):
    """Creates the hook.

    Args:
        desk_name (str): The desk, used to tag the event.
        backend_name (str): The backend's name in trading_agent.toml.
        event_bus (event_bus_module.EventBus): Where the event is published.
        on_compaction (Callable[[], None]): Called once per compaction, so the runner can resend its instructions.
    """
    self._desk_name = desk_name
    self._backend_name = backend_name
    self._event_bus = event_bus
    self._on_compaction = on_compaction

__call__ async

__call__(hook_input, tool_use_id, context)

Notes the compaction, publishes it if it is automatic, and lets it go ahead.

Parameters:

Name	Type	Description	Default
hook_input	dict[str, Any]	The hook's input from Claude Code, including trigger ("auto" or "manual").	required
tool_use_id	str | None	Unused; compaction is not a tool call.	required
context	Any	The hook context, unused.	required

Returns:

Type	Description
dict[str, Any]	An empty dict, which lets the compaction proceed.

Source code in src/trading_agent/backends/claude_pre_compact_hook.py

async def __call__(
    self,
    hook_input: dict[str, Any],
    tool_use_id: str | None,
    context: Any,
) -> dict[str, Any]:
    """Notes the compaction, publishes it if it is automatic, and lets it go ahead.

    Args:
        hook_input (dict[str, Any]): The hook's input from Claude Code, including trigger ("auto" or "manual").
        tool_use_id (str | None): Unused; compaction is not a tool call.
        context (Any): The hook context, unused.

    Returns:
        dict[str, Any]: An empty dict, which lets the compaction proceed.
    """
    del tool_use_id, context
    self._on_compaction()
    trigger = hook_input.get('trigger', 'auto')
    if trigger == 'manual':
        return {}
    self._event_bus.publish(
        event_module.Event.now(
            event_module.EventKind.CONTEXT_COMPRESSED,
            self._desk_name,
            {
                'backend': self._backend_name,
                'reason': f'claude_code_{trigger}',
            },
        )
    )
    return {}