trading_agent.skills.skill_usage

Records each use of a skill, and whether the turn that used it went smoothly.

The tracker listens to the event bus, like the journal. A load_skill call opens a use; every failed tool call after it in the same turn counts against it; when the desk's turn finishes, the use is written as one JSON line to workspace/desks/<desk>/skills/usage.jsonl. A turn that ends in an error marks the use as failed.

"Went smoothly" is a rough signal, not a verdict: a tool can fail for reasons the skill did not cause. The curator reads these numbers as hints about which skills to review.

Typical usage example:

tracker = SkillUsageTracker(event_bus, layout) task = asyncio.create_task(tracker.run()) statistics = SkillUsageReader(usage_file).statistics()

SkillStatistics dataclass

SkillStatistics(uses, troubled_uses, last_used)

How one skill has been used.

Attributes:

Name	Type	Description
uses	int	How many turns loaded it.
troubled_uses	int	How many of those turns had failed tool calls after loading it, or ended in an error.
last_used	str	The ISO 8601 time of the last use.

SkillUsageTracker

SkillUsageTracker(event_bus, layout)

Turns the event stream into skill-use records.

Subscribes to the bus at once, so no load_skill call is missed.

Parameters:

Name	Type	Description	Default
event_bus	EventBus	The bus to listen to.	required
layout	WorkspaceLayout	Where each desk's usage file goes.	required

Source code in src/trading_agent/skills/skill_usage.py

def __init__(
    self,
    event_bus: event_bus_module.EventBus,
    layout: workspace_layout.WorkspaceLayout,
):
    """Subscribes to the bus at once, so no load_skill call is missed.

    Args:
        event_bus (event_bus_module.EventBus): The bus to listen to.
        layout (workspace_layout.WorkspaceLayout): Where each desk's usage file goes.
    """
    self._layout = layout
    self._subscription = event_bus.subscribe(maximum_queued=_QUEUE_SIZE)
    self._open_uses: dict[str, list[_OpenUse]] = {}
    self._failed_desks = set()

run async

run()

Handles events until the task is cancelled.

Source code in src/trading_agent/skills/skill_usage.py

async def run(self) -> None:
    """Handles events until the task is cancelled."""
    try:
        while True:
            published_event = await self._subscription.next_event()
            self.handle(published_event)
    finally:
        self._subscription.close()

handle

handle(published_event)

Updates the open uses with one event.

Parameters:

Name	Type	Description	Default
published_event	Event	The event.	required

Source code in src/trading_agent/skills/skill_usage.py

def handle(self, published_event: event_module.Event) -> None:
    """Updates the open uses with one event.

    Args:
        published_event (event_module.Event): The event.
    """
    desk = published_event.desk
    if desk is None:
        return
    kind = published_event.kind
    payload = published_event.payload
    if kind == event_module.EventKind.TOOL_CALL:
        if payload.get('name') == _LOAD_SKILL_TOOL_NAME:
            arguments = payload.get('arguments') or {}
            self._open_uses.setdefault(desk, []).append(
                _OpenUse(
                    skill=str(arguments.get('name', '')),
                    call_id=str(payload.get('call_id', '')),
                    trigger_id=str(payload.get('trigger_id', '')),
                )
            )
    elif kind == event_module.EventKind.TOOL_RESULT:
        self._count_result(desk, payload)
    elif kind == event_module.EventKind.ERROR:
        if self._open_uses.get(desk):
            self._failed_desks.add(desk)
    elif kind == event_module.EventKind.TURN_FINISHED:
        self._finish(desk)

SkillUsageReader

SkillUsageReader(usage_file)

Summarises one desk's usage file.

Creates the reader.

Parameters:

Name	Type	Description	Default
usage_file	Path	The desk's usage.jsonl.	required

Source code in src/trading_agent/skills/skill_usage.py

def __init__(self, usage_file: Path):
    """Creates the reader.

    Args:
        usage_file (Path): The desk's usage.jsonl.
    """
    self._usage_file = usage_file

statistics

statistics()

Counts each skill's uses.

Returns:

Type	Description
dict[str, SkillStatistics]	The statistics, keyed by skill name; skills never used are absent.

Source code in src/trading_agent/skills/skill_usage.py

def statistics(self) -> dict[str, SkillStatistics]:
    """Counts each skill's uses.

    Returns:
        dict[str, SkillStatistics]: The statistics, keyed by skill name; skills never used are absent.
    """
    uses = {}
    troubled = {}
    last_used = {}
    if not self._usage_file.is_file():
        return {}
    for line in self._usage_file.read_text(encoding='utf-8').splitlines():
        try:
            record = json.loads(line)
        except ValueError:
            continue
        name = record.get('skill', '')
        uses[name] = uses.get(name, 0) + 1
        if record.get('tool_errors_after') or record.get('turn_failed'):
            troubled[name] = troubled.get(name, 0) + 1
        last_used[name] = max(last_used.get(name, ''), record.get('at', ''))
    statistics = {}
    for name, count in uses.items():
        statistics[name] = SkillStatistics(
            uses=count,
            troubled_uses=troubled.get(name, 0),
            last_used=last_used[name],
        )
    return statistics