Skip to content

gds.blocks

Base

Bases: Tagged, ABC

Abstract base for all Blocks — both atomic and composite.

Every block has a name and an interface describing its boundary ports. Composition operators (>>, |, .feedback(), .loop()) build composite blocks from simpler ones.

Source code in packages/gds-framework/gds/blocks/base.py 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
class Block(Tagged, ABC):
    """Abstract base for all Blocks — both atomic and composite.

    Every block has a ``name`` and an ``interface`` describing its boundary
    ports. Composition operators (``>>``, ``|``, ``.feedback()``, ``.loop()``)
    build composite blocks from simpler ones.
    """

    name: str
    interface: Interface = Interface()

    @abstractmethod
    def flatten(self) -> list[AtomicBlock]:
        """Return all atomic blocks in evaluation order."""

    def __rshift__(self, other: Block) -> StackComposition:
        """``a >> b`` — stack (sequential) composition."""
        from gds.blocks.composition import StackComposition

        return StackComposition(
            name=f"{self.name} >> {other.name}",
            first=self,
            second=other,
        )

    def __or__(self, other: Block) -> ParallelComposition:
        """``a | b`` — parallel composition."""
        from gds.blocks.composition import ParallelComposition

        return ParallelComposition(
            name=f"{self.name} | {other.name}",
            left=self,
            right=other,
        )

    def feedback(self, wiring: list[Wiring]) -> FeedbackLoop:
        """Wrap with backward feedback within a single timestep."""
        from gds.blocks.composition import FeedbackLoop

        return FeedbackLoop(
            name=f"{self.name} [feedback]",
            inner=self,
            feedback_wiring=wiring,
        )

    def loop(self, wiring: list[Wiring], exit_condition: str = "") -> TemporalLoop:
        """Wrap with forward temporal iteration across timesteps."""
        from gds.blocks.composition import TemporalLoop

        return TemporalLoop(
            name=f"{self.name} [loop]",
            inner=self,
            temporal_wiring=wiring,
            exit_condition=exit_condition,
        )

flatten() abstractmethod

Return all atomic blocks in evaluation order.

Source code in packages/gds-framework/gds/blocks/base.py 
38
39
40
@abstractmethod
def flatten(self) -> list[AtomicBlock]:
    """Return all atomic blocks in evaluation order."""

__rshift__(other)

a >> b — stack (sequential) composition.

Source code in packages/gds-framework/gds/blocks/base.py 
42
43
44
45
46
47
48
49
50
def __rshift__(self, other: Block) -> StackComposition:
    """``a >> b`` — stack (sequential) composition."""
    from gds.blocks.composition import StackComposition

    return StackComposition(
        name=f"{self.name} >> {other.name}",
        first=self,
        second=other,
    )

__or__(other)

a | b — parallel composition.

Source code in packages/gds-framework/gds/blocks/base.py 
52
53
54
55
56
57
58
59
60
def __or__(self, other: Block) -> ParallelComposition:
    """``a | b`` — parallel composition."""
    from gds.blocks.composition import ParallelComposition

    return ParallelComposition(
        name=f"{self.name} | {other.name}",
        left=self,
        right=other,
    )

feedback(wiring)

Wrap with backward feedback within a single timestep.

Source code in packages/gds-framework/gds/blocks/base.py 
62
63
64
65
66
67
68
69
70
def feedback(self, wiring: list[Wiring]) -> FeedbackLoop:
    """Wrap with backward feedback within a single timestep."""
    from gds.blocks.composition import FeedbackLoop

    return FeedbackLoop(
        name=f"{self.name} [feedback]",
        inner=self,
        feedback_wiring=wiring,
    )

loop(wiring, exit_condition='')

Wrap with forward temporal iteration across timesteps.

Source code in packages/gds-framework/gds/blocks/base.py 
72
73
74
75
76
77
78
79
80
81
def loop(self, wiring: list[Wiring], exit_condition: str = "") -> TemporalLoop:
    """Wrap with forward temporal iteration across timesteps."""
    from gds.blocks.composition import TemporalLoop

    return TemporalLoop(
        name=f"{self.name} [loop]",
        inner=self,
        temporal_wiring=wiring,
        exit_condition=exit_condition,
    )

Bases: Block

Base class for non-decomposable (leaf) blocks.

Domain packages subclass this to define their own atomic block types.

Source code in packages/gds-framework/gds/blocks/base.py 
84
85
86
87
88
89
90
91
class AtomicBlock(Block):
    """Base class for non-decomposable (leaf) blocks.

    Domain packages subclass this to define their own atomic block types.
    """

    def flatten(self) -> list[AtomicBlock]:
        return [self]

Composition

Bases: Block

a >> b — stack (sequential) composition.

Output of the first block feeds input of the second. If no explicit wiring is provided, the validator checks that forward_out tokens overlap with forward_in tokens.

Source code in packages/gds-framework/gds/blocks/composition.py 
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
class StackComposition(Block):
    """``a >> b`` — stack (sequential) composition.

    Output of the first block feeds input of the second. If no explicit
    ``wiring`` is provided, the validator checks that forward_out tokens
    overlap with forward_in tokens.
    """

    first: Block
    second: Block
    wiring: list[Wiring] = Field(default_factory=list)

    @model_validator(mode="after")
    def _compute_interface_and_validate(self) -> Self:
        if not self.wiring:
            first_out_tokens = _collect_tokens(self.first.interface.forward_out)
            second_in_tokens = _collect_tokens(self.second.interface.forward_in)

            if (
                first_out_tokens
                and second_in_tokens
                and not (first_out_tokens & second_in_tokens)
            ):
                raise GDSTypeError(
                    f"Stack composition {self.name!r}: "
                    f"first.forward_out tokens {first_out_tokens} have no overlap with "
                    f"second.forward_in tokens {second_in_tokens}"
                )

        self.interface = Interface(
            forward_in=self.first.interface.forward_in
            + self.second.interface.forward_in,
            forward_out=self.first.interface.forward_out
            + self.second.interface.forward_out,
            backward_in=self.first.interface.backward_in
            + self.second.interface.backward_in,
            backward_out=self.first.interface.backward_out
            + self.second.interface.backward_out,
        )
        return self

    def flatten(self) -> list[AtomicBlock]:
        return self.first.flatten() + self.second.flatten()

Bases: Block

a | b — parallel composition: blocks run independently.

Source code in packages/gds-framework/gds/blocks/composition.py 
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
class ParallelComposition(Block):
    """``a | b`` — parallel composition: blocks run independently."""

    left: Block
    right: Block

    @model_validator(mode="after")
    def _compute_interface(self) -> Self:
        self.interface = Interface(
            forward_in=self.left.interface.forward_in + self.right.interface.forward_in,
            forward_out=self.left.interface.forward_out
            + self.right.interface.forward_out,
            backward_in=self.left.interface.backward_in
            + self.right.interface.backward_in,
            backward_out=self.left.interface.backward_out
            + self.right.interface.backward_out,
        )
        return self

    def flatten(self) -> list[AtomicBlock]:
        return self.left.flatten() + self.right.flatten()

Bases: Block

Backward feedback within a single timestep (backward_out -> backward_in).

Source code in packages/gds-framework/gds/blocks/composition.py 
105
106
107
108
109
110
111
112
113
114
115
116
117
class FeedbackLoop(Block):
    """Backward feedback within a single timestep (backward_out -> backward_in)."""

    inner: Block
    feedback_wiring: list[Wiring]

    @model_validator(mode="after")
    def _validate_and_compute_interface(self) -> Self:
        self.interface = self.inner.interface
        return self

    def flatten(self) -> list[AtomicBlock]:
        return self.inner.flatten()

Bases: Block

Forward temporal iteration across timesteps (forward_out -> forward_in).

All temporal wiring must be covariant direction.

Source code in packages/gds-framework/gds/blocks/composition.py 
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
class TemporalLoop(Block):
    """Forward temporal iteration across timesteps (forward_out -> forward_in).

    All temporal wiring must be covariant direction.
    """

    inner: Block
    temporal_wiring: list[Wiring]
    exit_condition: str = ""

    @model_validator(mode="after")
    def _validate_and_compute_interface(self) -> Self:
        for w in self.temporal_wiring:
            if w.direction != FlowDirection.COVARIANT:
                raise GDSTypeError(
                    f"TemporalLoop {self.name!r}: temporal wiring "
                    f"{w.source_block}.{w.source_port} → "
                    f"{w.target_block}.{w.target_port} "
                    f"must be COVARIANT (got {w.direction.value})"
                )

        self.interface = self.inner.interface
        return self

    def flatten(self) -> list[AtomicBlock]:
        return self.inner.flatten()

Bases: BaseModel

An explicit connection between two blocks.

Covariant wirings (the default) carry data forward; contravariant wirings carry feedback backward.

Source code in packages/gds-framework/gds/blocks/composition.py 
23
24
25
26
27
28
29
30
31
32
33
34
class Wiring(BaseModel, frozen=True):
    """An explicit connection between two blocks.

    Covariant wirings (the default) carry data forward; contravariant
    wirings carry feedback backward.
    """

    source_block: str
    source_port: str
    target_block: str
    target_port: str
    direction: FlowDirection = FlowDirection.COVARIANT

Roles

Bases: AtomicBlock

Exogenous input — enters the system from outside.

In GDS terms: part of the admissible input set U. Boundary actions model external agents, oracles, user inputs, environmental signals — anything the system doesn't control.

Enforces forward_in = () since boundary actions receive no internal forward signals.

Source code in packages/gds-framework/gds/blocks/roles.py 
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
class BoundaryAction(AtomicBlock):
    """Exogenous input — enters the system from outside.

    In GDS terms: part of the admissible input set U.
    Boundary actions model external agents, oracles, user inputs,
    environmental signals — anything the system doesn't control.

    Enforces ``forward_in = ()`` since boundary actions receive no
    internal forward signals.
    """

    kind: str = "boundary"
    options: list[str] = Field(default_factory=list)
    params_used: list[str] = Field(default_factory=list)
    constraints: list[str] = Field(default_factory=list)

    @model_validator(mode="after")
    def _enforce_no_forward_in(self) -> Self:
        if self.interface.forward_in:
            raise GDSCompositionError(
                f"BoundaryAction {self.name!r}: forward_in must be empty "
                f"(boundary actions receive no internal forward signals)"
            )
        return self

Bases: AtomicBlock

Decision logic — maps signals to mechanism inputs.

Policies select from feasible actions. Named options support scenario analysis and A/B testing.

In GDS terms: policies implement the decision mapping within the admissibility constraint.

Source code in packages/gds-framework/gds/blocks/roles.py 
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
class Policy(AtomicBlock):
    """Decision logic — maps signals to mechanism inputs.

    Policies select from feasible actions. Named options support
    scenario analysis and A/B testing.

    In GDS terms: policies implement the decision mapping
    within the admissibility constraint.
    """

    kind: str = "policy"
    options: list[str] = Field(default_factory=list)
    params_used: list[str] = Field(default_factory=list)
    constraints: list[str] = Field(default_factory=list)

Bases: AtomicBlock

Endogenous control — reads state, emits control signals.

These are internal feedback loops: the system observing itself and generating signals that influence downstream policy/mechanism blocks.

Source code in packages/gds-framework/gds/blocks/roles.py 
78
79
80
81
82
83
84
85
86
87
88
class ControlAction(AtomicBlock):
    """Endogenous control — reads state, emits control signals.

    These are internal feedback loops: the system observing itself
    and generating signals that influence downstream policy/mechanism blocks.
    """

    kind: str = "control"
    options: list[str] = Field(default_factory=list)
    params_used: list[str] = Field(default_factory=list)
    constraints: list[str] = Field(default_factory=list)

Bases: AtomicBlock

State update — the only block type that writes to state.

Mechanisms are the atomic state transitions that compose into h. They have no backward ports (state writes don't propagate signals).

updates lists (entity_name, variable_name) pairs specifying which state variables this mechanism modifies.

Source code in packages/gds-framework/gds/blocks/roles.py 
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
class Mechanism(AtomicBlock):
    """State update — the only block type that writes to state.

    Mechanisms are the atomic state transitions that compose into h.
    They have no backward ports (state writes don't propagate signals).

    ``updates`` lists (entity_name, variable_name) pairs specifying
    which state variables this mechanism modifies.
    """

    kind: str = "mechanism"
    updates: list[tuple[str, str]] = Field(default_factory=list)
    params_used: list[str] = Field(default_factory=list)
    constraints: list[str] = Field(default_factory=list)

    @model_validator(mode="after")
    def _enforce_no_backward(self) -> Self:
        if self.interface.backward_in or self.interface.backward_out:
            raise GDSCompositionError(
                f"Mechanism {self.name!r}: backward ports must be empty "
                f"(mechanisms write state, they don't pass backward signals)"
            )
        return self

Errors

Bases: Exception

Base class for all GDS errors.

Source code in packages/gds-framework/gds/blocks/errors.py 
8
9
class GDSError(Exception):
    """Base class for all GDS errors."""

Bases: GDSError

Port type mismatch or invalid port structure during construction.

Source code in packages/gds-framework/gds/blocks/errors.py 
12
13
class GDSTypeError(GDSError):
    """Port type mismatch or invalid port structure during construction."""

Bases: GDSError

Invalid composition structure.

Source code in packages/gds-framework/gds/blocks/errors.py 
16
17
class GDSCompositionError(GDSError):
    """Invalid composition structure."""