ogs.verification

Engine

Run verification checks against a PatternIR.

Parameters:

Name	Type	Description	Default
pattern	PatternIR	The pattern to verify.	required
checks	list[Callable[[PatternIR], list[Finding]]] | None	Optional subset of OGS domain checks to run. Defaults to ALL_CHECKS (8 OGS-specific checks).	None
include_gds_checks	bool	Run GDS generic checks (G-001..G-006) via to_system_ir() projection. Defaults to True.	True

Returns:

Type	Description
VerificationReport	A VerificationReport with all findings.

Source code in packages/gds-games/ogs/verification/engine.py

def verify(
    pattern: PatternIR,
    checks: list[Callable[[PatternIR], list[Finding]]] | None = None,
    include_gds_checks: bool = True,
) -> VerificationReport:
    """Run verification checks against a PatternIR.

    Args:
        pattern: The pattern to verify.
        checks: Optional subset of OGS domain checks to run. Defaults to
            ``ALL_CHECKS`` (8 OGS-specific checks).
        include_gds_checks: Run GDS generic checks (G-001..G-006) via
            ``to_system_ir()`` projection. Defaults to True.

    Returns:
        A VerificationReport with all findings.
    """
    checks = checks or ALL_CHECKS
    findings: list[Finding] = []
    for check_fn in checks:
        findings.extend(check_fn(pattern))

    if include_gds_checks:
        system_ir = pattern.to_system_ir()
        for check_fn_gds in GDS_ALL_CHECKS:
            findings.extend(check_fn_gds(system_ir))

    return VerificationReport(pattern_name=pattern.name, findings=findings)

Findings

Bases: VerificationReport

OGS-compatible verification report with pattern_name alias.

Source code in packages/gds-games/ogs/verification/findings.py

class VerificationReport(_GDSVerificationReport):
    """OGS-compatible verification report with pattern_name alias."""

    def __init__(self, **data):
        # Map pattern_name to system_name for GDS base
        if "pattern_name" in data and "system_name" not in data:
            data["system_name"] = data["pattern_name"]
        super().__init__(**data)

    @property
    def pattern_name(self) -> str:
        """Alias for system_name — OGS calls it 'pattern_name'."""
        return self.system_name

pattern_name property

Alias for system_name — OGS calls it 'pattern_name'.

Type Checks

Type consistency checks T-001 through T-006.

These checks verify that individual flows and game signatures are internally consistent — that flow labels match game port types, that games have complete signatures, and that flow directions match their semantic types.

All comparisons use token-based matching (see tokens.py): signature strings are split on + and , into normalized tokens, and compatibility is checked via set containment (subset) or overlap.

Key distinction from structural checks (S-series): type checks validate per-flow / per-game properties, while structural checks validate global composition invariants (acyclicity, independence, etc.).

check_t001_domain_codomain_matching(pattern)

T-001: For every covariant game-to-game flow, verify the flow label is consistent with source codomain (Y) or target domain (X).

Uses token-based comparison: flow label tokens must be a subset of the source Y tokens OR the target X tokens.

Source code in packages/gds-games/ogs/verification/type_checks.py

def check_t001_domain_codomain_matching(pattern: PatternIR) -> list[Finding]:
    """T-001: For every covariant game-to-game flow, verify the flow label
    is consistent with source codomain (Y) or target domain (X).

    Uses token-based comparison: flow label tokens must be a subset of
    the source Y tokens OR the target X tokens.
    """
    findings = []
    game_sigs = {g.name: g.signature for g in pattern.games}

    for flow in pattern.flows:
        if flow.direction != FlowDirection.COVARIANT:
            continue
        if flow.source not in game_sigs or flow.target not in game_sigs:
            continue  # input-to-game flows handled by T-006

        src_y = game_sigs[flow.source][1]  # codomain (Y)
        tgt_x = game_sigs[flow.target][0]  # domain (X)

        if not src_y or not tgt_x:
            findings.append(
                Finding(
                    check_id="T-001",
                    severity=Severity.ERROR,
                    message=(
                        f"Cannot verify domain/codomain: "
                        f"{flow.source} Y={src_y!r}, {flow.target} X={tgt_x!r}"
                    ),
                    source_elements=[flow.source, flow.target],
                    passed=False,
                )
            )
            continue

        compatible = tokens_subset(flow.label, src_y) or tokens_subset(
            flow.label, tgt_x
        )
        findings.append(
            Finding(
                check_id="T-001",
                severity=Severity.ERROR,
                message=(
                    f"Flow {flow.label!r}: {flow.source} "
                    f"Y={src_y!r} -> {flow.target} X={tgt_x!r}"
                    + ("" if compatible else " — MISMATCH")
                ),
                source_elements=[flow.source, flow.target],
                passed=compatible,
            )
        )

    return findings

check_t002_signature_completeness(pattern)

T-002: Every OpenGameIR must have all four (X,Y,R,S) slots.

Slots must be defined (even if empty set).

A game is valid if it has at least one non-empty input slot (X or R) and at least one non-empty output slot (Y or S). Games that only produce contravariant output (utility computations) have empty Y but non-empty S, which is valid.

Source code in packages/gds-games/ogs/verification/type_checks.py

def check_t002_signature_completeness(pattern: PatternIR) -> list[Finding]:
    """T-002: Every OpenGameIR must have all four (X,Y,R,S) slots.

    Slots must be defined (even if empty set).

    A game is valid if it has at least one non-empty input slot (X or R)
    and at least one non-empty output slot (Y or S). Games that only
    produce contravariant output (utility computations) have empty Y
    but non-empty S, which is valid.
    """
    findings = []
    for game in pattern.games:
        x, y, r, s = game.signature
        has_input = bool(x) or bool(r)
        has_output = bool(y) or bool(s)
        has_required = has_input and has_output

        missing = []
        if not has_input:
            missing.append("no inputs (X or R)")
        if not has_output:
            missing.append("no outputs (Y or S)")

        findings.append(
            Finding(
                check_id="T-002",
                severity=Severity.ERROR,
                message=(
                    f"{game.name}: signature ({x!r}, {y!r}, {r!r}, {s!r})"
                    + (f" — {', '.join(missing)}" if missing else "")
                ),
                source_elements=[game.name],
                passed=has_required,
            )
        )

    return findings

check_t003_flow_type_consistency(pattern)

T-003: Covariant flows must not be utility/coutility; contravariant must be.

Source code in packages/gds-games/ogs/verification/type_checks.py

def check_t003_flow_type_consistency(pattern: PatternIR) -> list[Finding]:
    """T-003: Covariant flows must not be utility/coutility; contravariant must be."""
    findings = []
    for flow in pattern.flows:
        if flow.direction == FlowDirection.COVARIANT:
            ok = flow.flow_type != FlowType.UTILITY_COUTILITY
            findings.append(
                Finding(
                    check_id="T-003",
                    severity=Severity.ERROR,
                    message=(
                        f"Covariant flow {flow.label!r} "
                        f"({flow.source} -> {flow.target})"
                        f" has type {flow.flow_type.value}"
                        + ("" if ok else " — should not be utility/coutility")
                    ),
                    source_elements=[flow.source, flow.target],
                    passed=ok,
                )
            )
        else:
            ok = flow.flow_type == FlowType.UTILITY_COUTILITY
            findings.append(
                Finding(
                    check_id="T-003",
                    severity=Severity.ERROR,
                    message=(
                        f"Contravariant flow {flow.label!r} "
                        f"({flow.source} -> {flow.target})"
                        f" has type {flow.flow_type.value}"
                        + ("" if ok else " — should be utility/coutility")
                    ),
                    source_elements=[flow.source, flow.target],
                    passed=ok,
                )
            )

    return findings

check_t004_input_type_resolution(pattern)

T-004: Every InputIR.schema_hint must resolve to a known type.

Source code in packages/gds-games/ogs/verification/type_checks.py

def check_t004_input_type_resolution(pattern: PatternIR) -> list[Finding]:
    """T-004: Every InputIR.schema_hint must resolve to a known type."""
    findings = []
    for inp in pattern.inputs:
        has_hint = bool(inp.schema_hint)
        findings.append(
            Finding(
                check_id="T-004",
                severity=Severity.WARNING,
                message=(
                    f"Input {inp.name!r}: schema_hint={inp.schema_hint!r}"
                    + ("" if has_hint else " — no schema hint")
                ),
                source_elements=[inp.name],
                passed=has_hint,
            )
        )

    return findings

check_t005_unused_inputs(pattern)

T-005: Flag inputs with no outgoing flows.

Source code in packages/gds-games/ogs/verification/type_checks.py

def check_t005_unused_inputs(pattern: PatternIR) -> list[Finding]:
    """T-005: Flag inputs with no outgoing flows."""
    findings = []
    flow_sources = {f.source for f in pattern.flows}

    for inp in pattern.inputs:
        used = inp.name in flow_sources
        findings.append(
            Finding(
                check_id="T-005",
                severity=Severity.INFO,
                message=(
                    f"Input {inp.name!r}"
                    + ("" if used else " — unused (no outgoing flows)")
                ),
                source_elements=[inp.name],
                passed=used,
            )
        )

    return findings

check_t006_dangling_flows(pattern)

T-006: Flag flows whose source or target is not in the pattern.

Source code in packages/gds-games/ogs/verification/type_checks.py

def check_t006_dangling_flows(pattern: PatternIR) -> list[Finding]:
    """T-006: Flag flows whose source or target is not in the pattern."""
    findings = []
    known_names = {g.name for g in pattern.games} | {i.name for i in pattern.inputs}

    for flow in pattern.flows:
        src_ok = flow.source in known_names
        tgt_ok = flow.target in known_names
        ok = src_ok and tgt_ok

        issues = []
        if not src_ok:
            issues.append(f"source {flow.source!r} unknown")
        if not tgt_ok:
            issues.append(f"target {flow.target!r} unknown")

        findings.append(
            Finding(
                check_id="T-006",
                severity=Severity.ERROR,
                message=(
                    f"Flow {flow.label!r} ({flow.source} -> {flow.target})"
                    + (f" — {', '.join(issues)}" if issues else "")
                ),
                source_elements=[flow.source, flow.target],
                passed=ok,
            )
        )

    return findings

Structural Checks

Structural composition checks S-001 through S-007.

These checks verify that the composition structure of a pattern is well-formed — that games are wired correctly, flows respect their direction, and the overall graph has valid topology. They operate on the flat IR representation (PatternIR) after compilation.

Unlike type checks (T-series), which verify individual flow labels against game signatures, structural checks verify global properties like acyclicity (S-004) and composition-specific invariants.

check_s001_sequential_type_compatibility(pattern)

S-001: In sequential composition G1;G2, verify Y1 = X2.

For each covariant game-to-game flow, the flow label tokens must be a subset of BOTH the source Y and target X tokens. This verifies the structural composition requirement.

Source code in packages/gds-games/ogs/verification/structural_checks.py

def check_s001_sequential_type_compatibility(pattern: PatternIR) -> list[Finding]:
    """S-001: In sequential composition G1;G2, verify Y1 = X2.

    For each covariant game-to-game flow, the flow label tokens must be
    a subset of BOTH the source Y and target X tokens. This verifies
    the structural composition requirement.
    """
    findings = []
    game_sigs = {g.name: g.signature for g in pattern.games}
    game_names = set(game_sigs.keys())

    for flow in pattern.flows:
        if flow.direction != FlowDirection.COVARIANT:
            continue
        if flow.is_corecursive:
            continue  # Corecursive flows are temporal Y→X, not within-step
        if flow.source not in game_names or flow.target not in game_names:
            continue

        src_y = game_sigs[flow.source][1]  # Y (codomain)
        tgt_x = game_sigs[flow.target][0]  # X (domain)

        if not src_y or not tgt_x:
            continue  # T-002 handles missing signatures

        label_in_y = tokens_subset(flow.label, src_y)
        label_in_x = tokens_subset(flow.label, tgt_x)
        compatible = label_in_y and label_in_x

        findings.append(
            Finding(
                check_id="S-001",
                severity=Severity.ERROR,
                message=(
                    f"Sequential {flow.source} ; {flow.target}: "
                    f"Y={src_y!r}, X={tgt_x!r}, flow={flow.label!r}"
                    + ("" if compatible else " — type mismatch")
                ),
                source_elements=[flow.source, flow.target],
                passed=compatible,
            )
        )

    return findings

check_s002_parallel_independence(pattern)

S-002: Games in parallel composition should share no direct flows.

Source code in packages/gds-games/ogs/verification/structural_checks.py

def check_s002_parallel_independence(pattern: PatternIR) -> list[Finding]:
    """S-002: Games in parallel composition should share no direct flows."""
    findings = []

    if pattern.composition_type != CompositionType.PARALLEL:
        findings.append(
            Finding(
                check_id="S-002",
                severity=Severity.WARNING,
                message="Pattern is not parallel composition — S-002 not applicable",
                source_elements=[],
                passed=True,
            )
        )
        return findings

    game_names = {g.name for g in pattern.games}

    violations = [
        f for f in pattern.flows if f.source in game_names and f.target in game_names
    ]

    if violations:
        for flow in violations:
            findings.append(
                Finding(
                    check_id="S-002",
                    severity=Severity.WARNING,
                    message=(
                        f"Parallel independence violation: direct flow {flow.label!r} "
                        f"from {flow.source} to {flow.target}"
                    ),
                    source_elements=[flow.source, flow.target],
                    passed=False,
                )
            )
    else:
        findings.append(
            Finding(
                check_id="S-002",
                severity=Severity.WARNING,
                message=(
                    "Parallel composition: no direct game-to-game flows (independent)"
                ),
                source_elements=[],
                passed=True,
            )
        )

    return findings

check_s003_feedback_type_compatibility(pattern)

S-003: In feedback composition, verify the feedback flow label is consistent with the source's S (coutility) slot.

For each feedback flow, the flow label tokens must be a subset of the source S tokens.

Source code in packages/gds-games/ogs/verification/structural_checks.py

def check_s003_feedback_type_compatibility(pattern: PatternIR) -> list[Finding]:
    """S-003: In feedback composition, verify the feedback flow label is
    consistent with the source's S (coutility) slot.

    For each feedback flow, the flow label tokens must be a subset of
    the source S tokens.
    """
    findings = []
    game_sigs = {g.name: g.signature for g in pattern.games}

    for flow in pattern.flows:
        if not flow.is_feedback:
            continue
        if flow.source not in game_sigs or flow.target not in game_sigs:
            continue

        src_s = game_sigs[flow.source][3]  # S (coutility output)
        tgt_x = game_sigs[flow.target][0]  # X (observation input)

        if not src_s and not tgt_x:
            findings.append(
                Finding(
                    check_id="S-003",
                    severity=Severity.WARNING,
                    message=(
                        f"Feedback {flow.source} -> "
                        f"{flow.target}: both S and X are empty"
                    ),
                    source_elements=[flow.source, flow.target],
                    passed=True,
                )
            )
            continue

        compatible = tokens_subset(flow.label, src_s)

        findings.append(
            Finding(
                check_id="S-003",
                severity=Severity.ERROR,
                message=(
                    f"Feedback {flow.source} -> {flow.target}: "
                    f"S={src_s!r}, X={tgt_x!r}, flow={flow.label!r}"
                    + ("" if compatible else " — type mismatch")
                ),
                source_elements=[flow.source, flow.target],
                passed=compatible,
            )
        )

    return findings

check_s004_covariant_acyclicity(pattern)

S-004: Covariant flow graph must be a DAG (no cycles).

Within a single timestep, covariant data must flow in one direction — cycles would create infinite loops. Corecursive flows (Y→X across timesteps) and feedback flows (S→R, contravariant) are excluded from this check since they represent legitimate temporal or backward links.

Source code in packages/gds-games/ogs/verification/structural_checks.py

def check_s004_covariant_acyclicity(pattern: PatternIR) -> list[Finding]:
    """S-004: Covariant flow graph must be a DAG (no cycles).

    Within a single timestep, covariant data must flow in one direction —
    cycles would create infinite loops. Corecursive flows (Y→X across
    timesteps) and feedback flows (S→R, contravariant) are excluded from
    this check since they represent legitimate temporal or backward links.
    """
    # Build adjacency list from covariant game-to-game flows
    game_names = {g.name for g in pattern.games}
    adj: dict[str, list[str]] = {name: [] for name in game_names}

    for flow in pattern.flows:
        if flow.direction != FlowDirection.COVARIANT:
            continue
        if flow.is_corecursive:
            continue  # Corecursive Y→X flows are temporal, not within-step
        if flow.source in game_names and flow.target in game_names:
            adj[flow.source].append(flow.target)

    # DFS cycle detection with coloring
    WHITE, GRAY, BLACK = 0, 1, 2
    color = {name: WHITE for name in game_names}
    cycle_path: list[str] = []
    has_cycle = False

    def dfs(node: str) -> bool:
        nonlocal has_cycle
        color[node] = GRAY
        cycle_path.append(node)
        for neighbor in adj[node]:
            if color[neighbor] == GRAY:
                # Found cycle — trim path to show only the cycle
                idx = cycle_path.index(neighbor)
                cycle_path[:] = cycle_path[idx:]
                has_cycle = True
                return True
            if color[neighbor] == WHITE and dfs(neighbor):
                return True
        cycle_path.pop()
        color[node] = BLACK
        return False

    for node in game_names:
        if color[node] == WHITE and dfs(node):
            break

    if has_cycle:
        return [
            Finding(
                check_id="S-004",
                severity=Severity.ERROR,
                message=(
                    f"Covariant flow graph contains a cycle: {' -> '.join(cycle_path)}"
                ),
                source_elements=cycle_path,
                passed=False,
            )
        ]

    return [
        Finding(
            check_id="S-004",
            severity=Severity.ERROR,
            message="Covariant flow graph is acyclic (DAG)",
            source_elements=[],
            passed=True,
        )
    ]

check_s005_decision_space_validation(pattern)

S-005: Every decision game must have a non-empty Y (decision output) and at least one incoming contravariant flow (utility feedback).

Source code in packages/gds-games/ogs/verification/structural_checks.py

def check_s005_decision_space_validation(pattern: PatternIR) -> list[Finding]:
    """S-005: Every decision game must have a non-empty Y (decision output)
    and at least one incoming contravariant flow (utility feedback).
    """
    findings = []

    contra_targets = set()
    for flow in pattern.flows:
        if flow.direction == FlowDirection.CONTRAVARIANT:
            contra_targets.add(flow.target)

    for game in pattern.games:
        if game.game_type != GameType.DECISION:
            continue

        y_slot = game.signature[1]
        has_y = bool(y_slot)
        has_contra = game.name in contra_targets

        issues = []
        if not has_y:
            issues.append("empty Y slot (no decision output)")
        if not has_contra:
            issues.append("no incoming contravariant flow (no utility feedback)")

        passed = has_y and has_contra
        findings.append(
            Finding(
                check_id="S-005",
                severity=Severity.WARNING,
                message=(
                    f"Decision game {game.name!r}: Y={y_slot!r}"
                    + (f" — {'; '.join(issues)}" if issues else "")
                ),
                source_elements=[game.name],
                passed=passed,
            )
        )

    return findings

check_s006_corecursive_wiring(pattern)

S-006: Validate corecursive (temporal Y→X) flow wiring.

Corecursive flows must satisfy two invariants: 1. Direction must be covariant (they carry forward data across timesteps, not backward utility). 2. The flow label tokens must be a subset of the source game's Y tokens (the data being forwarded must actually exist in the source's output).

Only runs when corecursive flows are present in the pattern.

Source code in packages/gds-games/ogs/verification/structural_checks.py

def check_s006_corecursive_wiring(pattern: PatternIR) -> list[Finding]:
    """S-006: Validate corecursive (temporal Y→X) flow wiring.

    Corecursive flows must satisfy two invariants:
    1. Direction must be covariant (they carry forward data across timesteps,
       not backward utility).
    2. The flow label tokens must be a subset of the source game's Y tokens
       (the data being forwarded must actually exist in the source's output).

    Only runs when corecursive flows are present in the pattern.
    """
    findings: list[Finding] = []
    game_sigs = {g.name: g.signature for g in pattern.games}

    corecursive_flows = [f for f in pattern.flows if f.is_corecursive]
    if not corecursive_flows:
        return findings

    for flow in corecursive_flows:
        # Must be covariant direction
        if flow.direction != FlowDirection.COVARIANT:
            findings.append(
                Finding(
                    check_id="S-006",
                    severity=Severity.ERROR,
                    message=(
                        f"Corecursive flow {flow.source} → {flow.target}: "
                        f"must be covariant (got {flow.direction.value})"
                    ),
                    source_elements=[flow.source, flow.target],
                    passed=False,
                )
            )
            continue

        # Source Y tokens should overlap with target X tokens
        if flow.source not in game_sigs or flow.target not in game_sigs:
            continue

        src_y = game_sigs[flow.source][1]  # Y
        tgt_x = game_sigs[flow.target][0]  # X

        if not src_y or not tgt_x:
            findings.append(
                Finding(
                    check_id="S-006",
                    severity=Severity.WARNING,
                    message=(
                        f"Corecursive flow {flow.source} → {flow.target}: "
                        f"Y={src_y!r}, X={tgt_x!r} — cannot verify token overlap"
                    ),
                    source_elements=[flow.source, flow.target],
                    passed=True,
                )
            )
            continue

        compatible = tokens_subset(flow.label, src_y)
        findings.append(
            Finding(
                check_id="S-006",
                severity=Severity.ERROR,
                message=(
                    f"Corecursive flow {flow.source} → {flow.target}: "
                    f"Y={src_y!r}, X={tgt_x!r}, flow={flow.label!r}"
                    + ("" if compatible else " — label not in source Y")
                ),
                source_elements=[flow.source, flow.target],
                passed=compatible,
            )
        )

    return findings

check_s007_initialization_completeness(pattern)

S-007: Every initialization input must have at least one outgoing flow.

Source code in packages/gds-games/ogs/verification/structural_checks.py

def check_s007_initialization_completeness(pattern: PatternIR) -> list[Finding]:
    """S-007: Every initialization input must have at least one outgoing flow."""
    findings = []

    flow_sources = {f.source for f in pattern.flows}
    init_inputs = [
        i for i in pattern.inputs if i.input_type == InputType.INITIALIZATION
    ]

    if not init_inputs:
        findings.append(
            Finding(
                check_id="S-007",
                severity=Severity.WARNING,
                message="No initialization inputs found",
                source_elements=[],
                passed=True,
            )
        )
        return findings

    for inp in init_inputs:
        connected = inp.name in flow_sources
        findings.append(
            Finding(
                check_id="S-007",
                severity=Severity.WARNING,
                message=(
                    f"Initialization input {inp.name!r}"
                    + ("" if connected else " — not connected to any game")
                ),
                source_elements=[inp.name],
                passed=connected,
            )
        )

    return findings