harnest.eval_capture

All Python modules · Import and compatibility guidance

Exports

Name	Kind
`CAPTURE_SCHEMA_VERSION`	Value or type alias
`CaptureEvent`	Class
`CaptureRecord`	Class
`CaptureStore`	Class
`LocalCaptureStore`	Class
`ToolActivity`	Class
`default_redactor`	Function
`redact_text`	Function

CAPTURE_SCHEMA_VERSION

CAPTURE_SCHEMA_VERSION = 1

Example

from harnest.eval_capture import CAPTURE_SCHEMA_VERSION

CaptureEvent

Facts about one finished invocation, offered to the capture rule. The event exists only in memory. user_id, session_id and metadata let a rule target specific traffic, but none is copied into the stored record.

CaptureEvent(
    agent_name: str,
    invocation_id: str,
    session_id: str,
    user_id: str,
    transport: str | None,
    status: str,
    input_text: str,
    output_text: str,
    tool_calls: tuple[ToolActivity, ...] = (),
    error_type: str | None = None,
    metadata: Mapping[str, Any] = ...,
) -> None

Fields (— means no declared default):

Field	Type	Default
`agent_name`	`str`	`—`
`invocation_id`	`str`	`—`
`session_id`	`str`	`—`
`user_id`	`str`	`—`
`transport`	`str \| None`	`—`
`status`	`str`	`—`
`input_text`	`str`	`—`
`output_text`	`str`	`—`
`tool_calls`	`tuple[ToolActivity, ...]`	`()`
`error_type`	`str \| None`	`None`
`metadata`	`Mapping[str, Any]`	`dict()`

Parameters and example

Parameter	Type	Default
`agent_name`	`str`	Required
`invocation_id`	`str`	Required
`session_id`	`str`	Required
`user_id`	`str`	Required
`transport`	`str \| None`	Required
`status`	`str`	Required
`input_text`	`str`	Required
`output_text`	`str`	Required
`tool_calls`	`tuple[ToolActivity, ...]`	`()`
`error_type`	`str \| None`	`None`
`metadata`	`Mapping[str, Any]`	`...`

from harnest.eval_capture import CaptureEvent

event = CaptureEvent(
    agent_name=...,
    invocation_id=...,
    session_id=...,
    user_id=...,
    transport=...,
    status=...,
    input_text=...,
    output_text=...,
)

CaptureEvent.failed

Return whether the invocation raised instead of completing.

Example

from harnest.eval_capture import CaptureEvent

event: CaptureEvent
value = event.failed

CaptureEvent.tool_error

Return whether any tool result looked like an error. This is a heuristic over framework-neutral events (an error field or an error status); write an explicit check in the rule when it matters.

Example

from harnest.eval_capture import CaptureEvent

event: CaptureEvent
value = event.tool_error

CaptureRecord

One invocation with structure-only or team-redacted content. content="omitted" serializes empty input/output text and null tool arguments/results. content="redacted" records that the team’s redactor ran, not a guarantee that all personal data was removed.

CaptureRecord(
    id: str,
    captured_at: str,
    agent_name: str,
    invocation_id: str,
    reasons: tuple[str, ...],
    status: str,
    input_text: str,
    output_text: str,
    tool_calls: tuple[ToolActivity, ...] = (),
    error_type: str | None = None,
    content: Literal['omitted', 'redacted'] = 'omitted',
    schema_version: int = 1,
) -> None

Fields (— means no declared default):

Field	Type	Default
`id`	`str`	`—`
`captured_at`	`str`	`—`
`agent_name`	`str`	`—`
`invocation_id`	`str`	`—`
`reasons`	`tuple[str, ...]`	`—`
`status`	`str`	`—`
`input_text`	`str`	`—`
`output_text`	`str`	`—`
`tool_calls`	`tuple[ToolActivity, ...]`	`()`
`error_type`	`str \| None`	`None`
`content`	`Literal['omitted', 'redacted']`	`'omitted'`
`schema_version`	`int`	`1`

Parameters and example

Parameter	Type	Default
`id`	`str`	Required
`captured_at`	`str`	Required
`agent_name`	`str`	Required
`invocation_id`	`str`	Required
`reasons`	`tuple[str, ...]`	Required
`status`	`str`	Required
`input_text`	`str`	Required
`output_text`	`str`	Required
`tool_calls`	`tuple[ToolActivity, ...]`	`()`
`error_type`	`str \| None`	`None`
`content`	`Literal['omitted', 'redacted']`	`'omitted'`
`schema_version`	`int`	`1`

from harnest.eval_capture import CaptureRecord

record = CaptureRecord(
    id=...,
    captured_at=...,
    agent_name=...,
    invocation_id=...,
    reasons=...,
    status=...,
    input_text=...,
    output_text=...,
)

CaptureRecord.to_json

CaptureRecord.to_json() -> dict[str, Any]

Serialize schema version 1 without content unless explicitly redacted.

Example

from harnest.eval_capture import CaptureRecord

record: CaptureRecord
result = record.to_json()

CaptureRecord.from_json

CaptureRecord.from_json(value: Mapping[str, Any]) -> 'CaptureRecord'

Read a record, omitting content without an explicit content policy.

Parameters and example

Parameter	Type	Default
`value`	`Mapping[str, Any]`	Required

from harnest.eval_capture import CaptureRecord

record = CaptureRecord.from_json(value=...)

CaptureStore

Storage for capture records, registered with @lifecycle.storage.custom.

CaptureStore(*args, **kwargs)

Example

from harnest.eval_capture import CaptureStore

class MyCaptureStore(CaptureStore):
    ...

CaptureStore.start

async CaptureStore.start() -> None

Open the underlying storage resource.

Example

from harnest.eval_capture import CaptureStore

store: CaptureStore
await store.start()

CaptureStore.close

async CaptureStore.close() -> None

Release the underlying storage resource.

Example

from harnest.eval_capture import CaptureStore

store: CaptureStore
await store.close()

CaptureStore.append

async CaptureStore.append(record: CaptureRecord) -> None

Persist one structure-only or team-redacted record.

Parameters and example

Parameter	Type	Default
`record`	`CaptureRecord`	Required

from harnest.eval_capture import CaptureStore

store: CaptureStore
await store.append(record=...)

CaptureStore.list

async CaptureStore.list(*, since: datetime | None = None, limit: int = 100) -> Sequence[CaptureRecord]

Return records captured at or after since, newest first.

Parameters and example

Parameter	Type	Default
`since`	`datetime \| None`	`None`
`limit`	`int`	`100`

from harnest.eval_capture import CaptureStore

store: CaptureStore
result = await store.list()

LocalCaptureStore

Reference store writing one JSON Lines file per UTC day. Suitable for local development and single-instance deployments with a persistent volume. Files older than retention_days are deleted on start and whenever a new day’s file is opened.

LocalCaptureStore(directory: str | Path, *, retention_days: int = 30) -> None

Parameters and example

Parameter	Type	Default
`directory`	`str \| Path`	Required
`retention_days`	`int`	`30`

from harnest.eval_capture import LocalCaptureStore

store = LocalCaptureStore(directory=...)

LocalCaptureStore.start

async LocalCaptureStore.start() -> None

Create the directory and apply retention.

Example

from harnest.eval_capture import LocalCaptureStore

store: LocalCaptureStore
await store.start()

LocalCaptureStore.close

async LocalCaptureStore.close() -> None

Nothing to release; every append closes its file.

Example

from harnest.eval_capture import LocalCaptureStore

store: LocalCaptureStore
await store.close()

LocalCaptureStore.append

async LocalCaptureStore.append(record: CaptureRecord) -> None

Append one record to today’s file, serialized across tasks.

Parameters and example

Parameter	Type	Default
`record`	`CaptureRecord`	Required

from harnest.eval_capture import LocalCaptureStore

store: LocalCaptureStore
await store.append(record=...)

LocalCaptureStore.list

async LocalCaptureStore.list(*, since: datetime | None = None, limit: int = 100) -> Sequence[CaptureRecord]

Read records newest first, stopping at limit.

Parameters and example

Parameter	Type	Default
`since`	`datetime \| None`	`None`
`limit`	`int`	`100`

from harnest.eval_capture import LocalCaptureStore

store: LocalCaptureStore
result = await store.list()

ToolActivity

One tool call in a captured invocation, with its result when one arrived.

ToolActivity(name: str, arguments: Any = None, result: Any = None, error: bool = False) -> None

Fields (— means no declared default):

Field	Type	Default
`name`	`str`	`—`
`arguments`	`Any`	`None`
`result`	`Any`	`None`
`error`	`bool`	`False`

Parameters and example

Parameter	Type	Default
`name`	`str`	Required
`arguments`	`Any`	`None`
`result`	`Any`	`None`
`error`	`bool`	`False`

from harnest.eval_capture import ToolActivity

activity = ToolActivity(name=...)

ToolActivity.to_json

ToolActivity.to_json() -> dict[str, Any]

Return a JSON-safe mapping for stores.

Example

from harnest.eval_capture import ToolActivity

activity: ToolActivity
result = activity.to_json()

ToolActivity.from_json

ToolActivity.from_json(value: Mapping[str, Any]) -> 'ToolActivity'

Rebuild one tool activity written by to_json.

Parameters and example

Parameter	Type	Default
`value`	`Mapping[str, Any]`	Required

from harnest.eval_capture import ToolActivity

activity = ToolActivity.from_json(value=...)

default_redactor

Omit all input, output and tool payloads, preserving only structure. This existing helper name remains importable, but no longer guesses which substrings contain personal data. Register @lifecycle.evals.redact to retain content under a team-owned policy instead.

default_redactor(record: CaptureRecord) -> CaptureRecord

Parameters and example

Parameter	Type	Default
`record`	`CaptureRecord`	Required

from harnest.eval_capture import default_redactor

record = default_redactor(record=...)

redact_text

Omit an entire string; retained for import compatibility, not PII detection.

redact_text(value: str) -> str

Parameters and example

Parameter	Type	Default
`value`	`str`	Required

from harnest.eval_capture import redact_text

result = redact_text(value=...)

Overview

Get started

Harnest Studio

Guides

API reference

harnest.eval_capture

Exports

CAPTURE_SCHEMA_VERSION

CaptureEvent

CaptureEvent.failed

CaptureEvent.tool_error

CaptureRecord

CaptureRecord.to_json

CaptureRecord.from_json

CaptureStore

CaptureStore.start

CaptureStore.close

CaptureStore.append

CaptureStore.list

LocalCaptureStore

LocalCaptureStore.start

LocalCaptureStore.close

LocalCaptureStore.append

LocalCaptureStore.list

ToolActivity

ToolActivity.to_json

ToolActivity.from_json

default_redactor

redact_text

​Exports

​CAPTURE_SCHEMA_VERSION

​CaptureEvent

​CaptureEvent.failed

​CaptureEvent.tool_error

​CaptureRecord

​CaptureRecord.to_json

​CaptureRecord.from_json

​CaptureStore

​CaptureStore.start

​CaptureStore.close

​CaptureStore.append

​CaptureStore.list

​LocalCaptureStore

​LocalCaptureStore.start

​LocalCaptureStore.close

​LocalCaptureStore.append

​LocalCaptureStore.list

​ToolActivity

​ToolActivity.to_json

​ToolActivity.from_json

​default_redactor

​redact_text

Exports

CAPTURE_SCHEMA_VERSION

CaptureEvent

CaptureEvent.failed

CaptureEvent.tool_error

CaptureRecord

CaptureRecord.to_json

CaptureRecord.from_json

CaptureStore

CaptureStore.start

CaptureStore.close

CaptureStore.append

CaptureStore.list

LocalCaptureStore

LocalCaptureStore.start

LocalCaptureStore.close

LocalCaptureStore.append

LocalCaptureStore.list

ToolActivity

ToolActivity.to_json

ToolActivity.from_json

default_redactor

redact_text