gds.verification

Engine

Run verification checks against a SystemIR.

Parameters:

Name	Type	Description	Default
system	SystemIR	The system to verify.	required
checks	list[Callable[[SystemIR], list[Finding]]] | None	Optional subset of checks. Defaults to all generic checks.	None

Source code in packages/gds-framework/gds/verification/engine.py

def verify(
    system: SystemIR,
    checks: list[Callable[[SystemIR], list[Finding]]] | None = None,
) -> VerificationReport:
    """Run verification checks against a SystemIR.

    Args:
        system: The system to verify.
        checks: Optional subset of checks. Defaults to all generic checks.
    """
    checks = checks or ALL_CHECKS
    findings: list[Finding] = []
    for check_fn in checks:
        findings.extend(check_fn(system))
    return VerificationReport(system_name=system.name, findings=findings)

Findings

Bases: BaseModel

A single verification check result — pass or fail with context.

Source code in packages/gds-framework/gds/verification/findings.py

class Finding(BaseModel):
    """A single verification check result — pass or fail with context."""

    check_id: str
    severity: Severity
    message: str
    source_elements: list[str] = Field(default_factory=list)
    passed: bool
    exportable_predicate: str = ""

Bases: StrEnum

Severity level of a verification finding.

Source code in packages/gds-framework/gds/verification/findings.py

class Severity(StrEnum):
    """Severity level of a verification finding."""

    ERROR = "error"
    WARNING = "warning"
    INFO = "info"

Bases: BaseModel

Aggregated verification results for a system.

Source code in packages/gds-framework/gds/verification/findings.py

class VerificationReport(BaseModel):
    """Aggregated verification results for a system."""

    system_name: str
    findings: list[Finding] = Field(default_factory=list)

    @property
    def errors(self) -> int:
        return sum(
            1 for f in self.findings if not f.passed and f.severity == Severity.ERROR
        )

    @property
    def warnings(self) -> int:
        return sum(
            1 for f in self.findings if not f.passed and f.severity == Severity.WARNING
        )

    @property
    def info_count(self) -> int:
        return sum(
            1 for f in self.findings if not f.passed and f.severity == Severity.INFO
        )

    @property
    def checks_passed(self) -> int:
        return sum(1 for f in self.findings if f.passed)

    @property
    def checks_total(self) -> int:
        return len(self.findings)

Generic Checks

Generic verification checks G-001 through G-006.

These checks operate on the domain-neutral SystemIR. They verify type consistency, structural completeness, and graph topology without referencing any domain-specific block types or semantics.

check_g001_domain_codomain_matching(system)

G-001: For every covariant block-to-block wiring, verify the label is consistent with source forward_out or target forward_in.

Source code in packages/gds-framework/gds/verification/generic_checks.py

def check_g001_domain_codomain_matching(system: SystemIR) -> list[Finding]:
    """G-001: For every covariant block-to-block wiring, verify the label
    is consistent with source forward_out or target forward_in.
    """
    findings = []
    block_sigs = {b.name: b.signature for b in system.blocks}

    for wiring in system.wirings:
        if wiring.direction != FlowDirection.COVARIANT:
            continue
        if wiring.source not in block_sigs or wiring.target not in block_sigs:
            continue

        src_out = block_sigs[wiring.source][1]  # forward_out
        tgt_in = block_sigs[wiring.target][0]  # forward_in

        if not src_out or not tgt_in:
            findings.append(
                Finding(
                    check_id="G-001",
                    severity=Severity.ERROR,
                    message=(
                        f"Cannot verify domain/codomain: "
                        f"{wiring.source} out={src_out!r}, "
                        f"{wiring.target} in={tgt_in!r}"
                    ),
                    source_elements=[wiring.source, wiring.target],
                    passed=False,
                )
            )
            continue

        compatible = tokens_subset(wiring.label, src_out) or tokens_subset(
            wiring.label, tgt_in
        )
        findings.append(
            Finding(
                check_id="G-001",
                severity=Severity.ERROR,
                message=(
                    f"Wiring {wiring.label!r}: "
                    f"{wiring.source} out={src_out!r} -> {wiring.target} in={tgt_in!r}"
                    + ("" if compatible else " — MISMATCH")
                ),
                source_elements=[wiring.source, wiring.target],
                passed=compatible,
            )
        )

    return findings

check_g002_signature_completeness(system)

G-002: Every block must have at least one non-empty input slot and at least one non-empty output slot.

Source code in packages/gds-framework/gds/verification/generic_checks.py

def check_g002_signature_completeness(system: SystemIR) -> list[Finding]:
    """G-002: Every block must have at least one non-empty input slot
    and at least one non-empty output slot.
    """
    findings = []
    for block in system.blocks:
        fwd_in, fwd_out, bwd_in, bwd_out = block.signature
        has_input = bool(fwd_in) or bool(bwd_in)
        has_output = bool(fwd_out) or bool(bwd_out)
        has_required = has_input and has_output

        missing = []
        if not has_input:
            missing.append("no inputs")
        if not has_output:
            missing.append("no outputs")

        findings.append(
            Finding(
                check_id="G-002",
                severity=Severity.ERROR,
                message=(
                    f"{block.name}: signature "
                    f"({fwd_in!r}, {fwd_out!r}, "
                    f"{bwd_in!r}, {bwd_out!r})"
                    + (f" — {', '.join(missing)}" if missing else "")
                ),
                source_elements=[block.name],
                passed=has_required,
            )
        )

    return findings

check_g003_direction_consistency(system)

G-003: Validate direction flag consistency and contravariant port-slot matching.

Two validations:

A) Flag consistency — direction, is_feedback, is_temporal must not contradict: - COVARIANT + is_feedback → ERROR (feedback implies contravariant) - CONTRAVARIANT + is_temporal → ERROR (temporal implies covariant)

B) Contravariant port-slot matching — for CONTRAVARIANT wirings, the label must be a token-subset of the source's backward_out (signature[3]) or the target's backward_in (signature[2]). G-001 already covers the covariant side.

Source code in packages/gds-framework/gds/verification/generic_checks.py

def check_g003_direction_consistency(system: SystemIR) -> list[Finding]:
    """G-003: Validate direction flag consistency and contravariant port-slot matching.

    Two validations:

    A) Flag consistency — ``direction``, ``is_feedback``, ``is_temporal`` must
       not contradict:
       - COVARIANT + is_feedback → ERROR (feedback implies contravariant)
       - CONTRAVARIANT + is_temporal → ERROR (temporal implies covariant)

    B) Contravariant port-slot matching — for CONTRAVARIANT wirings, the label
       must be a token-subset of the source's backward_out (signature[3]) or
       the target's backward_in (signature[2]). G-001 already covers the
       covariant side.
    """
    findings = []
    block_sigs = {b.name: b.signature for b in system.blocks}

    for wiring in system.wirings:
        # A) Flag consistency
        if wiring.direction == FlowDirection.COVARIANT and wiring.is_feedback:
            findings.append(
                Finding(
                    check_id="G-003",
                    severity=Severity.ERROR,
                    message=(
                        f"Wiring {wiring.label!r} "
                        f"({wiring.source} -> {wiring.target}): "
                        f"COVARIANT + is_feedback — contradiction"
                    ),
                    source_elements=[wiring.source, wiring.target],
                    passed=False,
                )
            )
            continue

        if wiring.direction == FlowDirection.CONTRAVARIANT and wiring.is_temporal:
            findings.append(
                Finding(
                    check_id="G-003",
                    severity=Severity.ERROR,
                    message=(
                        f"Wiring {wiring.label!r} "
                        f"({wiring.source} -> {wiring.target}): "
                        f"CONTRAVARIANT + is_temporal — contradiction"
                    ),
                    source_elements=[wiring.source, wiring.target],
                    passed=False,
                )
            )
            continue

        # B) Contravariant port-slot matching (G-001 covers covariant)
        if wiring.direction == FlowDirection.CONTRAVARIANT:
            if wiring.source not in block_sigs or wiring.target not in block_sigs:
                # Non-block endpoints — G-004 handles dangling references
                continue

            src_bwd_out = block_sigs[wiring.source][3]  # backward_out
            tgt_bwd_in = block_sigs[wiring.target][2]  # backward_in

            if not src_bwd_out and not tgt_bwd_in:
                findings.append(
                    Finding(
                        check_id="G-003",
                        severity=Severity.ERROR,
                        message=(
                            f"Wiring {wiring.label!r} "
                            f"({wiring.source} -> {wiring.target}): "
                            f"CONTRAVARIANT but both backward "
                            f"ports are empty"
                        ),
                        source_elements=[wiring.source, wiring.target],
                        passed=False,
                    )
                )
                continue

            compatible = tokens_subset(wiring.label, src_bwd_out) or tokens_subset(
                wiring.label, tgt_bwd_in
            )
            findings.append(
                Finding(
                    check_id="G-003",
                    severity=Severity.ERROR,
                    message=(
                        f"Wiring {wiring.label!r}: "
                        f"{wiring.source} bwd_out={src_bwd_out!r} -> "
                        f"{wiring.target} bwd_in={tgt_bwd_in!r}"
                        + ("" if compatible else " — MISMATCH")
                    ),
                    source_elements=[wiring.source, wiring.target],
                    passed=compatible,
                )
            )

    return findings

check_g004_dangling_wirings(system)

G-004: Flag wirings whose source or target is not in the system.

Source code in packages/gds-framework/gds/verification/generic_checks.py

def check_g004_dangling_wirings(system: SystemIR) -> list[Finding]:
    """G-004: Flag wirings whose source or target is not in the system."""
    findings = []
    known_names = {b.name for b in system.blocks}
    for inp in system.inputs:
        known_names.add(inp.name)

    for wiring in system.wirings:
        src_ok = wiring.source in known_names
        tgt_ok = wiring.target in known_names
        ok = src_ok and tgt_ok

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

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

    return findings

check_g005_sequential_type_compatibility(system)

G-005: In stack composition, wiring label must be subset of BOTH source forward_out AND target forward_in.

Source code in packages/gds-framework/gds/verification/generic_checks.py

def check_g005_sequential_type_compatibility(system: SystemIR) -> list[Finding]:
    """G-005: In stack composition, wiring label must be subset of
    BOTH source forward_out AND target forward_in.
    """
    findings = []
    block_sigs = {b.name: b.signature for b in system.blocks}
    block_names = set(block_sigs.keys())

    for wiring in system.wirings:
        if wiring.direction != FlowDirection.COVARIANT:
            continue
        if wiring.is_temporal:
            continue
        if wiring.source not in block_names or wiring.target not in block_names:
            continue

        src_out = block_sigs[wiring.source][1]  # forward_out
        tgt_in = block_sigs[wiring.target][0]  # forward_in

        if not src_out or not tgt_in:
            continue

        label_in_out = tokens_subset(wiring.label, src_out)
        label_in_in = tokens_subset(wiring.label, tgt_in)
        compatible = label_in_out and label_in_in

        findings.append(
            Finding(
                check_id="G-005",
                severity=Severity.ERROR,
                message=(
                    f"Stack {wiring.source} ; {wiring.target}: "
                    f"out={src_out!r}, in={tgt_in!r}, wiring={wiring.label!r}"
                    + ("" if compatible else " — type mismatch")
                ),
                source_elements=[wiring.source, wiring.target],
                passed=compatible,
            )
        )

    return findings

check_g006_covariant_acyclicity(system)

G-006: Covariant flow graph must be a DAG (no cycles within a timestep).

Temporal wirings and contravariant wirings are excluded.

Source code in packages/gds-framework/gds/verification/generic_checks.py

def check_g006_covariant_acyclicity(system: SystemIR) -> list[Finding]:
    """G-006: Covariant flow graph must be a DAG (no cycles within a timestep).

    Temporal wirings and contravariant wirings are excluded.
    """
    block_names = {b.name for b in system.blocks}
    adj: dict[str, list[str]] = {name: [] for name in block_names}

    for wiring in system.wirings:
        if wiring.direction != FlowDirection.COVARIANT:
            continue
        if wiring.is_temporal:
            continue
        if wiring.source in block_names and wiring.target in block_names:
            adj[wiring.source].append(wiring.target)

    # DFS cycle detection
    WHITE, GRAY, BLACK = 0, 1, 2
    color = {name: WHITE for name in block_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:
                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 block_names:
        if color[node] == WHITE and dfs(node):
            break

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

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

Spec Checks

Every entity variable is updated by at least one mechanism.

Detects orphan state variables that can never change — a likely specification error.

Source code in packages/gds-framework/gds/verification/spec_checks.py

def check_completeness(spec: GDSSpec) -> list[Finding]:
    """Every entity variable is updated by at least one mechanism.

    Detects orphan state variables that can never change — a likely
    specification error.
    """
    findings: list[Finding] = []

    all_updates: set[tuple[str, str]] = set()
    for block in spec.blocks.values():
        if isinstance(block, Mechanism):
            for entity_name, var_name in block.updates:
                all_updates.add((entity_name, var_name))

    orphans: list[str] = []
    for entity in spec.entities.values():
        for var_name in entity.variables:
            if (entity.name, var_name) not in all_updates:
                orphans.append(f"{entity.name}.{var_name}")

    if orphans:
        findings.append(
            Finding(
                check_id="SC-001",
                severity=Severity.WARNING,
                message=(
                    f"Orphan state variables never updated by any mechanism: {orphans}"
                ),
                source_elements=orphans,
                passed=False,
            )
        )
    else:
        findings.append(
            Finding(
                check_id="SC-001",
                severity=Severity.INFO,
                message="All state variables are updated by at least one mechanism",
                passed=True,
            )
        )

    return findings

Within each wiring, no two mechanisms update the same variable.

Detects write conflicts where multiple mechanisms try to modify the same state variable within the same composition.

Source code in packages/gds-framework/gds/verification/spec_checks.py

def check_determinism(spec: GDSSpec) -> list[Finding]:
    """Within each wiring, no two mechanisms update the same variable.

    Detects write conflicts where multiple mechanisms try to modify
    the same state variable within the same composition.
    """
    findings: list[Finding] = []

    for wiring in spec.wirings.values():
        update_map: dict[tuple[str, str], list[str]] = defaultdict(list)
        for bname in wiring.block_names:
            block = spec.blocks.get(bname)
            if block is not None and isinstance(block, Mechanism):
                for entity_name, var_name in block.updates:
                    update_map[(entity_name, var_name)].append(bname)

        for (ename, vname), mechs in update_map.items():
            if len(mechs) > 1:
                findings.append(
                    Finding(
                        check_id="SC-002",
                        severity=Severity.ERROR,
                        message=(
                            f"Write conflict in wiring '{wiring.name}': "
                            f"{ename}.{vname} updated by {mechs}"
                        ),
                        source_elements=mechs,
                        passed=False,
                    )
                )

    if not any(f.check_id == "SC-002" for f in findings):
        findings.append(
            Finding(
                check_id="SC-002",
                severity=Severity.INFO,
                message="No write conflicts detected",
                passed=True,
            )
        )

    return findings

Can signals reach from one block to another through wiring?

Maps to GDS attainability correspondence.

Source code in packages/gds-framework/gds/verification/spec_checks.py

def check_reachability(spec: GDSSpec, from_block: str, to_block: str) -> list[Finding]:
    """Can signals reach from one block to another through wiring?

    Maps to GDS attainability correspondence.
    """
    adj: dict[str, set[str]] = defaultdict(set)
    for wiring in spec.wirings.values():
        for wire in wiring.wires:
            adj[wire.source].add(wire.target)

    visited: set[str] = set()
    queue = [from_block]
    reachable = False
    while queue:
        current = queue.pop(0)
        if current == to_block:
            reachable = True
            break
        if current in visited:
            continue
        visited.add(current)
        queue.extend(adj.get(current, set()))

    if reachable:
        return [
            Finding(
                check_id="SC-003",
                severity=Severity.INFO,
                message=f"Block '{from_block}' can reach '{to_block}'",
                source_elements=[from_block, to_block],
                passed=True,
            )
        ]
    return [
        Finding(
            check_id="SC-003",
            severity=Severity.WARNING,
            message=f"Block '{from_block}' cannot reach '{to_block}'",
            source_elements=[from_block, to_block],
            passed=False,
        )
    ]

Wire spaces match source and target block expectations.

Verifies that space references on wires correspond to registered spaces and that source/target blocks are connected to compatible spaces.

Source code in packages/gds-framework/gds/verification/spec_checks.py

def check_type_safety(spec: GDSSpec) -> list[Finding]:
    """Wire spaces match source and target block expectations.

    Verifies that space references on wires correspond to registered
    spaces and that source/target blocks are connected to compatible spaces.
    """
    findings: list[Finding] = []

    for wiring in spec.wirings.values():
        for wire in wiring.wires:
            if wire.space and wire.space not in spec.spaces:
                findings.append(
                    Finding(
                        check_id="SC-004",
                        severity=Severity.ERROR,
                        message=(
                            f"Wire {wire.source} -> {wire.target} references "
                            f"unregistered space '{wire.space}'"
                        ),
                        source_elements=[wire.source, wire.target],
                        passed=False,
                    )
                )

    if not any(f.check_id == "SC-004" for f in findings):
        findings.append(
            Finding(
                check_id="SC-004",
                severity=Severity.INFO,
                message="All wire space references are valid",
                passed=True,
            )
        )

    return findings

All parameter references in blocks resolve to registered parameters.

Validates that every params_used entry on blocks corresponds to a parameter definition in the spec's parameter_schema.

Source code in packages/gds-framework/gds/verification/spec_checks.py

def check_parameter_references(spec: GDSSpec) -> list[Finding]:
    """All parameter references in blocks resolve to registered parameters.

    Validates that every ``params_used`` entry on blocks corresponds to
    a parameter definition in the spec's ``parameter_schema``.
    """
    findings: list[Finding] = []

    param_names = spec.parameter_schema.names()
    unresolved: list[str] = []
    for bname, block in spec.blocks.items():
        if isinstance(block, HasParams):
            for param in block.params_used:
                if param not in param_names:
                    unresolved.append(f"{bname} -> {param}")

    if unresolved:
        findings.append(
            Finding(
                check_id="SC-005",
                severity=Severity.ERROR,
                message=f"Unresolved parameter references: {unresolved}",
                source_elements=unresolved,
                passed=False,
            )
        )
    else:
        findings.append(
            Finding(
                check_id="SC-005",
                severity=Severity.INFO,
                message="All parameter references resolve to registered definitions",
                passed=True,
            )
        )

    return findings

Canonical projection structural validity.

Checks: - SC-006: At least one mechanism exists (f is non-empty) - SC-007: State space X is non-empty (entities with variables exist)

Source code in packages/gds-framework/gds/verification/spec_checks.py

def check_canonical_wellformedness(spec: GDSSpec) -> list[Finding]:
    """Canonical projection structural validity.

    Checks:
    - SC-006: At least one mechanism exists (f is non-empty)
    - SC-007: State space X is non-empty (entities with variables exist)
    """
    findings: list[Finding] = []
    canonical = project_canonical(spec)

    if not canonical.mechanism_blocks:
        findings.append(
            Finding(
                check_id="SC-006",
                severity=Severity.WARNING,
                message="No mechanisms found — state transition f is empty",
                passed=False,
            )
        )
    else:
        findings.append(
            Finding(
                check_id="SC-006",
                severity=Severity.INFO,
                message=(
                    "State transition f has "
                    f"{len(canonical.mechanism_blocks)} mechanism(s)"
                ),
                passed=True,
            )
        )

    if not canonical.state_variables:
        findings.append(
            Finding(
                check_id="SC-007",
                severity=Severity.WARNING,
                message="State space X is empty — no entity variables defined",
                passed=False,
            )
        )
    else:
        findings.append(
            Finding(
                check_id="SC-007",
                severity=Severity.INFO,
                message=(
                    f"State space X has {len(canonical.state_variables)} variable(s)"
                ),
                passed=True,
            )
        )

    return findings