Poka Yoke¶
mvp.poka_yoke — G6 Poka-Yoke (mistake-proofing) component.
Cluster: Uncategorised | Type: component | MCP Tools: None
Overview¶
Public API¶
MistakeModeCatalogue¶
Registry of Safeguards indexed by domain.
Constructor:
| Parameter | Type | Default |
|---|---|---|
safeguards_by_domain | dict[str, list[Safeguard]] \| None | None |
Methods:
register(safeguard: Safeguard) -> None¶
Add a single Safeguard to the catalogue.
register_from_domain(domain: str, safeguards: list[Safeguard]) -> None¶
Bulk-register a list of Safeguards under domain.
safeguards_for(ctx: InteractionContext) -> list[Safeguard]¶
Return all safeguards applicable to ctx.
load_domain(domain: str) -> list[Safeguard]¶
Lazy loader hook for the Phase 8.3 harvester.
clear() -> None¶
Remove all registered safeguards (useful for test isolation).
PokaYokeContractCarrier¶
Methods:
evaluate(ctx: InteractionContext | None = None, result: InterceptionResult | None = None) -> dict[str, Any]¶
InterceptionResult¶
Immutable result returned by PokaYokeInterceptor.intercept().
| Field | Type | Default |
|---|---|---|
verdict | SafeguardOutcome | required |
fired_safeguards | tuple[str, ...] | required |
evaluated_count | int | required |
audit_id | str | required |
ctx | InteractionContext | required |
was_disabled | bool | field(default=False) |
overrides_applied | tuple[dict[str, Any], ...] | field(default=()) |
was_shadow | bool | field(default=False) |
PokaYokeInterceptor¶
Orchestrates safeguard evaluation for an InteractionContext.
Constructor:
| Parameter | Type | Default |
|---|---|---|
catalogue | MistakeModeCatalogue \| None | None |
Methods:
is_disabled() -> bool¶
Return True when the
G6_POKA_YOKE_DISABLEDbypass is active.
intercept(ctx: InteractionContext) -> InterceptionResult¶
Run safeguard evaluation for ctx and return an InterceptionResult.
SafeguardOutcome(StrEnum)¶
Verdict returned by a single Safeguard.check() call.
Severity(StrEnum)¶
Per-domain or per-rule severity level (mirrors compliance.rule_engine).
SafeguardType(StrEnum)¶
Toyota-classical safeguard kind discriminator.
Surface(StrEnum)¶
Entry-point surface where the interception fires.
MistakeMode¶
A single identifiable mistake that a user could make in a domain.
| Field | Type | Default |
|---|---|---|
id | str | required |
domain | str | required |
title | str | required |
why_user_misses_this | str | required |
severity | Severity | required |
safeguard_type | SafeguardType | required |
predicate_ref | str | required |
escalation | str | required |
evidence_required | tuple[str, ...] | field(default=()) |
authority_refs | tuple[str, ...] | field(default=()) |
InteractionContext¶
Captures everything known about a user action at interception time.
| Field | Type | Default |
|---|---|---|
operation | str | required |
params | dict[str, Any] | required |
prior_state | dict[str, Any] | required |
user_context | dict[str, Any] | required |
surface | Surface | required |
timestamp | float | field(default_factory=time.time) |
Safeguard¶
A runtime safeguard that wraps one or more MistakeModes.
| Field | Type | Default |
|---|---|---|
name | str | required |
mistake_modes | tuple[MistakeMode, ...] | required |
predicate_ref | str | required |
domain | str | required |
severity_default | Severity | required |
safeguard_type | SafeguardType | required |
Methods:
check(ctx: InteractionContext, predicate_resolver: Callable[[str], Callable[..., SafeguardOutcome]] | None = None) -> SafeguardOutcome¶
Invoke the predicate and return a SafeguardOutcome.
Functions¶
log_interception(ctx: InteractionContext, fired: list[Safeguard], verdict: SafeguardOutcome, override_reason: str | None = None, evaluated_count: int | None = None, overrides_applied: list[dict] | None = None, shadow_mode: bool = False, would_have_been_verdict: str | None = None) -> str¶
Write one audit row and return the audit_id.
get_global_catalogue() -> MistakeModeCatalogue¶
Return the process-wide MistakeModeCatalogue singleton.
evaluate_with_envelope(ctx: InteractionContext | None = None, result: InterceptionResult | None = None) -> dict[str, Any]¶
Return an honest reliability envelope for a poka_yoke interception result.
register_poka_yoke_failure_modes() -> None¶
mistakeproof(domain: str | None = None, operation: str | None = None) -> Callable[[type], type]¶
Class decorator for AIBlocks.
is_shadow_mode() -> bool¶
Return True when G6_POKA_YOKE_SHADOW bypass is active.
get_interceptor() -> PokaYokeInterceptor¶
Return the process-wide PokaYokeInterceptor singleton.
check_cli_input(user_input: str, user_context: dict | None = None) -> InterceptionResult¶
Hook for the CLI entry point (before shlex.split).
check_mcp_tool_call(component: str, op: str, params: dict, user_context: dict | None = None) -> InterceptionResult¶
Hook for MCP tool calls (before registry.invoke).
load_policy_overrides_cached(domain: str) -> dict¶
Return policy overrides for domain, caching the result in-process.
clear_policy_cache() -> None¶
Clear the in-process policy override cache.
default_severity_for(domain: str) -> Severity¶
Return the default Severity for domain.
load_policy_overrides(domain: str) -> dict[str, Any]¶
Load optional JSON policy overrides for domain.
install_poka_yoke(interceptor: PokaYokeInterceptor | None = None) -> None¶
Monkey-patch ComponentRegistry.invoke() with a Poka-Yoke wrapper.
uninstall_poka_yoke() -> None¶
Restore ComponentRegistry.invoke to its pre-patch state.
is_installed() -> bool¶
Return True if the Poka-Yoke monkey-patch is currently active.
ContactCheck(name: str, mistake_modes: tuple[MistakeMode, ...] = (), predicate_ref: str, domain: str, severity_default: Severity = Severity.WARN) -> Safeguard¶
Factory: schema/type/citation existence safeguard.
ValueCheck(name: str, mistake_modes: tuple[MistakeMode, ...] = (), predicate_ref: str, domain: str, severity_default: Severity = Severity.WARN) -> Safeguard¶
Factory: numeric range / quorum / cross-reference safeguard.
SequenceCheck(name: str, mistake_modes: tuple[MistakeMode, ...] = (), predicate_ref: str, domain: str, severity_default: Severity = Severity.WARN) -> Safeguard¶
Factory: prerequisite ordering / state-machine / staleness safeguard.
aggregate_outcomes(outcomes: list[SafeguardOutcome]) -> SafeguardOutcome¶
Aggregate a list of SafeguardOutcome values into a single verdict.