AutoGineer
Generated banner for Poka Yoke
Generated logo for Poka Yoke

Poka Yoke

GitHubOfficialFirst committed by rainmanjamDistributed by GitHub

  • Skill
  • Guidelines
  • Software development
  • MIT
  • Low risk
  • Has scripts
Download ZIP

From the author

Mistake-proof code so misuse cannot be expressed, rather than warning against it. Use when designing an interface, schema, or state machine and the user wants it hard to get wrong ("make invalid states unrepresentable", "so callers cannot screw it up", "type-safe API", "pit of success"); when auditing existing code for footguns ("what could bite us here", "what is easy to misuse", "poka-yoke this repo", "review this diff for ways to get it wrong"); or when a bug has recurred and the fix must close the class rather than the case ("make sure this never happens again", "this is the third time"). Especially for money, auth, permissions, deletion, migrations, and pipelines where failure is silent. Classifies every finding by what happens when the mistake occurs and how the device notices, which is what keeps it from collapsing into generic code review.
— the description by rainmanjam

AutoGineer summary

Written by AutoGineer from the skill’s own files, not by the author

Poka-yoke is a systematic approach to mistake-proofing code by replacing human vigilance with structural constraints. Rather than relying on documentation or warnings, this skill focuses on creating devices—such as type-safe APIs, exhaustive state machines, and boundary-level validation—that make invalid states unrepresentable. It provides a framework for auditing existing codebases to identify hazards, ranking them by their potential for silent failure, and implementing robust fixes that close entire classes of bugs rather than patching individual incidents.

What it does

  • Provides a taxonomy of common coding hazards organized by contact, fixed-value, and motion-step lenses.
  • Defines a four-rung hierarchy of error prevention, from control (impossible) to detection (after the fact).
  • Offers a scanner to identify textually detectable hazards in codebases.
  • Guides the transition from training-based fixes (comments/docs) to device-based fixes (types/constraints).
  • Encourages pushing validation to the system boundary to catch errors at the source.
  • Supplies language-specific patterns for Python, TypeScript, Rust, and Go.

When to use it

  • When designing new interfaces, schemas, or state machines where reliability is critical.
  • When auditing code for 'footguns' that are easy to misuse silently.
  • When a bug has recurred and you need to close the entire class of error permanently.
  • When working on high-stakes systems like payments, authentication, permissions, or data migrations.
  • When you want to move from reactive debugging to proactive structural safety.

Advantages

  • Reduces reliance on human memory and vigilance, which are prone to degradation.
  • Creates self-documenting code where the design limits are enforced by the compiler or runtime.
  • Provides a clear, objective framework for prioritizing technical debt and refactoring efforts.
  • Prevents silent failures by ensuring errors announce themselves immediately.
  • Improves code review quality by focusing on structural hazards rather than stylistic preferences.

Usage examples

“Review this Python function for potential misuse: def transfer(from_id: str, to_id: str, amount: float): ...”
The agent identifies the risk of swapping the two string IDs and suggests using NewType or keyword-only arguments to make the mistake impossible to compile.
“How can I make my state machine for order processing more robust against invalid transitions?”
The agent suggests using a discriminated union or exhaustive pattern matching to ensure every state transition is explicitly handled and illegal states cannot be constructed.
“Audit this repository for silent failure hazards.”
The agent uses the provided scanner to find unchecked external inputs or unbounded destructive operations and ranks them by blast radius.
“We keep having issues with developers forgetting to set the region in our config. How do I fix this?”
The agent recommends using a Pydantic-based settings model with no defaults, forcing the application to fail at startup if the configuration is incomplete.

Install

  1. Download the skill as a ZIP:

    https://autogineer.com/agent-skills/github/poka-yoke/download

  2. Unzip it into the cross-client skills folder of your project:

    mkdir -p .agents/skills && unzip poka-yoke.zip -d .agents/skills

Documentation by rainmanjam

Poka-Yoke: Make the Mistake Unsayable

People will always make mistakes. That is not the problem worth solving. The problem is letting a mistake become a defect.

Shigeo Shingo, a Japanese industrial engineer, worked this out on a switch assembly line in 1961. Workers kept forgetting a spring. The fix was not a reminder: the job was split so the worker first laid both springs in a dish, then fitted them from the dish. A spring left over was the error announcing itself, before the unit could move on.

The dish is a device. "Please remember the spring" is not.

The line that does most of the work

A comment, a docstring, a wiki page, a review checklist, or a line in an instructions file saying "don't do X" is not a poka-yoke. It is training, and training degrades. A device does not. If your fix relies on someone remembering something, keep going.

This applies to your own instructions too. A rule written into a config file competes for attention with every other rule there and loses a little more as the file grows. A check that fails the build does not.

What this changes about the output

Given a design, models will readily list what to fix. They rarely state what the fix makes impossible, and that is the difference between advice you agree with and a constraint you can rely on. That habit is most of what this skill is for.

The other half is refusing to accept a non-device as a fix. "Add validation", "be careful with this function", "document the invariant" are all rung zero. Each has a real device behind it, and naming that device is the work.

Axis 1: what happens when the mistake occurs

Rank every finding on this ladder, and say which rung the current code sits on and which rung your fix reaches.

RungNameMeaning
1ControlThe wrong action cannot be performed. Type error, database constraint, missing permission.
2WarningIt is possible, but announces itself as it happens. A linter, a runtime assertion, a confirmation you cannot skip.
3DetectionIt happens, and you find out afterwards. Tests, logging, monitoring, code review.
0rung zeroTelling people to be careful. Docs, comments, "please remember to".

Detection is not failure; sometimes it is all that is available. But a plan that stops at Detection should say so, rather than presenting it as prevention.

Axis 2: how the device notices

Shingo's three inspection lenses. They are a checklist for finding hazards, not decoration:

  • Contact — can the wrong thing physically fit? Two adjacent parameters of the same type can be swapped silently. A string that should be one of four values. Money as a float.

  • Fixed-value — is the set complete? A switch with no exhaustiveness check. A config where a missing key silently means "off". An enum handled in three of five places.

  • Motion-step — is the order right, and did every step happen? A two-phase write with no transaction. A retry with no idempotency key. A resource acquired on one path and released on another.

Inspect at the source

The cheapest place to catch a mistake is where it is made, not where it surfaces. A validation that runs three layers below the input has already let the bad value travel, and the stack trace will point at the wrong module. Push the check to the boundary the value crosses.

Designing something new

Mistake-proofing is cheapest before the code has callers. Once it has them, every device is a migration; before it has them, a device is free.

Work from the call site. A signature that reads fine in isolation often reads terribly where it is used:

# the mistake is expressible: nothing stops refunding an order that was never paid
def refund(order: dict) -> Refund:
    return payments.refund(order["payment_id"])

# the mistake is no longer expressible
def refund(order: PaidOrder) -> Refund: ...

The moves, roughly in order of how often they apply:

Make invalid states unrepresentable. A bag of optional fields where only certain combinations are legal becomes a discriminated union where the illegal ones cannot be constructed.

Parse, don't validate. Convert unstructured input into a type that carries proof at the boundary, once, rather than re-checking the same string in nine places.

Distinguish concepts that share a primitive. transfer(from: str, to: str) accepts its arguments transposed. Distinct types for the two concepts, or keyword-only parameters, make the transposition a compile error.

Encode the order. When calls must happen in sequence, let each step return the type the next step requires, so the wrong order does not typecheck.

Make the destructive path narrower than the safe one. A required, non-defaulting argument for the scope of a delete. A default that means "nothing" rather than "everything".

Close by naming what the design now makes impossible, and, just as importantly, what you deliberately left possible and why. A design whose limits are unstated will be trusted past them.

Auditing code that already exists

You are not looking for bugs. A bug is a mistake that already happened. You are looking for mistakes that are available: places where doing the wrong thing is easy, silent, and looks correct.

Run the bundled scanner first for the textually detectable shapes, then read for the ones no scanner can see:

python3 scripts/detect_hazards.py --paths .          # whole tree
python3 scripts/detect_hazards.py --staged           # pre-commit
python3 scripts/detect_hazards.py --diff --json      # CI, exits non-zero on findings
python3 scripts/detect_hazards.py --severity high

No dependencies, so it runs in CI and in a pre-commit hook without an install step. It reports what it scanned: a scan of zero files exits non-zero rather than reporting a clean bill of health, because an all-clear you got by typo is worse than no check.

Rank findings by blast radius times ease of the mistake. An unchecked value reaching a write, a delete, a payment or an auth decision outranks one that can only produce a clean crash. For each finding, state: where it is, what the mistake is, what the consequence is, what device exists today, what device would close it, and which rung that reaches.

references/hazard-catalog.md is the taxonomy of shapes with their IDs and devices. Language-specific patterns are in references/lang-python.md, references/lang-typescript.md and references/lang-rust-go.md.

After an incident

Separate three things that get conflated, because the fix belongs to the third:

  • Defect — what the user experienced.

  • Mistake — the specific wrong action someone took.

  • Hazard — the property of the system that made that mistake available.

Fixing the mistake fixes one case. Fixing the hazard fixes the class. Then sweep: the same shape almost certainly exists elsewhere, and finding the second and third instance is the difference between a patch and a lesson.

Attribute cause to the system rather than to a person. Not primarily for kindness: "they made a mistake" is a complete-sounding explanation that predicts nothing and prevents nothing, and it ends the investigation early.

What good output looks like

  • Anchored to lines. orders.py:142, apply_discount is reviewable; "the discount logic" is not.

  • Ranked, with the ranking visible, so a reader who stops halfway has still covered the ones that matter.

  • A named device per finding, not "add validation".

  • The rung stated, before and after.

  • The limits stated. What the fix does not cover is the part readers most need and most often do not get.

  • Sized honestly. Three findings that matter beat eleven padded to a round number.

What to avoid

Accepting rung zero as a fix. If the proposal is a comment, a doc, or a convention, the work is not finished.

Devices nobody can bypass being confused with devices nobody does bypass. A pre-commit hook is skippable with --no-verify; it needs CI behind it to be a real gate. Say which one you are proposing.

Over-fitting to one incident. Machinery that prevents one specific failure must itself be understood and maintained. Ask whether the shape is common enough to justify it.

Treating monitoring as prevention. Detection lowers the cost of a failure; it does not lower the likelihood. Both are worth having, and conflating them means the likelihood never gets addressed.

Evidence, and its limits

This method was benchmarked at 591 blind-graded runs across six model families, scored against assertions written before the runs by a grader that never saw which configuration produced a response. The behaviour it most reliably changes is stating what a design forecloses: 45% of responses did that unprompted, 80% with the method applied, across 132 graded verdicts.

That average conceals where the effect lives. Asked squarely to design an interface, models already do it 77% of the time; the skills add eleven points. The large gains are in tasks where nobody asked for a design review — writing an endpoint goes 14% to 79%, shipping an agent feature 33% to 83%, building a form 29% to 64%.

Stated honestly, because the limits matter: every run was the first turn of a fresh session, so this measures the ceiling rather than what survives a long working session. The comparison was against no methodology at all, not against a different one, so it does not establish that this method is what produced the gain. And the method costs something measurable: responses became somewhat worse at spotting the specific defect already on the page while becoming better at changing the shape that allowed it. If you want the bug in front of you found, use a reviewer. If you want that class of bug to stop being expressible, use this.

Raw runs, the harness and the assertion checklists are at https://github.com/rainmanjam/poka-yoke.

Files

scripts/detect_hazards.py

#!/usr/bin/env python3
"""Heuristic detector for poka-yoke hazards, shapes in code that make mistakes easy.

This is a fast first pass, not an oracle. It finds textually-detectable hazards so a
reviewer can spend their attention on the interface-level questions a regex cannot ask.
Expect real false positives; every hit is a question, not a verdict.

Hazard IDs match references/hazard-catalog.md. Standard library only.

Examples:
    detect_hazards.py --diff                  # uncommitted changes, changed lines only
    detect_hazards.py --staged                # staged changes
    detect_hazards.py --since HEAD~10         # last 10 commits
    detect_hazards.py --paths src/ lib/       # explicit paths
    detect_hazards.py --diff --severity high  # only the ones that bite hardest
    detect_hazards.py --paths . --json        # machine-readable
"""

from __future__ import annotations

import argparse
import ast
import json
import os
import re
import subprocess
import sys
from dataclasses import dataclass, field
from pathlib import Path

# --------------------------------------------------------------------------------------
# Rule definitions
# --------------------------------------------------------------------------------------

PY = {".py", ".pyi"}
TS = {".ts", ".tsx", ".js", ".jsx", ".mjs", ".cjs"}
GO = {".go"}
RS = {".rs"}
SQL = {".sql"}
ALL_EXTS = PY | TS | GO | RS | SQL

LENS = {"C": "contact", "F": "fixed-value", "M": "motion-step", "X": "removed-device"}


@dataclass(frozen=True)
class Rule:
    id: str
    name: str
    severity: str  # high | medium | low
    exts: frozenset
    pattern: re.Pattern
    device: str
    negate: re.Pattern | None = None  # if this also matches the line, skip


def R(id, name, severity, exts, pattern, device, negate=None, flags=0):
    return Rule(
        id=id,
        name=name,
        severity=severity,
        exts=frozenset(exts),
        pattern=re.compile(pattern, flags),
        device=device,
        negate=re.compile(negate, flags) if negate else None,
    )


RULES: list[Rule] = [
    # ---- X: devices that were removed -------------------------------------------------
    R("X1", "Swallowed error", "high", TS,
      r"catch\s*(\([^)]*\))?\s*\{\s*\}",
      "Handle it or let it propagate; catching to do nothing turns a loud failure quiet."),
    R("X1", "Swallowed error", "high", TS,
      r"\.catch\s*\(\s*\(\s*\)\s*=>\s*(\{\s*\}|null|undefined)\s*\)",
      "Handle the rejection or let it propagate."),
    R("X1", "Bare except", "high", PY,
      r"^\s*except\s*:",
      "Catch the specific exception; bare except also swallows KeyboardInterrupt/SystemExit."),
    R("X1", "Discarded error return", "high", GO,
      r",\s*_\s*:?=\s*\w|^\s*_\s*=\s*\w[\w.]*\(",
      "Check the error. Enable errcheck in golangci-lint to make this a build failure."),
    R("X2", "Unwrap / expect on a fallible value", "medium", RS,
      r"\.(unwrap|expect)\s*\(",
      "Propagate with ? or handle the error; deny clippy::unwrap_used."),
    R("X2", "Silent default on error", "medium", RS,
      r"\.unwrap_or_default\s*\(\s*\)",
      "A default on an error path hides the failure; branch on the error explicitly."),
    R("X2", "parseInt without radix", "medium", TS,
      r"parseInt\s*\(\s*[^,)]+\)",
      "Pass the radix and check for NaN, or use a schema parse at the boundary."),
    R("X3", "Focused test disables the suite", "high", TS,
      r"\b(it|test|describe|context)\.only\s*\(|\bfdescribe\s*\(|\bfit\s*\(",
      "Remove before merge; fail CI on focused tests."),
    R("X3", "Skipped test", "medium", PY,
      r"@pytest\.mark\.skip|@unittest\.skip",
      "A skipped test is a detection device switched off. Fix or delete it."),
    R("X3", "Skipped test", "medium", TS,
      r"\b(it|test|describe)\.skip\s*\(|\bxit\s*\(|\bxdescribe\s*\(",
      "A skipped test is a detection device switched off. Fix or delete it."),
    R("X3", "Skipped test", "medium", GO, r"\bt\.Skip\s*\(",
      "A skipped test is a detection device switched off. Fix or delete it."),
    R("X3", "Skipped test", "medium", RS, r"^\s*#\[ignore\]",
      "A skipped test is a detection device switched off. Fix or delete it."),
    R("X4", "Type-checker suppression", "medium", TS,
      r"@ts-ignore|@ts-nocheck|\bas\s+unknown\s+as\b|eslint-disable(?!-next-line\s+\S+\s+--)",
      "Each suppression is a hole in the guarantee. Require a reason and an issue link."),
    R("X4", "Explicit any", "medium", TS,
      r":\s*any\b|<any>|Array<any>|as\s+any\b",
      "any disables the type system exactly where guarantees matter. Parse at the boundary."),
    R("X4", "Type-checker suppression", "medium", PY,
      r"#\s*type:\s*ignore(?!\[)",
      "Narrow it to a specific error code and add a reason."),
    R("X4", "Untyped container", "low", GO,
      r"\binterface\{\}|\bany\b\s*[,)\]]",
      "Prefer a concrete type or a constrained generic."),
    R("X4", "unsafe block", "medium", RS, r"\bunsafe\s*\{",
      "Require a // SAFETY: comment stating the invariant being upheld."),
    R("X5", "Mutable default argument", "high", PY,
      r"def\s+\w+\s*\([^)]*=\s*(\[\s*\]|\{\s*\}|set\s*\(\s*\))",
      "Use None and construct inside the function; the default is shared across all calls."),

    # ---- F: fixed-value ---------------------------------------------------------------
    R("F2", "Unbounded DELETE", "high", SQL | PY | TS | GO | RS,
      r"\bDELETE\s+FROM\b(?!.*\bWHERE\b)",
      "Require a WHERE clause; refuse an empty predicate.", flags=re.I),
    R("F2", "Unbounded UPDATE", "high", SQL | PY | TS | GO | RS,
      r"\bUPDATE\s+[\w.\"`\[\]]+\s+SET\b(?!.*\bWHERE\b)",
      "Require a WHERE clause; refuse an empty predicate.", flags=re.I),
    R("F2", "Destructive DDL", "high", SQL | PY | TS | GO | RS,
      r"\b(DROP\s+(TABLE|COLUMN|DATABASE|SCHEMA)|TRUNCATE\s+TABLE)\b",
      "Use expand/contract; gate destructive DDL behind an explicit CI acknowledgment.",
      flags=re.I),
    R("F2", "Bulk delete", "high", TS | PY,
      r"\.(deleteMany|delete_many|destroy_all|delete_all|drop_all|removeMany)\s*\(\s*\)",
      "Refuse an empty filter; cap the affected count and require confirmation above it."),
    R("F2", "Recursive force remove", "high", ALL_EXTS,
      r"rm\s+-[a-zA-Z]*[rR][a-zA-Z]*f|rm\s+-[a-zA-Z]*f[a-zA-Z]*[rR]",
      "Validate the path is non-empty and inside the expected root before deleting."),
    R("F4", "Config read away from startup", "medium", PY,
      r"os\.(getenv|environ)",
      "Parse the whole config into a typed object at startup so a missing key fails the deploy.",
      negate=r"(settings|config|conf|env)\.py"),
    R("F4", "Config read away from startup", "medium", TS,
      r"process\.env\.\w+",
      "Parse the whole config into a typed object at startup so a missing key fails the deploy.",
      negate=r"(config|env|settings)\.(ts|js)"),
    R("F7", "Unbounded read", "low", PY | TS,
      r"\.read\s*\(\s*\)|\.readAll\s*\(|ioutil\.ReadAll",
      "Cap the size at the boundary; an unbounded read is a resource-exhaustion incident."),

    # ---- C: contact -------------------------------------------------------------------
    R("C2", "Boolean flag parameter", "medium", TS,
      r"\b\w+\s*:\s*boolean\s*[,)]",
      "Use an enum, a named options object, or two functions; booleans are unreadable at the call site."),
    R("C2", "Boolean flag parameter", "medium", GO,
      r"func\s+\w+\s*\([^)]*\bbool\b[^)]*\)",
      "Use a named option type; a bare bool is unreadable at the call site."),
    R("C2", "Boolean default parameter", "medium", PY,
      r"def\s+\w+\s*\([^)]*\b\w+\s*(:\s*bool\s*)?=\s*(True|False)",
      "Use an enum, or at minimum make it keyword-only so the name appears at the call site."),
    R("C5", "Duration without a unit", "medium", TS | GO | PY,
      r"\b(timeout|delay|interval|ttl|expiry|duration|retryAfter|retry_after)\s*:?\s*(number|int|float|=\s*\d+)",
      "Encode the unit in the type (Duration) or in the name (timeoutMs). Unit mismatches are silent."),
    R("C6", "Money as a float", "high", PY | TS | GO | RS,
      r"\b(price|amount|total|balance|cost|fee|subtotal|revenue)\w*\s*:\s*(float|number|f32|f64)\b"
      r"|\bfloat\s*\(\s*\w*(price|amount|total|balance)",
      "Use integer minor units in a Money type carrying its currency, or a decimal type."),
    R("C7", "Unvalidated parse", "high", TS,
      r"JSON\.parse\s*\(",
      "Parse into a schema (zod/valibot) at the boundary; JSON.parse returns any."),
    R("C7", "Unvalidated request body", "high", PY,
      r"(request|req)\.(json|get_json)\s*\(\s*\)(?!\s*\))",
      "Parse into a Pydantic model with extra='forbid' so unknown or missing fields fail loudly."),
    R("C9", "Naive datetime", "medium", PY,
      r"datetime\.utcnow\s*\(\s*\)|datetime\.now\s*\(\s*\)",
      "Use datetime.now(timezone.utc), and inject a clock so time is testable."),

    # ---- M: motion-step ---------------------------------------------------------------
    R("M4", "Unmanaged resource", "medium", PY,
      r"^\s*(\w+\s*=\s*)?open\s*\(",
      "Use a context manager; the error path will leak otherwise.",
      negate=r"\bwith\b"),
    R("M6", "Dangling async task", "high", PY,
      r"^\s*(await\s+)?asyncio\.create_task\s*\(",
      "Keep a reference; an unreferenced task can be garbage collected mid-flight (ruff RUF006).",
      negate=r"=\s*(await\s+)?asyncio\.create_task"),
    R("M6", "Unawaited promise-returning call", "low", TS,
      r"^\s*\w+\.(save|update|create|delete|insert|write|send|publish|commit)\s*\(",
      "If this returns a promise, await it: a floating write is silently lost. "
      "Enable @typescript-eslint/no-floating-promises.",
      negate=r"\b(await|return|yield|void)\b|\.then\(|=\s"),
    R("M2", "Retryable effect without an idempotency key", "high", ALL_EXTS,
      r"\b(def|func|function|fn|async\s+function)\s+\w*(charge|refund|capture|payout|transfer|"
      r"sendEmail|send_email|publish|notify)\w*\s*[(<]",
      "Require an idempotency key parameter, backed by a unique constraint on (entity, key).",
      negate=r"idempot", flags=re.I),
    R("M1", "Two-phase construction", "medium", ALL_EXTS,
      r"\b(def|func|function|fn)\s+(init|initialize|connect|setup|configure|start)\s*[(<]",
      "Have the constructor or a factory return a ready object, or use typestate; "
      "'call this first' is not enforceable.",
      negate=r"__init__|func\s+init\s*\(\s*\)\s*\{"),

    # ---- F1: exhaustiveness -----------------------------------------------------------
    R("F1", "Wildcard match arm", "medium", RS,
      r"^\s*_\s*=>",
      "In domain logic a wildcard turns a future compile error into a silent fallthrough."),
    R("F1", "Switch without exhaustiveness check", "low", TS,
      r"^\s*switch\s*\(",
      "Add a default arm calling assertNever(x: never) so a new variant breaks the build."),
    R("F1", "Switch without a default", "low", GO,
      r"^\s*switch\s+\w+\s*\{",
      "Enable the 'exhaustive' linter with default-signifies-exhaustive: false."),
]

# Rules a real linter already does better. They stay available behind --all for repos that
# do not run those linters, but they are off by default: a tool that does eight things
# nothing else does is more useful than one doing forty things worse. The value here is the
# pointer, knowing which linter to enable beats a second-rate reimplementation of it.
COVERED_BY: dict[tuple[str, str], str] = {
    ("X1", "Swallowed error"): "eslint no-empty",
    ("X1", "Bare except"): "ruff E722",
    ("X1", "Discarded error return"): "golangci-lint errcheck",
    ("X2", "Unwrap / expect on a fallible value"): "clippy::unwrap_used",
    ("X2", "Silent default on error"): "clippy",
    ("X2", "parseInt without radix"): "eslint radix",
    ("X3", "Focused test disables the suite"): "eslint jest/no-focused-tests",
    ("X3", "Skipped test"): "eslint jest/no-disabled-tests",
    ("X4", "Type-checker suppression"): "@typescript-eslint/ban-ts-comment, mypy --strict",
    ("X4", "Explicit any"): "@typescript-eslint/no-explicit-any",
    ("X4", "Untyped container"): "golangci-lint",
    ("X4", "unsafe block"): "clippy",
    ("X5", "Mutable default argument"): "ruff B006",
    ("C9", "Naive datetime"): "ruff DTZ",
    ("M4", "Unmanaged resource"): "ruff SIM115",
    ("M6", "Dangling async task"): "ruff RUF006",
    ("F1", "Wildcard match arm"): "clippy::wildcard_enum_match_arm",
    ("F7", "Unbounded read"): "",
    ("F3", "assert used for validation"): "ruff S101",
    ("C6", "Equality comparison on a float"): "ruff PLR0133",
}


def covered(rule_id: str, name: str) -> str:
    return COVERED_BY.get((rule_id, name), "")


# --------------------------------------------------------------------------------------
# AST pass (Python only), catches what regexes can't see
# --------------------------------------------------------------------------------------


def python_ast_findings(path: Path, source: str) -> list[dict]:
    """Structural checks that need real parsing: adjacent same-type params, assert-as-
    validation, and equality comparison on floats."""
    out = []
    try:
        tree = ast.parse(source, filename=str(path))
    except SyntaxError:
        return out

    for node in ast.walk(tree):
        if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
            args = node.args.posonlyargs + node.args.args
            # skip self/cls
            if args and args[0].arg in ("self", "cls"):
                args = args[1:]
            annotated = [(a.arg, ast.unparse(a.annotation)) for a in args if a.annotation]
            for i in range(len(annotated) - 1):
                (n1, t1), (n2, t2) = annotated[i], annotated[i + 1]
                if t1 == t2 and t1 in ("str", "int", "float", "bytes", "bool", "UUID"):
                    out.append({
                        "id": "C1",
                        "name": "Adjacent same-type parameters",
                        "severity": "high",
                        "line": node.lineno,
                        "snippet": f"def {node.name}(..., {n1}: {t1}, {n2}: {t2}, ...)",
                        "device": f"'{n1}' and '{n2}' are both {t1} and can be swapped silently. "
                                  "Use NewType per concept, or make them keyword-only.",
                    })
            # positional args on a wide signature
            if len(args) >= 4 and not node.args.kwonlyargs:
                out.append({
                    "id": "C1",
                    "name": "Wide positional signature",
                    "severity": "low",
                    "line": node.lineno,
                    "snippet": f"def {node.name}({len(args)} positional params)",
                    "device": "Make parameters keyword-only with '*' so names appear at the call site.",
                })

        elif isinstance(node, ast.Assert):
            out.append({
                "id": "F3",
                "name": "assert used for validation",
                "severity": "medium",
                "line": node.lineno,
                "snippet": ast.unparse(node)[:100],
                "device": "assert is stripped under python -O. Raise an explicit exception instead.",
            })

        elif isinstance(node, ast.Compare):
            for op in node.ops:
                if isinstance(op, (ast.Eq, ast.NotEq)):
                    src = ast.unparse(node)
                    if re.search(r"\d+\.\d+", src):
                        out.append({
                            "id": "C6",
                            "name": "Equality comparison on a float",
                            "severity": "medium",
                            "line": node.lineno,
                            "snippet": src[:100],
                            "device": "Use math.isclose, or a Decimal/integer-minor-unit type.",
                        })
    return out


# --------------------------------------------------------------------------------------
# File and diff collection
# --------------------------------------------------------------------------------------

SKIP_DIRS = {
    ".git", "node_modules", "vendor", "dist", "build", "target", "__pycache__",
    ".venv", "venv", ".tox", ".mypy_cache", ".pytest_cache", ".next", "coverage",
    ".terraform", "site-packages",
}


def git(*args: str, cwd: Path) -> str:
    try:
        r = subprocess.run(["git", *args], cwd=cwd, capture_output=True, text=True, timeout=30)
        return r.stdout if r.returncode == 0 else ""
    except (subprocess.SubprocessError, FileNotFoundError):
        return ""


def changed_files_and_lines(cwd: Path, mode: str, since: str | None):
    """Return {path: set(changed_line_numbers)}. Empty set means 'whole file'."""
    if mode == "staged":
        diff_args = ["diff", "--cached", "-U0"]
    elif mode == "since":
        diff_args = ["diff", f"{since}..HEAD", "-U0"]
    else:
        diff_args = ["diff", "HEAD", "-U0"]

    raw = git(*diff_args, cwd=cwd)
    if not raw.strip() and mode == "diff":
        # Clean tree, fall back to recent commits, which is what the user usually means.
        raw = git("diff", "HEAD~5..HEAD", "-U0", cwd=cwd)

    result: dict[str, set[int]] = {}
    current = None
    for line in raw.splitlines():
        if line.startswith("+++ b/"):
            current = line[6:]
            result.setdefault(current, set())
        elif line.startswith("@@") and current:
            m = re.search(r"\+(\d+)(?:,(\d+))?", line)
            if m:
                start = int(m.group(1))
                count = int(m.group(2) or 1)
                result[current].update(range(start, start + count))
    return {k: v for k, v in result.items() if v}


def collect_paths(roots: list[str]) -> list[Path]:
    out = []
    for root in roots:
        p = Path(root)
        if p.is_file():
            if p.suffix in ALL_EXTS:
                out.append(p)
        elif p.is_dir():
            for dirpath, dirnames, filenames in os.walk(p):
                dirnames[:] = [d for d in dirnames if d not in SKIP_DIRS and not d.startswith(".")]
                for fn in filenames:
                    fp = Path(dirpath) / fn
                    if fp.suffix in ALL_EXTS:
                        out.append(fp)
    return out


# --------------------------------------------------------------------------------------
# Scanning
# --------------------------------------------------------------------------------------

COMMENT_ONLY = re.compile(r"^\s*(//|#|/\*|\*|--)")


def scan_file(path: Path, only_lines: set[int] | None) -> list[dict]:
    try:
        source = path.read_text(encoding="utf-8", errors="replace")
    except (OSError, UnicodeDecodeError):
        return []
    if len(source) > 2_000_000:
        return []

    findings = []
    ext = path.suffix
    lines = source.splitlines()

    for lineno, line in enumerate(lines, 1):
        if only_lines and lineno not in only_lines:
            continue
        if COMMENT_ONLY.match(line) or len(line) > 500:
            continue
        for rule in RULES:
            if ext not in rule.exts:
                continue
            if not INCLUDE_COVERED and covered(rule.id, rule.name):
                continue
            if rule.negate and (rule.negate.search(line) or rule.negate.search(str(path))):
                continue
            if rule.pattern.search(line):
                findings.append({
                    "id": rule.id,
                    "name": rule.name,
                    "severity": rule.severity,
                    "line": lineno,
                    "snippet": line.strip()[:120],
                    "device": rule.device,
                })

    if ext in PY:
        for f in python_ast_findings(path, source):
            if not INCLUDE_COVERED and covered(f["id"], f["name"]):
                continue
            if not only_lines or f["line"] in only_lines:
                findings.append(f)

    for f in findings:
        f["file"] = str(path)
        f["lens"] = LENS.get(f["id"][0], "unknown")
    return findings


# --------------------------------------------------------------------------------------
# Output
# --------------------------------------------------------------------------------------

INCLUDE_COVERED = False

SEV_ORDER = {"high": 0, "medium": 1, "low": 2}
COLOR = {"high": "\033[31m", "medium": "\033[33m", "low": "\033[90m"}
RESET = "\033[0m"


def render(findings: list[dict], scope: str, use_color: bool) -> str:
    if not findings:
        return f"No hazards detected in {scope}.\n\nThe lenses still apply, run them by hand:\n" \
               "  contact:     can the wrong thing fit?\n" \
               "  fixed-value: can an incomplete or wrong-sized set pass?\n" \
               "  motion-step: can the steps happen in the wrong order?"

    findings.sort(key=lambda f: (SEV_ORDER[f["severity"]], f["file"], f["line"]))
    counts = {"high": 0, "medium": 0, "low": 0}
    for f in findings:
        counts[f["severity"]] += 1

    out = [
        f"Poka-yoke hazard scan, {scope}",
        f"{counts['high']} high · {counts['medium']} medium · {counts['low']} low",
        "",
        "Heuristics with real false positives. Read the surrounding code before acting.",
        "",
    ]

    grouped: dict[str, list[dict]] = {}
    for f in findings:
        grouped.setdefault(f"{f['id']} {f['name']}", []).append(f)

    for key, group in sorted(grouped.items(), key=lambda kv: SEV_ORDER[kv[1][0]["severity"]]):
        sev = group[0]["severity"]
        tag = f"{COLOR[sev]}{sev.upper():<6}{RESET}" if use_color else f"{sev.upper():<6}"
        out.append(f"{tag} {key}  ({group[0]['lens']} lens, {len(group)} site"
                   f"{'s' if len(group) > 1 else ''})")
        out.append(f"       device: {group[0]['device']}")
        for f in group[:8]:
            out.append(f"       {f['file']}:{f['line']}  {f['snippet']}")
        if len(group) > 8:
            out.append(f"       … and {len(group) - 8} more")
        out.append("")

    return "\n".join(out)


def main() -> int:
    ap = argparse.ArgumentParser(
        description="Detect poka-yoke hazards, shapes in code that make mistakes easy.",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog=__doc__.split("Examples:")[-1],
    )
    src = ap.add_mutually_exclusive_group()
    src.add_argument("--diff", action="store_true",
                     help="scan uncommitted changes (falls back to HEAD~5..HEAD if clean)")
    src.add_argument("--staged", action="store_true", help="scan staged changes")
    src.add_argument("--since", metavar="REF", help="scan changes since REF (e.g. HEAD~10)")
    src.add_argument("--paths", nargs="+", metavar="PATH", help="scan these files or directories")
    ap.add_argument("--severity", choices=["high", "medium", "low"], default="low",
                    help="minimum severity to report (default: low)")
    ap.add_argument("--id", nargs="+", metavar="ID",
                    help="only report these hazard IDs (e.g. --id C1 F2 M2)")
    ap.add_argument("--all", action="store_true", dest="include_covered",
                    help="also run the rules a real linter does better (off by default)")
    ap.add_argument("--json", action="store_true", help="emit JSON")
    ap.add_argument("--repo", default=".", help="repository root (default: .)")
    args = ap.parse_args()

    global INCLUDE_COVERED
    INCLUDE_COVERED = args.include_covered
    repo = Path(args.repo).resolve()
    findings: list[dict] = []

    # poka-yoke: an empty scan reports itself instead of looking like a clean bill of health [control]
    scanned = 0

    if args.paths:
        scope = f"paths: {', '.join(args.paths)}"
        for p in collect_paths(args.paths):
            scanned += 1
            findings += scan_file(p, None)
        if scanned == 0:
            # Zero findings from zero files is not an all-clear, and it used to be
            # indistinguishable from one. Exit non-zero: failing to do the job should
            # not look like doing the job and finding nothing.
            msg = ("Scanned 0 files. This is NOT an all-clear.\n"
                   f"Nothing under {', '.join(args.paths)} has a supported extension.\n"
                   f"Supported: {', '.join(sorted(ALL_EXTS))}")
            print(json.dumps({"scope": scope, "files_scanned": 0, "count": 0,
                              "findings": [], "error": msg}, indent=2)
                  if args.json else msg, file=sys.stdout if args.json else sys.stderr)
            return 2
    else:
        mode = "staged" if args.staged else ("since" if args.since else "diff")
        scope = {"staged": "staged changes",
                 "since": f"changes since {args.since}",
                 "diff": "uncommitted changes"}[mode]
        changed = changed_files_and_lines(repo, mode, args.since)
        if not changed:
            msg = ("No changed files found. The tree may be clean and have no recent commits, "
                   "or this may not be a git repository.\nUse --paths to scan explicitly, "
                   "e.g. detect_hazards.py --paths src/")
            print(json.dumps({"findings": [], "note": msg}) if args.json else msg)
            return 0
        for rel, lines in changed.items():
            fp = repo / rel
            if fp.suffix in ALL_EXTS and fp.exists():
                scanned += 1
                findings += scan_file(fp, lines)

    threshold = SEV_ORDER[args.severity]
    findings = [f for f in findings if SEV_ORDER[f["severity"]] <= threshold]
    if args.id:
        wanted = {i.upper() for i in args.id}
        findings = [f for f in findings if f["id"] in wanted]

    if args.json:
        print(json.dumps({"scope": scope, "files_scanned": scanned,
                          "count": len(findings), "findings": findings}, indent=2))
    else:
        print(render(findings, scope, use_color=sys.stdout.isatty()))
        print(f"\nScanned {scanned} file{'' if scanned == 1 else 's'}.")
        if not INCLUDE_COVERED:
            tools = sorted({v.split(",")[0].split()[0] for v in COVERED_BY.values() if v})
            # len(COVERED_BY) counts ENTRIES, and one entry can suppress several
            # per-language rules, so it under-reported by three. Count the rules.
            n_suppressed = sum(1 for r in RULES if (r.id, r.name) in COVERED_BY)
            print(f"\nNot checked here, {n_suppressed} further hazard rules are covered "
                  f"better by {', '.join(tools)}.\nEnable those rather than relying on this: "
                  f"see assets/devices/lint/. Use --all to run them anyway.")

    return 0


if __name__ == "__main__":
    sys.exit(main())

Requirements and compatibility

Written for: GitHub Copilot

Works with: Claude Code, Claude apps, OpenAI Codex, GitHub Copilot, Cursor, Gemini CLI, Kiro, OpenCode, Goose, Amp, Roo Code, JetBrains Junie, Factory Droid, Devin (ex-Windsurf), Mistral Vibe, OpenHands, Letta

“Works with” means the tool reads the open Agent Skills format; a skill written for one tool may use features others lack.

Needs: Needs Python

The author says: Cross-platform. The bundled scanner needs Python 3.9+ and no third-party packages. Everything else is language-agnostic guidance; worked examples are TypeScript, Python, Go, Rust and SQL.

Security

What our scanner noticed· Low risk

  • Scripts that your agent may run

This skill includes scripts that your agent may run. Read them before you use it.

A static scan is not a guarantee. Skills are third-party content: review a skill before you let an agent use it.

Attribution and licence

First committed by
GitHub avatar of rainmanjamrainmanjam
Publisher
GitHub (Official)
Distributed by
GitHub
Source
github/awesome-copilot · skills/poka-yoke · commit 797337f
Licence
MIT License (redistribution allowed)
  • LICENSE (added by AutoGineer from LICENSE)
  • ATTRIBUTION.md (written by AutoGineer)
Synced
Contributors
rainmanjam

AutoGineer is not affiliated with or endorsed by rainmanjam or GitHub.

Download ZIP

The ZIP holds the skill’s files, its licence files and an ATTRIBUTION.md. Download poka-yoke.zip