Harness and Session¶

hw.load(path) is the same as Harness.load(path).

Harness ¶

Harness(
    effective: EffectiveSpec,
    *,
    registry: Registry,
    bus: HookBus,
    tasks: Mapping[str, TaskEntry],
    models: ModelAccess,
    state: StateStore,
    prompts: PromptStore,
    tools: ToolBackend | None,
    clock: Clock,
    telemetry: Telemetry | None = None,
    governor: ToolGovernor | None = None,
    policy: PolicyEngine | None = None,
    approvals: ApprovalChannel | None = None,
    context: ContextGovernor | None = None,
    memory: Memory | None = None,
)

Bases: ReplayMixin

The runtime built from one app's effective spec.

Build it with load or from_spec. Use async with harness: to submit background runs; run and session work without it.

spec `property` ¶

spec: EffectiveSpec

The effective spec this harness was built from.

explain ¶

explain() -> list[ExplainRow]

Every effective value with its layer, lock and bounds (spec explain).

runner_of ¶

runner_of(task: str) -> RunnerInfo

The runner a task uses, with the hook points it reaches.

Raises:

Type	Description
`NotFound`	For an unknown task.

hooks ¶

hooks(
    task: str | None = None,
) -> dict[HookPoint, list[Registration]]

The middleware chain of every hook point, in execution order.

stream `async` ¶

stream(run_id: str) -> AsyncIterator[RunEvent]

Yield a run's events: its stored history, then live events until it stops.

status and approval events come from the state store; trace and chunk events are only seen while the run is live in this process.

run `async` ¶

run(
    task: str,
    input: Any = None,
    *,
    principal: Principal | str = "anonymous",
    tenant: str = "default",
    idempotency_key: str | None = None,
    budget: Mapping[str, object] | None = None,
    attributes: Mapping[str, str] | None = None,
) -> RunResult

Run task to a terminal state, or until it suspends for approval.

A repeated idempotency_key returns the first run's result instead of running again (DESIGNS No. 18). budget may lower the task's overridable limits for this run (R10).

Raises:

Type	Description
`NotFound`	For an unknown task.
`BoundViolation`	For a budget override that isn't allowed.

submit `async` ¶

submit(
    task: str,
    input: Any = None,
    *,
    principal: Principal | str = "anonymous",
    tenant: str = "default",
    idempotency_key: str | None = None,
    budget: Mapping[str, object] | None = None,
    attributes: Mapping[str, str] | None = None,
) -> str

Start task in the background and return its run id at once.

Needs async with harness:. Follow it with stream and result.

Raises:

Type	Description
`SpecError`	Outside `async with harness:`.

result `async` ¶

result(run_id: str) -> RunResult

The latest state of a run.

Raises:

Type	Description
`NotFound`	For an unknown run.

approve `async` ¶

approve(
    run_id: str,
    approval_id: str,
    decision: ApprovalAnswer = "approve",
    *,
    approver: Principal | str,
    edited_args: dict[str, JsonValue] | None = None,
    reason: str | None = None,
) -> RunResult

Answer a pending approval and continue the run (DESIGNS No. 18).

Checks the approver holds the request's approver_role.
Claims the request in the state store, so one answer wins (ADR-0044).
Fires approval.resolved.
On approve or edit, resumes; edited arguments go through tool.before again. On reject, the run ends blocked.

An expired request ends the run as approvals.on_timeout says.

Raises:

Type	Description
`NotFound`	If the run has no such pending approval, another answer won it, or this process is still suspending the run.
`PolicyDenied`	If the approver lacks the required role.
`ValidationFailed`	For an `edit` without arguments, or of an output.

resume `async` ¶

resume(run_id: str) -> RunResult

Continue a run whose approval was answered but which didn't finish (a crash).

Raises:

Type	Description
`NotFound`	If the run has nothing to resume, or is still waiting for approval.

cancel `async` ¶

cancel(run_id: str) -> RunResult

Cancel a run that is running here or waiting for approval.

A finished run is returned unchanged.

Raises:

Type	Description
`NotFound`	For an unknown run.

authorize `async` ¶

authorize(
    tool: str,
    args: Mapping[str, JsonValue] | None = None,
    *,
    principal: Principal | str = "anonymous",
    tenant: str = "default",
    task: str | None = None,
    labels: Labels | None = None,
) -> ToolVerdict

What the governor would decide about one tool call, without running it.

Steps 1 to 7 run as for a real call (DESIGNS No. 17) on a run with this principal, tenant, task and labels, as a dry run: a rate limit isn't used up and nothing is approved or executed. An approval's approver_role falls back to the task's and then the spec's approvals.approver_role, as the request a real run makes would. harnessweaver policy test is built on this.

Raises:

Type	Description
`NotFound`	For a task the spec doesn't define.

decide_policy `async` ¶

decide_policy(request: PolicyRequest) -> PolicyDecision

What the spec's policy engine answers, alone: policy.rules combined as every engine combines them (ADR-0025), without the governor's other steps.

authorize gives the governor's full verdict on a tool call; POST /v1/policy/evaluate on harnessd is built on this.

output_type ¶

output_type(task: str) -> type[Any] | None

The type a task's output is parsed into, or None when the task returns whatever the model says.

Raises:

Type	Description
`NotFound`	For a task the spec doesn't define.

output_schema ¶

output_schema(task: str) -> dict[str, JsonValue] | None

The JSON Schema of a task's output as run results carry it, or None when the task returns whatever the model says.

Raises:

Type	Description
`NotFound`	For a task the spec doesn't define.

samples `async` ¶

samples(task: str | None = None) -> list[OnlineSample]

The production runs evals.online sampled and graded, oldest first (harnessweaver eval report).

replay `async` ¶

replay(run_id: str) -> ReplayReport

Run a recorded run's task again against the current code and spec.

The model answers and tool results come from the recording, so nothing calls out and no tool executes; recorded approvals answer the approvals the replay asks for. harnessweaver replay prints the report.

Raises:

Type	Description
`NotFound`	For an unknown run.
`SpecError`	For a run that has no recording or no result yet, or whose task the spec no longer defines.

load `classmethod` ¶

load(
    path: str | Path,
    *,
    env: str | None = None,
    profile_dirs: Sequence[Path] = (),
    clock: Clock | None = None,
    tools: ToolBackend | None = None,
    catalog: PluginCatalog | None = None,
    telemetry: Telemetry | None = None,
) -> Self

Load harness.yaml (with profiles and overlay) and build the harness.

Loading is synchronous start-up work: it reads files and imports the plugins the spec references.

Raises:

Type	Description
`SpecError`	For anything wrong with the spec, its plugins or its tasks.

from_spec `classmethod` ¶

from_spec(
    effective: EffectiveSpec,
    *,
    clock: Clock | None = None,
    tools: ToolBackend | None = None,
    catalog: PluginCatalog | None = None,
    telemetry: Telemetry | None = None,
    model: ModelProvider | None = None,
    state: StateStore | None = None,
) -> Self

Build a harness from an already loaded spec. See load.

telemetry replaces the harness's own (for example to use specific OpenTelemetry providers); observability.capture_content still applies, and the harness then configures no OpenTelemetry SDK itself. model replaces models.provider with a provider object, for tests and offline runs (harnessweaver.testing.harness_for_test); aliases, prices and allow-lists still apply. state replaces state.use with a store object, as harnessweaver eval run does to keep eval runs out of the app's store.

aclose `async` ¶

aclose() -> None

Release the state store's resources.

instrument ¶

instrument(
    client: T,
    *,
    task: str | None = None,
    principal: Principal | str = "anonymous",
    tenant: str = "default",
    attributes: Mapping[str, str] | None = None,
    adapter: str | None = None,
) -> T

Return a governed version of client (DESIGNS No. 15, L0).

The adapter is found from the client's type: the plugin whose namespace is the client's top-level package (openai, anthropic) is loaded and asked for an adapter that wraps the type. Name one with adapter.

Raises:

Type	Description
`UnknownProvider`	If no installed plugin wraps the client's type.
`PluginNotAllowed`	If that plugin isn't in `plugins.allow`.
`UnreachableHook`	If the adapter can't reach a hook the baseline or the budgets need (R4).

wrap ¶

wrap(
    agent: T,
    *,
    task: str | None = None,
    principal: Principal | str = "anonymous",
    tenant: str = "default",
    attributes: Mapping[str, str] | None = None,
    adapter: str | None = None,
) -> T

Return a governed version of an agent from another framework (DESIGNS No. 15).

agent = harness.wrap(my_pydantic_agent, task="support_agent"): the adapter is found from the agent's type as instrument finds one, and each run of the returned agent is governed: a run of its own, or part of the harness run or session it runs inside. task names the task whose budget and middleware apply.

Raises:

Type	Description
`UnknownProvider`	If no installed plugin wraps the agent's type.
`PluginNotAllowed`	If that plugin isn't in `plugins.allow`.
`UnreachableHook`	If the adapter can't reach a hook the baseline or the budgets need (R4).

evaluate `async` ¶

evaluate(dataset: EvalDataset | str | Path) -> SuiteResult

Run an eval suite: each case's task run, graded (DESIGNS No. 23).

A dataset path is read relative to the spec's directory. Judges grade inside a session, so their model calls are governed and budgeted. harnessweaver eval run adds baselines and the shadow-guard report.

Raises:

Type	Description
`SpecError`	For a dataset that isn't valid, or graders that can't be built.
`NotFound`	For a task the spec doesn't define.

session ¶

session(
    task: str | None = None,
    *,
    principal: Principal | str = "anonymous",
    tenant: str = "default",
    budget: Budget | None = None,
    attributes: Mapping[str, str] | None = None,
) -> Session

Open a governed session: async with harness.session(...) as s: (DESIGNS No. 15).

A session is a run whose flow the app owns. Its budget is the spec's budgets.per_run, tightened by budget and by the task's own limits when task names a task in the spec. Suspend escalates here, because the app's code can't be resumed later.

Session ¶

Session(
    ctx: RunContext | None,
    *,
    harness: Harness | None = None,
    pending: PendingSession | None = None,
    step: str | None = None,
)

Governed model, tool, routing and validation calls inside one run.

Get one from harness.session(...) (standalone: the session is the run) or as the first argument of a session/workflow runner handler.

result `instance-attribute` ¶

result: RunResult | None = None

For a standalone session, the run's result once the async with exits.

run_context `property` ¶

run_context: RunContext

The underlying run context, for adapters.

memory `property` ¶

memory: SessionMemory

Entries kept for this tenant, principal and task (DESIGNS No. 21).

for_run `classmethod` ¶

for_run(
    ctx: RunContext, *, step: str | None = None
) -> Self

A session view of a run the harness is driving (for runner handlers).

prompt `async` ¶

prompt(ref: str, /, **variables: Any) -> str

Render a versioned prompt (ADR-0013).

generate `async` ¶

generate(
    *,
    prompt: str | None = None,
    messages: Sequence[Message] = (),
    model: str | None = None,
    output: type[Any] | None = None,
    stream: bool = False,
    temperature: float | None = None,
    max_tokens: int | None = None,
    **variables: Any,
) -> Any

Ask the model for an answer, governed end to end.

prompt is rendered with variables and sent as a user message after messages. With output, the answer is parsed into that type and repaired up to validation.max_repairs times.

Raises:

Type	Description
`RunInterrupted`	When a hook or a budget halts the run.
`ProviderError`	When the model provider fails.

tool `async` ¶

tool(name: str, /, **arguments: JsonValue) -> JsonValue

Call a tool, governed (tool.before / tool.after). Returns its output.

decide `async` ¶

decide(
    router: str,
    *,
    text: str | None = None,
    options: Iterable[str] = (),
    state: dict[str, JsonValue] | None = None,
    default: str | None = None,
) -> Decision

Ask the routing middleware to choose (route.decide).

context ¶

context(source: str) -> ContextSource

Governed access to a context source.

guard `async` ¶

guard(
    stage: Literal["input"], content: JsonValue
) -> JsonValue

Run the input guards on content the app received (input.before).

validate `async` ¶

validate(
    stage: Literal["output"],
    value: Any,
    *,
    output: type[Any] | None = None,
) -> Any

Check a final output (output.before). Returns it, possibly modified.

With output, the value is first parsed into that type, as generate parses a model's answer (a JSON string is decoded first).

Raises:

Type	Description
`ValidationFailed`	When the value isn't an `output`, or a middleware asks for a repair; in session style the app decides how to repair.
`RunInterrupted`	When a hook halts the run.

Harness and Session¶

Harness ¶

spec property ¶

explain ¶

runner_of ¶

hooks ¶

stream async ¶

run async ¶

submit async ¶

result async ¶

approve async ¶

resume async ¶

cancel async ¶

authorize async ¶

decide_policy async ¶

output_type ¶

output_schema ¶

samples async ¶

replay async ¶

load classmethod ¶

from_spec classmethod ¶

aclose async ¶

instrument ¶

wrap ¶

evaluate async ¶

session ¶

Session ¶

result instance-attribute ¶

run_context property ¶

memory property ¶

for_run classmethod ¶

prompt async ¶

generate async ¶

tool async ¶

decide async ¶

context ¶

guard async ¶

validate async ¶

spec `property` ¶

stream `async` ¶

run `async` ¶

submit `async` ¶

result `async` ¶

approve `async` ¶

resume `async` ¶

cancel `async` ¶

authorize `async` ¶

decide_policy `async` ¶

samples `async` ¶

replay `async` ¶

load `classmethod` ¶

from_spec `classmethod` ¶

aclose `async` ¶

evaluate `async` ¶

result `instance-attribute` ¶

run_context `property` ¶

memory `property` ¶

for_run `classmethod` ¶

prompt `async` ¶

generate `async` ¶

tool `async` ¶

decide `async` ¶

guard `async` ¶

validate `async` ¶