Reliability Pipeline¶

reliability_pipeline — G6's canonical reliability solver.

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

Overview¶

Public API¶

`ReliabilityRecipe`¶

Field	Type	Default
`problem`	`str`	required
`domain`	`str`	required
`risk_tier`	`str`	required
`corpus_agents`	`list[str]`	required
`steps`	`list[RecipeStep]`	required
`recommended_patterns`	`list[RecommendedPattern]`	required
`standards`	`list[StandardRef]`	required
`templates`	`list[TemplateRef]`	required
`exemplars`	`list[RetrievedExemplar]`	required
`library_cards`	`list[LibraryCard]`	required
`evidence`	`RecipeEvidence`	required
`expected_completion_state`	`str`	required
`escalation_guidance`	`str`	required
`degraded`	`bool`	required
`llm_used`	`bool`	required
`notes`	`list[str]`	required

`LLMError(RuntimeError)`¶

An LLM call failed — the calling phase must degrade honestly.

`LLMCaller(Protocol)`¶

Minimal LLM interface: one system+user prompt -> one completion.

Methods:

`complete(system: str, user: str, max_tokens: int = DEFAULT_MAX_TOKENS) -> str`¶

`LLMBlockCaller`¶

An :class:LLMCaller backed by mvp.llm_router.LLMBlock.

Constructor:

Parameter	Type	Default
`model`	`str`	'auto'
`name`	`str`	'reliability_pipeline'

Methods:

`complete(system: str, user: str, max_tokens: int = DEFAULT_MAX_TOKENS) -> str`¶

`FakeLLM`¶

A deterministic :class:LLMCaller for tests. Records every call.

Constructor:

Parameter	Type	Default
`responder`	`Callable[[str, str], str] \\| None`	None
`default`	`str`	'OK'

Methods:

`complete(system: str, user: str, max_tokens: int = DEFAULT_MAX_TOKENS) -> str`¶

`TimingLLMCaller`¶

An :class:LLMCaller that times every call for the live ETA.

Constructor:

Parameter	Type	Default
`inner`	`LLMCaller`	required
`on_latency`	`Callable[[float], None]`	required

Methods:

`complete(system: str, user: str, max_tokens: int = DEFAULT_MAX_TOKENS) -> str`¶

`ReliabilityPipelineBlock(AIBlock[TaskRequest, ReliabilityResult, None])`¶

G6's canonical reliability solver, as an :class:AIBlock.

Constructor:

Parameter	Type	Default
`llm`	`LLMCaller \\| None`	None
`name`	`str`	'reliability_pipeline'
`progress_sinks`	`list[Callable[[ProgressEvent], None]] \\| None`	None
`grounding`	`'GroundingCallable \\| None \\| _UseDefaultGrounding'`	USE_DEFAULT_GROUNDING
`value_completion`	`'bool \\| _UseDefaultValueCompletion'`	USE_DEFAULT_VALUE_COMPLETION
`alignment`	`'bool \\| _UseDefaultAlignment'`	USE_DEFAULT_ALIGNMENT
`clarification`	`'bool \\| _UseDefaultClarification'`	USE_DEFAULT_CLARIFICATION
`presolve_evidence`	`'bool \\| _UseDefaultPresolveEvidence'`	USE_DEFAULT_PRESOLVE_EVIDENCE

Methods:

`infer(data: TaskRequest | dict) -> Result[ReliabilityResult]`¶

Run the pipeline. Returns Result.ok with an honest

`envelope_from_result(result: ReliabilityResult, run_id: str = '', recipe_id: str = '', recipe_title: str = '', workflow_id: str = '', execution_mode: str = 'full', inputs: dict | None = None) -> 'AgenticWorkflowRun'`¶

RS-11-F01 + RS-11-F04: build the cross-surface run envelope from a

`PremortemFinding`¶

Field	Type	Default
`failure_mode`	`str`	required
`diagnostic_question`	`str`	required
`guardrail`	`str`	required
`likely`	`bool`	required

`ProgressEvent`¶

One honest snapshot of an in-flight run, handed to every sink.

Field	Type	Default
`phase`	`str`	required
`phase_index`	`int`	required
`phase_total`	`int`	4
`stage_label`	`str`	''
`calls_done`	`int`	0
`calls_total`	`int`	0
`elapsed_seconds`	`float`	0.0
`eta_seconds`	`float`	0.0
`fraction`	`float`	0.0
`calibrating`	`bool`	True
`message`	`str`	''
`completion_state`	`str`	''

`EtaEstimator`¶

Projects elapsed time, ETA and percentage from the call-count model.

Field	Type	Default
`model`	`str`	'auto'
`calls_total`	`int`	0
`calls_done`	`int`	0
`clock`	`Callable[[], float]`	time.monotonic

Methods:

`record_call(latency_seconds: float) -> None`¶

Record one completed LLM call and its wall-clock latency.

`revise_total(calls_total: int) -> None`¶

Replace the planned total once the real plan is known.

`calibrating() -> bool`¶

True while no real latency has been observed (ETA uses the prior).

`avg_latency() -> float`¶

Seconds per call — observed running mean, else the model prior.

`p50_latency() -> float`¶

Median observed seconds-per-call (0.0 until a call is timed).

`p95_latency() -> float`¶

95th-percentile observed seconds-per-call (0.0 until a call is timed).

`elapsed_seconds() -> float`¶

`eta_seconds() -> float`¶

Estimated seconds remaining — never negative.

`fraction() -> float`¶

Completion fraction in [0.0, 1.0].

`ProgressReporter`¶

Drives an :class:EtaEstimator and fans :class:ProgressEvents to sinks.

Constructor:

Parameter	Type	Default
`estimator`	`EtaEstimator \\| None`	None
`sinks`	`list[Callable[[ProgressEvent], None]] \\| None`	None

Methods:

`add_sink(sink: Callable[[ProgressEvent], None]) -> None`¶

`phase(name: str, index: int) -> None`¶

Mark the start of a pipeline phase (1=assessment .. 4=completion).

`plan(goal_stack_size: int, n_constraints: int) -> None`¶

Firm up the call-count plan once assessment knows the real shape.

`stage(label: str) -> None`¶

Set the within-phase stage label (e.g. 'goal 2 of 4').

`call_recorded(latency_seconds: float) -> None`¶

One LLM call completed — advance the estimate and emit.

`finish(completion_state: str) -> None`¶

Emit the terminal event for a completed run.

`GroundingSource`¶

A user-provided document that the grounding step may cite against.

Field	Type	Default
`label`	`str`	required
`text`	`str`	required
`origin`	`str`	required
`path`	`str`	''

`TaskRequest`¶

A task posed to the reliability pipeline.

Field	Type	Default
`goal`	`str`	required
`context`	`str`	''
`constraints`	`list[str]`	field(default_factory=list)
`acceptance_criteria`	`list[str]`	field(default_factory=lambda: ['deliverable answers the requested goal'])
`task_mode`	`str`	''
`success_criteria_structured`	`tuple[dict, ...]`	()
`deadline`	`str`	''
`max_tokens`	`int`	60000
`timeout_seconds`	`float`	0.0
`max_cost_usd`	`float`	0.0
`risk_hint`	`str`	''
`context_sources`	`list[GroundingSource]`	field(default_factory=list)
`decomposition_breadth`	`int`	0
`intent_answers`	`dict[str, object]`	field(default_factory=dict)
`declined_clarification`	`bool`	False
`accepted_risky_intent`	`bool`	False
`plan_steps`	`list[str]`	field(default_factory=list)
`touched_paths`	`list[str]`	field(default_factory=list)
`patch_diff`	`str`	''
`failing_tests`	`list[str]`	field(default_factory=list)
`relevant_source_files`	`list[str]`	field(default_factory=list)
`expected_behavior`	`str`	''
`failure_modes`	`list[str]`	field(default_factory=list)
`codex_approval_override`	`bool`	False
`self_correction_enabled`	`bool`	False
`self_correction_max_cycles`	`int`	1
`avoided_components`	`tuple[str, ...]`	()
`preferred_components`	`tuple[str, ...]`	()
`learning_tier`	`int`	-1
`max_paid_calls`	`int`	0
`max_paid_tokens`	`int`	0

`GoalNode`¶

One node in the assessment's goal tree.

Field	Type	Default
`id`	`str`	required
`goal`	`str`	required
`children`	`list['GoalNode']`	field(default_factory=list)
`depth`	`int`	0

`AssessmentResult`¶

Phase 1 output — the task understood, risk-tiered, and decomposed.

Field	Type	Default
`task_contract`	`TaskContract`	required
`risk_tier`	`RiskTier`	required
`epistemic_contract`	`EpistemicContract`	required
`premortem`	`list[PremortemFinding]`	required
`first_principles`	`list[str]`	required
`heuristics`	`list[str]`	required
`self_assessment`	`dict`	required
`satisficing_strategy`	`str`	required
`goal_tree`	`GoalNode`	required
`goal_stack`	`list[str]`	required

`ExecutionStep`¶

One goal solved during execution — initial draft, critique, revision.

Field	Type	Default
`goal`	`str`	required
`initial_output`	`str`	required
`critique`	`str`	required
`revised_output`	`str`	required
`tokens_used`	`int`	0

`ExecutionResult`¶

Phase 2 output — the goal solved (or honestly stopped on budget).

Field	Type	Default
`steps`	`list[ExecutionStep]`	required
`final_output`	`str`	required
`tokens_used`	`int`	required
`budget_exceeded`	`bool`	False
`assumptions_ledger`	`list[str]`	field(default_factory=list)
`decision_log`	`list[str]`	field(default_factory=list)
`source_list`	`list[str]`	field(default_factory=list)
`open_questions`	`list[str]`	field(default_factory=list)
`touched_components`	`list[str]`	field(default_factory=list)
`safety_flagged`	`bool`	False
`safety_findings`	`list[str]`	field(default_factory=list)
`stop_reason`	`str`	''
`evidence_source_count`	`int`	0
`checkpoints`	`list`	field(default_factory=list)

`EvaluationResult`¶

Phase 3 output — independent verification evidence.

Field	Type	Default
`verification_report`	`VerificationReport`	required
`constraint_critiques`	`dict`	required
`claims_ledger`	`list[ClaimsLedgerEntry]`	required
`final_critique`	`str`	required

`StageRecord`¶

One explicit state-machine transition emitted by the pipeline.

Field	Type	Default
`stage`	`str`	required
`state`	`str`	required
`reason`	`str`	''
`safe_next_action`	`str`	''
`metadata`	`dict`	field(default_factory=dict)

`HandoffPackage`¶

Human-usable summary of what the run produced and how to rely on it.

Field	Type	Default
`completion_state`	`str`	required
`reliability_status`	`str`	required
`summary`	`str`	required
`main_risks`	`list[str]`	field(default_factory=list)
`verified`	`list[str]`	field(default_factory=list)
`not_verified`	`list[str]`	field(default_factory=list)
`recommended_next_step`	`str`	''
`warning_card`	`dict`	field(default_factory=dict)

`RunMetrics`¶

What a run cost — recorded for the viewer (time/quality/cost trade-off).

Field	Type	Default
`elapsed_seconds`	`float`	0.0
`llm_calls`	`int`	0
`estimated_tokens`	`int`	0
`estimated_cost_usd`	`float`	0.0
`grounding_calls`	`int`	0
`grounding_backend`	`str`	''
`paid_llm_calls`	`int`	0
`billed_cents`	`int`	0
`budget_exhausted`	`bool`	False
`failed_attempts`	`int`	0
`duplicate_calls_avoided`	`int`	0
`p50_latency_ms`	`float`	0.0
`p95_latency_ms`	`float`	0.0
`avg_tokens_per_call`	`float`	0.0
`bound_hit`	`str`	''
`remaining_budget`	`dict`	field(default_factory=dict)
`cost_cap_basis`	`str`	''

`EvidenceRef`¶

A single source citation attached to a pipeline output.

Field	Type	Default
`source_id`	`str`	required
`label`	`str`	required
`score`	`float`	required
`preview`	`str`	required
`phase`	`str`	''
`section`	`str`	''
`url`	`str`	''
`chunk_id`	`str`	''
`origin`	`str`	''

`RecoveryCycle`¶

One bounded self-correction cycle result.

Field	Type	Default
`cycle`	`int`	required
`checks_failed`	`int`	required
`unverified_items`	`int`	required
`total_open`	`int`	required

`RecoveryMetadata`¶

Run-level self-correction metadata persisted with results.

Field	Type	Default
`enabled`	`bool`	False
`max_cycles`	`int`	0
`cycles_attempted`	`int`	0
`stop_reason`	`str`	'disabled'
`initial_checks_failed`	`int`	0
`initial_unverified_items`	`int`	0
`initial_total_open`	`int`	0
`cycles`	`list[RecoveryCycle]`	field(default_factory=list)

`ReliabilityResult`¶

The pipeline's honest, user-facing output.

Field	Type	Default
`completion_state`	`CompletionState`	required
`deliverable`	`str`	required
`warning_card`	`WarningCard`	required
`assessment`	`AssessmentResult \\| None`	None
`execution`	`ExecutionResult \\| None`	None
`evaluation`	`EvaluationResult \\| None`	None
`residual_risks`	`list[str]`	field(default_factory=list)
`how_to_verify`	`list[str]`	field(default_factory=list)
`run_metrics`	`RunMetrics \\| None`	None
`evidence_refs`	`list[EvidenceRef]`	field(default_factory=list)
`number_provenance`	`list[ProvenancedNumber]`	field(default_factory=list)
`content_provenance`	`list[ContentProvenance]`	field(default_factory=list)
`translation_fidelity`	`'TranslationFidelity \\| None'`	None
`recovery`	`RecoveryMetadata`	field(default_factory=RecoveryMetadata)
`stage_trace`	`list[StageRecord]`	field(default_factory=list)
`handoff_package`	`HandoffPackage \\| None`	None
`weakest_link`	`dict`	field(default_factory=dict)
`alignment`	`dict`	field(default_factory=dict)
`effective_policy_echo`	`dict`	field(default_factory=dict)
`self_confidence`	`dict`	field(default_factory=dict)
`refutation_evidence`	`dict`	field(default_factory=dict)
`escalation_decision`	`dict`	field(default_factory=dict)
`escalation_ticket`	`dict`	field(default_factory=dict)
`method_ir`	`dict`	field(default_factory=dict)

Functions¶

`run_assessment(request: TaskRequest, llm: LLMCaller, progress: ProgressReporter = NULL_REPORTER) -> AssessmentResult`¶

Run Phase 1 — assess, risk-tier, and decompose the task.

`run_completion(assessment: AssessmentResult, execution: ExecutionResult, evaluation: EvaluationResult, progress: ProgressReporter = NULL_REPORTER) -> ReliabilityResult`¶

Run Phase 4 — assemble the pipeline's honest, user-facing result.

`blocked_result(reason: str, assessment: AssessmentResult | None = None, execution: ExecutionResult | None = None, evaluation: EvaluationResult | None = None, deliverable: str = '', guidance: dict | None = None) -> ReliabilityResult`¶

A run that could not complete — used when a phase fails outright.

`run_evaluation(assessment: AssessmentResult, execution: ExecutionResult, llm: LLMCaller, progress: ProgressReporter = NULL_REPORTER, oracle_evidence: Mapping[str, object] | None = None) -> EvaluationResult`¶

Run Phase 3 — independently verify the deliverable.

`run_execution(assessment: AssessmentResult, llm: LLMCaller, max_tokens: int = 60000, max_execution_seconds: float | None = None, progress: ProgressReporter = NULL_REPORTER, context_packet_summary: dict | None = None, safety_policy: RuntimePolicy | None = None, evidence_block: str = '', evidence_source_count: int = 0, checkpoint_mode: str = 'off', contract: Any = None, method_proposal: Any = None, evidence_sources: list | None = None, max_corrections: int = 1) -> ExecutionResult`¶

Run Phase 2 — solve the goal stack bottom-up with critique -> revise.

`build_reliability_recipe(problem: str, domain: str, risk_tier: str | None = None, corpus_agents: list[str] | None = None, top_k: int = 5, log_to_ledger: bool = True) -> Result[ReliabilityRecipe]`¶

Build an advisory, corpus-grounded reliability recipe.

`estimate_cost(model: str, total_tokens: int) -> float`¶

Best-effort run cost in USD.

`prometheus_progress_sink(event: ProgressEvent) -> None`¶

A best-effort Prometheus sink for :class:ProgressReporter to attach.

Reliability Pipeline¶

Overview¶

Public API¶

ReliabilityRecipe¶

LLMError(RuntimeError)¶

LLMCaller(Protocol)¶

complete(system: str, user: str, max_tokens: int = DEFAULT_MAX_TOKENS) -> str¶

LLMBlockCaller¶

complete(system: str, user: str, max_tokens: int = DEFAULT_MAX_TOKENS) -> str¶

FakeLLM¶

complete(system: str, user: str, max_tokens: int = DEFAULT_MAX_TOKENS) -> str¶

TimingLLMCaller¶

complete(system: str, user: str, max_tokens: int = DEFAULT_MAX_TOKENS) -> str¶

ReliabilityPipelineBlock(AIBlock[TaskRequest, ReliabilityResult, None])¶

infer(data: TaskRequest | dict) -> Result[ReliabilityResult]¶

envelope_from_result(result: ReliabilityResult, run_id: str = '', recipe_id: str = '', recipe_title: str = '', workflow_id: str = '', execution_mode: str = 'full', inputs: dict | None = None) -> 'AgenticWorkflowRun'¶

PremortemFinding¶

ProgressEvent¶

EtaEstimator¶

record_call(latency_seconds: float) -> None¶

revise_total(calls_total: int) -> None¶

calibrating() -> bool¶

avg_latency() -> float¶

p50_latency() -> float¶

p95_latency() -> float¶

elapsed_seconds() -> float¶

eta_seconds() -> float¶

fraction() -> float¶

ProgressReporter¶

add_sink(sink: Callable[[ProgressEvent], None]) -> None¶

phase(name: str, index: int) -> None¶

plan(goal_stack_size: int, n_constraints: int) -> None¶

stage(label: str) -> None¶

call_recorded(latency_seconds: float) -> None¶

finish(completion_state: str) -> None¶

GroundingSource¶

TaskRequest¶

GoalNode¶

AssessmentResult¶

ExecutionStep¶

ExecutionResult¶

EvaluationResult¶

StageRecord¶

HandoffPackage¶

RunMetrics¶

EvidenceRef¶

RecoveryCycle¶

RecoveryMetadata¶

ReliabilityResult¶

Functions¶

run_assessment(request: TaskRequest, llm: LLMCaller, progress: ProgressReporter = NULL_REPORTER) -> AssessmentResult¶

run_completion(assessment: AssessmentResult, execution: ExecutionResult, evaluation: EvaluationResult, progress: ProgressReporter = NULL_REPORTER) -> ReliabilityResult¶

blocked_result(reason: str, assessment: AssessmentResult | None = None, execution: ExecutionResult | None = None, evaluation: EvaluationResult | None = None, deliverable: str = '', guidance: dict | None = None) -> ReliabilityResult¶

run_evaluation(assessment: AssessmentResult, execution: ExecutionResult, llm: LLMCaller, progress: ProgressReporter = NULL_REPORTER, oracle_evidence: Mapping[str, object] | None = None) -> EvaluationResult¶

build_reliability_recipe(problem: str, domain: str, risk_tier: str | None = None, corpus_agents: list[str] | None = None, top_k: int = 5, log_to_ledger: bool = True) -> Result[ReliabilityRecipe]¶

estimate_cost(model: str, total_tokens: int) -> float¶

prometheus_progress_sink(event: ProgressEvent) -> None¶

`ReliabilityRecipe`¶

`LLMError(RuntimeError)`¶

`LLMCaller(Protocol)`¶

`complete(system: str, user: str, max_tokens: int = DEFAULT_MAX_TOKENS) -> str`¶

`LLMBlockCaller`¶

`complete(system: str, user: str, max_tokens: int = DEFAULT_MAX_TOKENS) -> str`¶

`FakeLLM`¶

`complete(system: str, user: str, max_tokens: int = DEFAULT_MAX_TOKENS) -> str`¶

`TimingLLMCaller`¶

`complete(system: str, user: str, max_tokens: int = DEFAULT_MAX_TOKENS) -> str`¶

`ReliabilityPipelineBlock(AIBlock[TaskRequest, ReliabilityResult, None])`¶

`infer(data: TaskRequest | dict) -> Result[ReliabilityResult]`¶

`envelope_from_result(result: ReliabilityResult, run_id: str = '', recipe_id: str = '', recipe_title: str = '', workflow_id: str = '', execution_mode: str = 'full', inputs: dict | None = None) -> 'AgenticWorkflowRun'`¶

`PremortemFinding`¶

`ProgressEvent`¶

`EtaEstimator`¶

`record_call(latency_seconds: float) -> None`¶

`revise_total(calls_total: int) -> None`¶

`calibrating() -> bool`¶

`avg_latency() -> float`¶

`p50_latency() -> float`¶

`p95_latency() -> float`¶

`elapsed_seconds() -> float`¶

`eta_seconds() -> float`¶

`fraction() -> float`¶

`ProgressReporter`¶

`add_sink(sink: Callable[[ProgressEvent], None]) -> None`¶

`phase(name: str, index: int) -> None`¶

`plan(goal_stack_size: int, n_constraints: int) -> None`¶

`stage(label: str) -> None`¶

`call_recorded(latency_seconds: float) -> None`¶

`finish(completion_state: str) -> None`¶

`GroundingSource`¶

`TaskRequest`¶

`GoalNode`¶

`AssessmentResult`¶

`ExecutionStep`¶

`ExecutionResult`¶

`EvaluationResult`¶

`StageRecord`¶

`HandoffPackage`¶

`RunMetrics`¶

`EvidenceRef`¶

`RecoveryCycle`¶

`RecoveryMetadata`¶

`ReliabilityResult`¶

`run_assessment(request: TaskRequest, llm: LLMCaller, progress: ProgressReporter = NULL_REPORTER) -> AssessmentResult`¶

`run_completion(assessment: AssessmentResult, execution: ExecutionResult, evaluation: EvaluationResult, progress: ProgressReporter = NULL_REPORTER) -> ReliabilityResult`¶

`blocked_result(reason: str, assessment: AssessmentResult | None = None, execution: ExecutionResult | None = None, evaluation: EvaluationResult | None = None, deliverable: str = '', guidance: dict | None = None) -> ReliabilityResult`¶

`run_evaluation(assessment: AssessmentResult, execution: ExecutionResult, llm: LLMCaller, progress: ProgressReporter = NULL_REPORTER, oracle_evidence: Mapping[str, object] | None = None) -> EvaluationResult`¶

`build_reliability_recipe(problem: str, domain: str, risk_tier: str | None = None, corpus_agents: list[str] | None = None, top_k: int = 5, log_to_ledger: bool = True) -> Result[ReliabilityRecipe]`¶

`estimate_cost(model: str, total_tokens: int) -> float`¶

`prometheus_progress_sink(event: ProgressEvent) -> None`¶