Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 2 additions & 4 deletions packages/gds-business/gds_business/supplychain/model.py
Original file line number Diff line number Diff line change
Expand Up @@ -64,13 +64,11 @@ def _validate_structure(self) -> Self:
for s in self.shipments:
if s.source not in node_names:
errors.append(
f"Shipment {s.name!r} source {s.source!r} "
f"is not a declared node"
f"Shipment {s.name!r} source {s.source!r} is not a declared node"
)
if s.target not in node_names:
errors.append(
f"Shipment {s.name!r} target {s.target!r} "
f"is not a declared node"
f"Shipment {s.name!r} target {s.target!r} is not a declared node"
)

# 4. Demand target references a declared node
Expand Down
5 changes: 1 addition & 4 deletions packages/gds-framework/gds/verification/generic_checks.py
Original file line number Diff line number Diff line change
Expand Up @@ -78,10 +78,7 @@ def check_g002_signature_completeness(system: SystemIR) -> list[Finding]:

# BoundaryAction blocks have no inputs by design — only check outputs
is_boundary = block.block_type == "boundary"
if is_boundary:
has_required = has_output
else:
has_required = has_input and has_output
has_required = has_output if is_boundary else has_input and has_output

missing = []
if not has_input:
Expand Down
19 changes: 19 additions & 0 deletions packages/gds-games/ogs/dsl/pattern.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,11 @@
"""

from __future__ import annotations
from typing import TYPE_CHECKING

if TYPE_CHECKING:
from gds.ir.models import SystemIR
from gds.spec import GDSSpec

from pydantic import BaseModel, Field

Expand Down Expand Up @@ -140,3 +145,17 @@ def specialize(
else self.initializations,
source=source if source is not None else self.source,
)

# ── Compilation ─────────────────────────────────────────

def compile(self) -> GDSSpec:
"""Compile this pattern to a GDS specification."""
from ogs.dsl.spec_bridge import compile_pattern_to_spec

return compile_pattern_to_spec(self)

def compile_system(self) -> SystemIR:
"""Compile this pattern to a flat SystemIR for verification + visualization."""
from ogs.dsl.compile import compile_to_ir

return compile_to_ir(self).to_system_ir()
108 changes: 108 additions & 0 deletions packages/gds-games/tests/test_pattern_compile.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,108 @@
"""Tests for Pattern.compile() and Pattern.compile_system() convenience methods."""

from gds.ir.models import SystemIR
from gds.spec import GDSSpec
from gds.types.interface import port
from ogs.dsl.games import CovariantFunction, DecisionGame
from ogs.dsl.pattern import Pattern, PatternInput
from ogs.dsl.types import CompositionType, InputType, Signature


def _sequential_pattern() -> Pattern:
a = CovariantFunction(
name="Transform A",
signature=Signature(
x=(port("Raw Input"),),
y=(port("Intermediate"),),
),
)
b = CovariantFunction(
name="Transform B",
signature=Signature(
x=(port("Intermediate"),),
y=(port("Final Output"),),
),
)
return Pattern(
name="Simple Sequential",
game=a >> b,
composition_type=CompositionType.SEQUENTIAL,
)


def _pattern_with_inputs() -> Pattern:
agent = DecisionGame(
name="Agent",
signature=Signature(
x=(port("Observation"),),
y=(port("Action"),),
r=(port("Reward"),),
),
)
return Pattern(
name="Agent With Input",
game=agent,
inputs=[
PatternInput(
name="External Signal",
input_type=InputType.EXTERNAL_WORLD,
target_game="Agent",
flow_label="Observation",
),
],
composition_type=CompositionType.SEQUENTIAL,
)


class TestPatternCompile:
"""Pattern.compile() returns a GDSSpec."""

def test_returns_gds_spec(self):
spec = _sequential_pattern().compile()
assert isinstance(spec, GDSSpec)

def test_spec_name_matches_pattern(self):
spec = _sequential_pattern().compile()
assert spec.name == "Simple Sequential"

def test_spec_contains_blocks(self):
spec = _sequential_pattern().compile()
assert len(spec.blocks) >= 2

def test_compile_with_inputs(self):
spec = _pattern_with_inputs().compile()
assert isinstance(spec, GDSSpec)
assert "External Signal" in spec.blocks

def test_matches_standalone_function(self):
"""compile() produces the same result as calling the standalone function."""
from ogs.dsl.spec_bridge import compile_pattern_to_spec

pattern = _sequential_pattern()
assert pattern.compile().name == compile_pattern_to_spec(pattern).name


class TestPatternCompileSystem:
"""Pattern.compile_system() returns a SystemIR."""

def test_returns_system_ir(self):
sir = _sequential_pattern().compile_system()
assert isinstance(sir, SystemIR)

def test_system_ir_has_blocks(self):
sir = _sequential_pattern().compile_system()
assert len(sir.blocks) >= 2

def test_compile_system_with_inputs(self):
sir = _pattern_with_inputs().compile_system()
assert isinstance(sir, SystemIR)

def test_matches_standalone_pipeline(self):
"""compile_system() produces the same result as the standalone pipeline."""
from ogs.dsl.compile import compile_to_ir

pattern = _sequential_pattern()
expected = compile_to_ir(pattern).to_system_ir()
result = pattern.compile_system()
assert len(result.blocks) == len(expected.blocks)
assert len(result.wirings) == len(expected.wirings)