Harness¶

Cluster: Uncategorised | Type: component | MCP Tools: None

Overview¶

Workflow execution harness with per-step verification against a structured failure mode library and optional workflow-specific rubrics. Runs callable workflow steps, captures outputs, applies StepVerifier checks (severity-gated against built-in failure modes), evaluates user-defined checklist items, imports/exports editable JSON/YAML rubric files, and aggregates results into a HarnessResult with pass rate and failed-step drill-down.

The component also exposes a Capability Harness path for structured Scenario evaluation through fixed A/B/C/D strata bars. HarnessBlock.run_scenario runs under an explicit port_mode (fake/mixed/real) and returns a canonical envelope with completion_state, warning_card, evidence, request_id, task_id, and run_id. completion_state: verified is reserved for the real-port release lane (real ports + no release-blocking gate); fake, mixed, and legacy CI fake-mode paths are never verified, and if real ports cannot be constructed the run fails closed with G6_E_HARNESS_REAL_PORTS_UNAVAILABLE. A passing workflow graded by an auto-generated, non-expert-reviewed rubric is downgraded to qualified-draft (G6_E_HARNESS_RUBRIC_NOT_REVIEWED) until the caller sets rubric_expert_reviewed. Blocked release conditions are surfaced as blocked-escalated, not hidden as success. This real-vs-fake-port and rubric-review separation is pinned by tests/mvp/harness/test_real_port_lane_evidence.py.

When to use:

Wrapping an agent pipeline with structured failure detection at each step
Auditing a multi-step workflow against a library of known failure patterns
Adding domain-specific checklist gates for healthcare, legal, accounting, education, or custom workflows
Enforcing quality gates before committing intermediate outputs
Discovering Capability Harness strata, applied patterns, and skill metadata through the read-only harness_mcp surface

Example:

from mvp.harness import WorkflowHarness, WorkflowRubric, generate_workflow_rubric

rubric = generate_workflow_rubric(
    workflow_id="pipeline_v1",
    steps=["parse", "solve"],
    domain="legal",
    objective="contract risk",
)
rubric.save("pipeline_v1_rubric.yaml")  # edit this file, then load it later
rubric = WorkflowRubric.load("pipeline_v1_rubric.yaml")
harness = WorkflowHarness(
    workflow_id="pipeline_v1",
    severity_threshold=0.6,
    rubric=rubric,
)
result = harness.run({
    "parse": lambda: parse_input(raw),
    "solve": lambda: solve(),
})
# result.overall_passed, result.pass_rate, result.failed_steps

Production caveat

harness is a lightweight verification harness, not a complete evaluation platform or a guarantee of correctness. Built-in failure modes and generated rubrics are starter quality gates. For production workflows, especially healthcare, legal, accounting, regulated, or customer-facing decisions, treat generated rubrics as scaffolding only: review them with a domain expert, encode the actual acceptance criteria for the workflow, validate against held-out examples, and route high-stakes or uncertain outputs to human review or stronger verification. Passing regulated Capability Harness scenarios require expert_review_status == "approved" and a non-empty reviewer_id; otherwise the gate blocks release with blocked-escalated semantics.

Works well with: benchmark_runner, align_evals, solver

Public API¶

`EvalRunner`¶

Gathers per-scenario evidence from injected ports.

Constructor:

Parameter	Type	Default
`component_under_test`	`str`	'harness_demo'
`baseline_path`	`str`	_DEFAULT_BASELINE_PATH

Methods:

`evaluate(scenario: Scenario, ports: dict[str, Any]) -> dict[str, Any]`¶

Run scenario against injected ports and return evidence.

`FailureCategory(str, Enum)`¶

`FailureMode`¶

Field	Type	Default
`category`	`FailureCategory`	required
`description`	`str`	required
`trigger_keywords`	`List[str]`	required
`severity`	`float`	required

`FailureModeLibrary`¶

Detect known failure modes in text output.

Constructor:

Parameter	Type	Default
`modes`	`Optional[List[FailureMode]]`	None

Methods:

`detect(text: str) -> List[FailureMode]`¶

Return all failure modes whose trigger keywords appear in text.

`by_category(category: FailureCategory) -> FailureMode`¶

Return the configured mode for category.

`max_severity(text: str) -> float`¶

Return highest severity among detected modes, or 0.0 if none.

`Gate`¶

Release-gate aggregator.

Methods:

`evaluate(scenario_results: list[ScenarioResult], strata: list[Stratum], run_id: str = '', git_sha: str = '', evidence_bundle_path: str = '', control_hash_verified: bool = False, port_mode: str | None = None) -> GateResult`¶

Produce a GateResult.

`HarnessStepResult`¶

Field	Type	Default
`step_name`	`str`	required
`output`	`Any`	required
`verification`	`VerificationResult`	required
`passed`	`bool`	required

`HarnessResult`¶

Field	Type	Default
`workflow_id`	`str`	required
`steps`	`List[HarnessStepResult]`	field(default_factory=list)
`overall_passed`	`bool`	True

Methods:

`failed_steps() -> List[HarnessStepResult]`¶

`pass_rate() -> float`¶

`WorkflowHarness`¶

Run workflow steps and verify each against the failure mode library.

Constructor:

Parameter	Type	Default
`workflow_id`	`str`	required
`severity_threshold`	`float`	0.6
`library`	`Optional[FailureModeLibrary]`	None
`rubric`	`Optional[WorkflowRubric]`	None

Methods:

`run_step(step_name: str, fn: Callable[[], Any]) -> HarnessStepResult`¶

Execute fn(), capture output, verify it.

`run(steps: Dict[str, Callable[[], Any]]) -> HarnessResult`¶

Run all steps in order, collecting results.

`run_with_injection(steps: Dict[str, Callable[[], Any]], inject: Dict[str, 'FailureCategory'], seed: Optional[int] = None) -> HarnessResult`¶

Run steps but swap the listed step names for injected

`HarnessInput(BaseModel)`¶

Field	Type	Default
`op`	`str`	required
`parameters`	`dict[str, Any]`	Field(default_factory=dict)

`HarnessOutput(BaseModel)`¶

Field	Type	Default
`op`	`str`	''
`result`	`dict[str, Any]`	Field(default_factory=dict)
`message`	`str`	''
`completion_state`	`str`	'qualified-draft'
`warning_card`	`dict[str, Any] \\| None`	None
`evidence`	`dict[str, Any]`	Field(default_factory=dict)
`request_id`	`str`	''
`task_id`	`str`	''
`run_id`	`str`	''
`degraded`	`bool`	False
`degradation_reason`	`str`	''

`HarnessBlock(AIBlock)`¶

AIBlock wrapper for test harness workflow execution and verification.

Methods:

`infer(input: HarnessInput) -> Result[HarnessOutput]`¶

`FailureInjector`¶

Produce strings that match a chosen FailureMode signature.

Field	Type	Default
`library`	`List[FailureMode]`	field(default_factory=lambda: list(FAILURE_MODE_LIBRARY))
`seed`	`Optional[int]`	None

Methods:

`mode_for(category: FailureCategory) -> FailureMode`¶

`inject(category: FailureCategory, context: str = '') -> str`¶

Return a string whose content matches category's signature.

`inject_callable(category: FailureCategory, context: str = '') -> Callable[[], str]`¶

Convenience: return a Callable[[], str] shaped for WorkflowHarness.run_step.

`exercise_all(context: str = '') -> dict[str, str]`¶

Produce one injected output per category the library knows about.

`HarnessOrchestrator`¶

Run a Scenario through EVAL -> GAP -> (REMEDIATE) -> GATE.

Constructor:

Parameter	Type	Default
`eval_runner`	`EvalRunner \\| None`	None
`remediator`	`Remediator \\| None`	None
`max_attempts`	`int`	1
`_clock`	`Callable[[], float] \\| None`	None
`allow_t3`	`bool`	False

Methods:

`run(scenario: Scenario, ports: dict[str, Any], tier_cap: RemediationTier = RemediationTier.RECIPE) -> ScenarioResult`¶

`Remediator`¶

Runs a single remediation attempt at the specified tier.

Constructor:

Parameter	Type	Default
`allow_t3`	`bool`	False

Methods:

`remediate(gap: GapReport, ports: dict[str, Any], tier_cap: RemediationTier = RemediationTier.RECIPE, scenario: Scenario | None = None) -> RemediationAttempt`¶

Run Tier 1 (RECIPE) and return the attempt.

`RubricCheckResult`¶

Field	Type	Default
`name`	`str`	required
`passed`	`bool`	required
`reason`	`str`	required

`ChecklistItem`¶

A user-editable check for one workflow step.

Field	Type	Default
`name`	`str`	required
`description`	`str`	''
`required_terms`	`List[str]`	field(default_factory=list)
`forbidden_terms`	`List[str]`	field(default_factory=list)
`min_length`	`Optional[int]`	None
`validator`	`Optional[Callable[[Any], bool]]`	None

Methods:

`to_dict() -> Dict[str, Any]`¶

`from_dict(data: Dict[str, Any]) -> 'ChecklistItem'`¶

`evaluate(output: Any) -> RubricCheckResult`¶

`RubricResult`¶

Field	Type	Default
`passed`	`bool`	required
`checks`	`List[RubricCheckResult]`	required

Methods:

`failed_checks() -> List[RubricCheckResult]`¶

`WorkflowRubric`¶

Domain-specific checklist collection keyed by workflow step name.

Field	Type	Default
`workflow_id`	`str`	required
`step_checks`	`Dict[str, List[ChecklistItem]]`	field(default_factory=dict)

Methods:

`add_check(step_name: str, item: ChecklistItem) -> None`¶

`extend(step_name: str, items: Iterable[ChecklistItem]) -> None`¶

`evaluate(step_name: str, output: Any) -> RubricResult`¶

`from_dict(workflow_id: str, spec: Dict[str, Iterable[Dict[str, Any]]]) -> 'WorkflowRubric'`¶

`to_file_dict() -> Dict[str, Any]`¶

`from_file_dict(data: Dict[str, Any]) -> 'WorkflowRubric'`¶

`save(path: str | Path) -> None`¶

`load(path: str | Path) -> 'WorkflowRubric'`¶

`Stratum(str, Enum)`¶

The four reliability strata bars are scoped against.

`RemediationTier(str, Enum)`¶

Escalation ladder tiers used by the Remediator.

`AuthorityRef`¶

Reference to an external authoritative source (e.g., AHPRA, FDA).

Field	Type	Default
`authority`	`str`	required
`doc_id`	`str`	required
`citation_url`	`str`	required
`version`	`str`	required
`retrieved_at`	`float`	required

`Bar`¶

A single per-stratum acceptance bar.

Field	Type	Default
`stratum`	`Stratum`	required
`metric_name`	`str`	required
`threshold`	`float`	required
`comparator`	`Literal['ge', 'le']`	required
`window`	`Literal['per_run', 'rolling_7']`	required

`RawLLMControl`¶

Captured evidence of how a raw (unaided) LLM fails the scenario.

Field	Type	Default
`model`	`str`	required
`transcript_hash`	`str`	required
`failure_evidence`	`str`	required
`captured_at`	`float`	required

`Scenario`¶

A single capability scenario the harness evaluates against.

Field	Type	Default
`id`	`str`	required
`slug`	`str`	required
`vertical`	`str`	required
`stratum`	`Stratum`	required
`user_prompt`	`str`	required
`raw_llm_control`	`RawLLMControl`	required
`target_artifacts`	`tuple[str, ...]`	required
`authorities`	`tuple[AuthorityRef, ...]`	required
`rubric_ref`	`str`	required
`token_budget`	`int`	required
`wallclock_budget_s`	`float`	required
`friction_floor`	`float`	0.75
`expert_review_status`	`Literal['not_reviewed', 'pending', 'approved', 'rejected']`	'not_reviewed'
`reviewer_id`	`str`	''

`GapReport`¶

Result of detect_gaps for a scenario that failed at least one bar.

Field	Type	Default
`scenario_id`	`str`	required
`stratum`	`Stratum`	required
`failed_bars`	`tuple[Bar, ...]`	required
`failure_mode_ids`	`tuple[str, ...]`	required
`missing_authorities`	`tuple[AuthorityRef, ...]`	required
`reliability_score`	`float`	required
`theory_gap`	`bool`	required
`exemplar_gap`	`bool`	required
`recipe_gap`	`bool`	required

`RemediationAttempt`¶

Outcome of a single remediation attempt at a tier.

Field	Type	Default
`tier`	`RemediationTier`	required
`started_at`	`float`	required
`ended_at`	`float`	required
`tokens_in`	`int`	required
`tokens_out`	`int`	required
`wallclock_s`	`float`	required
`artifacts_written`	`tuple[str, ...]`	required
`succeeded`	`bool`	required
`escalation_reason`	`str`	required

`ScenarioResult`¶

Final orchestrator output for one scenario.

Field	Type	Default
`scenario_id`	`str`	required
`passed`	`bool`	required
`stratum`	`Stratum`	required
`gap_report`	`GapReport \\| None`	required
`remediation_attempts`	`tuple[RemediationAttempt, ...]`	required
`wallclock_s`	`float`	required
`tokens_total`	`int`	required
`friction_report`	`'FrictionReport \\| None'`	None
`expert_review_status`	`Literal['not_reviewed', 'pending', 'approved', 'rejected']`	'not_reviewed'
`reviewer_id`	`str`	''
`port_mode`	`Literal['fake', 'mixed', 'real', 'unknown']`	'unknown'

`GateResult`¶

Release-gate aggregate across all scenarios in a run.

Field	Type	Default
`run_id`	`str`	required
`git_sha`	`str`	required
`scenario_results`	`tuple[ScenarioResult, ...]`	required
`strata_status`	`dict[str, str]`	required
`release_blocking`	`bool`	required
`evidence_bundle_path`	`str`	required
`control_hash_verified`	`bool`	False
`port_mode`	`str`	'unknown'
`release_reasons`	`tuple[str, ...]`	()

`VerificationResult`¶

Field	Type	Default
`passed`	`bool`	required
`failures_detected`	`List[FailureMode]`	required
`max_severity`	`float`	required
`reason`	`str`	required
`rubric`	`Optional[RubricResult]`	None

`StepVerifier`¶

Verify a single workflow step output against the failure mode library.

Constructor:

Parameter	Type	Default
`severity_threshold`	`float`	0.6
`library`	`Optional[FailureModeLibrary]`	None

Methods:

`verify(output: Any) -> VerificationResult`¶

`verify_step(step_name: str, output: Any, rubric: Optional[WorkflowRubric] = None) -> VerificationResult`¶

Functions¶

`bars_for(stratum: Stratum, component: str | None = None) -> list[Bar]`¶

Return the list of bars active for stratum in this phase.

`verify_control_hash(control: RawLLMControl, control_md_path: Path) -> bool`¶

Hash the on-disk control file and compare with control.transcript_hash.

`detect_gaps(scenario: Scenario, bars: list[Bar], evidence: dict[str, Any]) -> GapReport | None`¶

Return a GapReport when any bar fails, otherwise None.

`generate_workflow_rubric(workflow_id: str, steps: Iterable[str], domain: str = 'general', objective: str = '') -> WorkflowRubric`¶

Generate a practical starter checklist users can edit.

`gate_result_to_dict(result: GateResult) -> dict[str, Any]`¶

Return a JSON-serialisable dict for result.

Harness¶

Overview¶

Public API¶

EvalRunner¶

evaluate(scenario: Scenario, ports: dict[str, Any]) -> dict[str, Any]¶

FailureCategory(str, Enum)¶

FailureMode¶

FailureModeLibrary¶

detect(text: str) -> List[FailureMode]¶

by_category(category: FailureCategory) -> FailureMode¶

max_severity(text: str) -> float¶

Gate¶

evaluate(scenario_results: list[ScenarioResult], strata: list[Stratum], run_id: str = '', git_sha: str = '', evidence_bundle_path: str = '', control_hash_verified: bool = False, port_mode: str | None = None) -> GateResult¶

HarnessStepResult¶

HarnessResult¶

failed_steps() -> List[HarnessStepResult]¶

pass_rate() -> float¶

WorkflowHarness¶

run_step(step_name: str, fn: Callable[[], Any]) -> HarnessStepResult¶

run(steps: Dict[str, Callable[[], Any]]) -> HarnessResult¶

run_with_injection(steps: Dict[str, Callable[[], Any]], inject: Dict[str, 'FailureCategory'], seed: Optional[int] = None) -> HarnessResult¶

HarnessInput(BaseModel)¶

HarnessOutput(BaseModel)¶

HarnessBlock(AIBlock)¶

infer(input: HarnessInput) -> Result[HarnessOutput]¶

FailureInjector¶

mode_for(category: FailureCategory) -> FailureMode¶

inject(category: FailureCategory, context: str = '') -> str¶

inject_callable(category: FailureCategory, context: str = '') -> Callable[[], str]¶

exercise_all(context: str = '') -> dict[str, str]¶

HarnessOrchestrator¶

run(scenario: Scenario, ports: dict[str, Any], tier_cap: RemediationTier = RemediationTier.RECIPE) -> ScenarioResult¶

Remediator¶

remediate(gap: GapReport, ports: dict[str, Any], tier_cap: RemediationTier = RemediationTier.RECIPE, scenario: Scenario | None = None) -> RemediationAttempt¶

RubricCheckResult¶

ChecklistItem¶

to_dict() -> Dict[str, Any]¶

from_dict(data: Dict[str, Any]) -> 'ChecklistItem'¶

evaluate(output: Any) -> RubricCheckResult¶

RubricResult¶

failed_checks() -> List[RubricCheckResult]¶

WorkflowRubric¶

add_check(step_name: str, item: ChecklistItem) -> None¶

extend(step_name: str, items: Iterable[ChecklistItem]) -> None¶

evaluate(step_name: str, output: Any) -> RubricResult¶

from_dict(workflow_id: str, spec: Dict[str, Iterable[Dict[str, Any]]]) -> 'WorkflowRubric'¶

to_file_dict() -> Dict[str, Any]¶

from_file_dict(data: Dict[str, Any]) -> 'WorkflowRubric'¶

save(path: str | Path) -> None¶

load(path: str | Path) -> 'WorkflowRubric'¶

Stratum(str, Enum)¶

RemediationTier(str, Enum)¶

AuthorityRef¶

Bar¶

RawLLMControl¶

Scenario¶

GapReport¶

RemediationAttempt¶

ScenarioResult¶

GateResult¶

VerificationResult¶

StepVerifier¶

verify(output: Any) -> VerificationResult¶

verify_step(step_name: str, output: Any, rubric: Optional[WorkflowRubric] = None) -> VerificationResult¶

Functions¶

bars_for(stratum: Stratum, component: str | None = None) -> list[Bar]¶

verify_control_hash(control: RawLLMControl, control_md_path: Path) -> bool¶

detect_gaps(scenario: Scenario, bars: list[Bar], evidence: dict[str, Any]) -> GapReport | None¶

generate_workflow_rubric(workflow_id: str, steps: Iterable[str], domain: str = 'general', objective: str = '') -> WorkflowRubric¶

gate_result_to_dict(result: GateResult) -> dict[str, Any]¶

`EvalRunner`¶

`evaluate(scenario: Scenario, ports: dict[str, Any]) -> dict[str, Any]`¶

`FailureCategory(str, Enum)`¶

`FailureMode`¶

`FailureModeLibrary`¶

`detect(text: str) -> List[FailureMode]`¶

`by_category(category: FailureCategory) -> FailureMode`¶

`max_severity(text: str) -> float`¶

`Gate`¶

`evaluate(scenario_results: list[ScenarioResult], strata: list[Stratum], run_id: str = '', git_sha: str = '', evidence_bundle_path: str = '', control_hash_verified: bool = False, port_mode: str | None = None) -> GateResult`¶

`HarnessStepResult`¶

`HarnessResult`¶

`failed_steps() -> List[HarnessStepResult]`¶

`pass_rate() -> float`¶

`WorkflowHarness`¶

`run_step(step_name: str, fn: Callable[[], Any]) -> HarnessStepResult`¶

`run(steps: Dict[str, Callable[[], Any]]) -> HarnessResult`¶

`run_with_injection(steps: Dict[str, Callable[[], Any]], inject: Dict[str, 'FailureCategory'], seed: Optional[int] = None) -> HarnessResult`¶

`HarnessInput(BaseModel)`¶

`HarnessOutput(BaseModel)`¶

`HarnessBlock(AIBlock)`¶

`infer(input: HarnessInput) -> Result[HarnessOutput]`¶

`FailureInjector`¶

`mode_for(category: FailureCategory) -> FailureMode`¶

`inject(category: FailureCategory, context: str = '') -> str`¶

`inject_callable(category: FailureCategory, context: str = '') -> Callable[[], str]`¶

`exercise_all(context: str = '') -> dict[str, str]`¶

`HarnessOrchestrator`¶

`run(scenario: Scenario, ports: dict[str, Any], tier_cap: RemediationTier = RemediationTier.RECIPE) -> ScenarioResult`¶

`Remediator`¶

`remediate(gap: GapReport, ports: dict[str, Any], tier_cap: RemediationTier = RemediationTier.RECIPE, scenario: Scenario | None = None) -> RemediationAttempt`¶

`RubricCheckResult`¶

`ChecklistItem`¶

`to_dict() -> Dict[str, Any]`¶

`from_dict(data: Dict[str, Any]) -> 'ChecklistItem'`¶

`evaluate(output: Any) -> RubricCheckResult`¶

`RubricResult`¶

`failed_checks() -> List[RubricCheckResult]`¶

`WorkflowRubric`¶

`add_check(step_name: str, item: ChecklistItem) -> None`¶

`extend(step_name: str, items: Iterable[ChecklistItem]) -> None`¶

`evaluate(step_name: str, output: Any) -> RubricResult`¶

`from_dict(workflow_id: str, spec: Dict[str, Iterable[Dict[str, Any]]]) -> 'WorkflowRubric'`¶

`to_file_dict() -> Dict[str, Any]`¶

`from_file_dict(data: Dict[str, Any]) -> 'WorkflowRubric'`¶

`save(path: str | Path) -> None`¶

`load(path: str | Path) -> 'WorkflowRubric'`¶

`Stratum(str, Enum)`¶

`RemediationTier(str, Enum)`¶

`AuthorityRef`¶

`Bar`¶

`RawLLMControl`¶

`Scenario`¶

`GapReport`¶

`RemediationAttempt`¶

`ScenarioResult`¶

`GateResult`¶

`VerificationResult`¶

`StepVerifier`¶

`verify(output: Any) -> VerificationResult`¶

`verify_step(step_name: str, output: Any, rubric: Optional[WorkflowRubric] = None) -> VerificationResult`¶

`bars_for(stratum: Stratum, component: str | None = None) -> list[Bar]`¶

`verify_control_hash(control: RawLLMControl, control_md_path: Path) -> bool`¶

`detect_gaps(scenario: Scenario, bars: list[Bar], evidence: dict[str, Any]) -> GapReport | None`¶

`generate_workflow_rubric(workflow_id: str, steps: Iterable[str], domain: str = 'general', objective: str = '') -> WorkflowRubric`¶

`gate_result_to_dict(result: GateResult) -> dict[str, Any]`¶