Context

The shared context stores metrics that are written to log storage each time a WorkflowRunner instance completes:

Step, stage and workflow durations
OpenAI token consumption rates
Downloaded bytes
And more

StepContextFilter

Bases: Filter

Stamp the executing workflow stage/step onto warnings and errors, everywhere, once.

Rather than teaching every client, extractor and connector to name the step it is running under, this filter reads the :data:step_var context set by the WorkflowRunner and appends [step: <stage> / <name>] to any WARNING or higher record. A truncation warning from the OpenAI client, a parse failure from the row extractor, an Azure error from a connector — all name their origin, with no per-module code and no change to anyone's log format.

INFO/DEBUG are left untouched to keep routine logging clean. The stamp is applied at most once per record (guarded by an attribute) so that attaching the filter to several handlers cannot append the suffix twice to the same shared record.

get_current_step

get_current_step()

The 'stage / step' label of the currently executing workflow step, or None.

get_metrics

get_metrics()

Retrieve a deep copy of the current metrics dictionary.

Returns:

Type	Description
`dict[str, Any]`	The context dictionary.

init_metrics

init_metrics()

Reset the metrics context for a new workflow run.

install_step_context_logging

install_step_context_logging(logger=None)

Attach a :class:StepContextFilter to a logger's handlers (idempotent).

Defaults to the root logger, so a single call annotates every handler in the process. Safe to call repeatedly — a handler that already carries the filter is skipped — which is why the WorkflowRunner can install it on every run without accumulating filters. If logging has not been configured yet (no handlers), this is a harmless no-op.

log_metric

log_metric(path, value)

Log a metric using a dotted path for nested metrics, e.g., "stage1.step1.tokens". It will sum numeric values or append string values to the list if keys already exist.

When rendering metrics in the Hoppa web app workspace dashboard, the app expects to find certain keys in the metrics JSON. An example JSON structure is provided below to help you construct the right dot notation when creating new class methods or functions.

{
    "duration_sec": {
        "stages": {
            "Hoppa Connector": {
                "Establish Hoppa session connection and get document": 0.3196,
                "total": 0.8427
            },
            "File Preprocessing": {
                "Preprocess file": 0.3173,
                "total": 0.8587
            },
            "Classification": {
                "Generate document description": 4.3365,
                "Classify site using ISO4 Spatial Layout classifier": 10.5834,
                "total": 15.9178
            },
            "Search Terms": {
                "Run search terms": 2.4235,
                "total": 2.9494
            }
        },
        "total": 20.7674
    },
    "bytes": 186069,
    "file_type": "svg",
    "counts": {
        "document_characters": 186069,
        "chat": {
            "prompt_tokens": 23645,
            "completion_tokens": 611,
            "total_tokens": 24256
        }
        "descriptions": 1
        "description_characters": 1399
        "classifiers": 6
        "searchTerms": 6
        "tags": 2
    },
}

Parameters:

Name	Type	Description	Default
`path`	`str`	The dotted path to insert into the context dictionary	required
`value`	`Any`	The value to be inserted	required

merge_metrics_into_main

merge_metrics_into_main(thread_metrics)

Merge collected metrics from a thread into the main metrics context. Supports nested dictionaries and sums numeric values.

Parameters:

Name	Type	Description	Default
`thread_metrics`	`dict`	Dictionary of key-value pairs to insert. Expects pre-structured (nested) dictionaries, as opposed to dot notation.	required

set_current_step

set_current_step(label)

Record the stage/step now executing (see :data:step_var).

Not reset between steps: each step overwrites it before doing any work, and a thread runs at most one step at a time, so a stale label can never be read during another step. A new thread starts with the default (None), so this never leaks across threads unless a caller deliberately propagates it (as the row extractors do into their worker pool).

truncate_context

truncate_context(context, max_bytes=30000)

Truncate context dictionary to fit within byte limit.

Parameters:

Name	Type	Description	Default
`context`	`Dict[str, Any]`	Context dictionary to truncate	required
`max_bytes`	`int`	Maximum size in bytes. Defaults to 30000.	`30000`

Returns:

Name	Type	Description
`str`	`str`	JSON string representation of context, truncated if necessary