Skip to content

FormalSpecification

Formal specification with temporal properties and constraints.

Overview

The FormalSpecification class defines what a system must do:

  • Properties: Temporal properties (liveness, safety)
  • Invariants: Properties that must always hold
  • Constraints: Hard constraints on resources and performance

Class Documentation

upir.core.specification.FormalSpecification dataclass

A formal specification for distributed system architecture.

Formal specifications capture all requirements and constraints that an architecture must satisfy, including temporal properties, resource constraints, and environmental assumptions.

Based on the TD Commons disclosure, specifications include: - Invariants: Properties that MUST always hold (safety properties) - Properties: Desired properties that should hold (liveness, performance) - Constraints: Resource limits (latency, throughput, cost, etc.) - Assumptions: Environmental conditions assumed to hold

Attributes:

Name Type Description
invariants List[TemporalProperty]

List of temporal properties that must always hold. Violations indicate architectural bugs.
properties List[TemporalProperty]

List of temporal properties that should hold. These are optimization targets.
constraints Dict[str, Dict[str, Any]]

Resource constraints as nested dicts. Format: {"resource_name": {"min": x, "max": y, "equals": z}}
assumptions List[str]

Environmental assumptions (e.g., "network_reliable", "nodes_fail_independently")
Example

spec = FormalSpecification( ... invariants=[ ... TemporalProperty( ... operator=TemporalOperator.ALWAYS, ... predicate="data_consistent" ... ) ... ], ... constraints={ ... "latency": {"max": 100}, ... "cost_per_month": {"max": 10000} ... }, ... assumptions=["network_partitions_rare"] ... )

References: - TD Commons: Formal specification structure - Design by Contract: Invariants and preconditions

Source code in upir/core/specification.py 
 24
 25
 26
 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
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
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
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
@dataclass
class FormalSpecification:
    """
    A formal specification for distributed system architecture.

    Formal specifications capture all requirements and constraints that an
    architecture must satisfy, including temporal properties, resource constraints,
    and environmental assumptions.

    Based on the TD Commons disclosure, specifications include:
    - Invariants: Properties that MUST always hold (safety properties)
    - Properties: Desired properties that should hold (liveness, performance)
    - Constraints: Resource limits (latency, throughput, cost, etc.)
    - Assumptions: Environmental conditions assumed to hold

    Attributes:
        invariants: List of temporal properties that must always hold.
                   Violations indicate architectural bugs.
        properties: List of temporal properties that should hold.
                   These are optimization targets.
        constraints: Resource constraints as nested dicts.
                    Format: {"resource_name": {"min": x, "max": y, "equals": z}}
        assumptions: Environmental assumptions (e.g., "network_reliable",
                    "nodes_fail_independently")

    Example:
        >>> spec = FormalSpecification(
        ...     invariants=[
        ...         TemporalProperty(
        ...             operator=TemporalOperator.ALWAYS,
        ...             predicate="data_consistent"
        ...         )
        ...     ],
        ...     constraints={
        ...         "latency": {"max": 100},
        ...         "cost_per_month": {"max": 10000}
        ...     },
        ...     assumptions=["network_partitions_rare"]
        ... )

    References:
    - TD Commons: Formal specification structure
    - Design by Contract: Invariants and preconditions
    """
    invariants: List[TemporalProperty] = field(default_factory=list)
    properties: List[TemporalProperty] = field(default_factory=list)
    constraints: Dict[str, Dict[str, Any]] = field(default_factory=dict)
    assumptions: List[str] = field(default_factory=list)

    def validate(self) -> bool:
        """
        Validate specification consistency and well-formedness.

        Checks performed:
        1. No duplicate invariants (same operator and predicate)
        2. No duplicate properties
        3. All constraints have valid fields (min, max, or equals)
        4. Constraint values are valid (min <= max if both present)
        5. No empty predicates in temporal properties

        Returns:
            True if specification is valid

        Raises:
            ValueError: If specification has consistency issues

        Example:
            >>> spec = FormalSpecification(
            ...     invariants=[
            ...         TemporalProperty(
            ...             operator=TemporalOperator.ALWAYS,
            ...             predicate="data_consistent"
            ...         )
            ...     ]
            ... )
            >>> spec.validate()
            True
        """
        # Check for duplicate invariants
        seen_invariants = set()
        for inv in self.invariants:
            key = (inv.operator, inv.predicate)
            if key in seen_invariants:
                raise ValueError(
                    f"Duplicate invariant: {inv.operator.value}({inv.predicate})"
                )
            seen_invariants.add(key)

        # Check for duplicate properties
        seen_properties = set()
        for prop in self.properties:
            key = (prop.operator, prop.predicate)
            if key in seen_properties:
                raise ValueError(
                    f"Duplicate property: {prop.operator.value}({prop.predicate})"
                )
            seen_properties.add(key)

        # Validate constraints
        for resource_name, constraint in self.constraints.items():
            if not isinstance(constraint, dict):
                raise ValueError(
                    f"Constraint '{resource_name}' must be a dictionary"
                )

            # Check that constraint has at least one valid field
            valid_fields = {"min", "max", "equals"}
            constraint_fields = set(constraint.keys())
            if not constraint_fields.intersection(valid_fields):
                raise ValueError(
                    f"Constraint '{resource_name}' must have at least one of: "
                    f"min, max, equals. Got: {constraint_fields}"
                )

            # Check that invalid fields are not present
            invalid_fields = constraint_fields - valid_fields
            if invalid_fields:
                raise ValueError(
                    f"Constraint '{resource_name}' has invalid fields: {invalid_fields}. "
                    f"Valid fields are: {valid_fields}"
                )

            # Validate min <= max if both present
            if "min" in constraint and "max" in constraint:
                min_val = constraint["min"]
                max_val = constraint["max"]
                if min_val > max_val:
                    raise ValueError(
                        f"Constraint '{resource_name}': min ({min_val}) > max ({max_val})"
                    )

            # If 'equals' is present, it should be the only field
            if "equals" in constraint and len(constraint) > 1:
                raise ValueError(
                    f"Constraint '{resource_name}': 'equals' cannot be combined "
                    f"with min/max"
                )

        return True

    def to_dict(self) -> Dict[str, Any]:
        """
        Serialize formal specification to JSON-compatible dictionary.

        Uses deterministic ordering (sorted keys) to ensure consistent
        serialization for hashing.

        Returns:
            Dictionary with all specification fields

        Example:
            >>> spec = FormalSpecification(
            ...     constraints={"latency": {"max": 100}}
            ... )
            >>> d = spec.to_dict()
            >>> d["constraints"]["latency"]["max"]
            100
        """
        return {
            "invariants": [inv.to_dict() for inv in self.invariants],
            "properties": [prop.to_dict() for prop in self.properties],
            "constraints": {
                k: dict(v) for k, v in sorted(self.constraints.items())
            },
            "assumptions": sorted(self.assumptions)
        }

    def to_json(self) -> str:
        """
        Serialize formal specification to JSON string.

        Returns:
            JSON string representation

        Example:
            >>> spec = FormalSpecification(
            ...     constraints={"latency": {"max": 100}}
            ... )
            >>> json_str = spec.to_json()
            >>> isinstance(json_str, str)
            True
        """
        return json.dumps(self.to_dict(), indent=2)

    @classmethod
    def from_dict(cls, data: Dict[str, Any]) -> "FormalSpecification":
        """
        Deserialize formal specification from dictionary.

        Args:
            data: Dictionary containing specification fields

        Returns:
            FormalSpecification instance

        Example:
            >>> data = {
            ...     "invariants": [],
            ...     "properties": [],
            ...     "constraints": {"latency": {"max": 100}},
            ...     "assumptions": ["network_reliable"]
            ... }
            >>> spec = FormalSpecification.from_dict(data)
            >>> spec.constraints["latency"]["max"]
            100
        """
        return cls(
            invariants=[
                TemporalProperty.from_dict(inv)
                for inv in data.get("invariants", [])
            ],
            properties=[
                TemporalProperty.from_dict(prop)
                for prop in data.get("properties", [])
            ],
            constraints=data.get("constraints", {}),
            assumptions=data.get("assumptions", [])
        )

    def hash(self) -> str:
        """
        Compute SHA-256 hash of specification for integrity verification.

        Uses deterministic JSON serialization (sorted keys) to ensure
        consistent hashes across runs and platforms.

        Returns:
            Hexadecimal string of SHA-256 hash

        Example:
            >>> spec1 = FormalSpecification(
            ...     constraints={"latency": {"max": 100}}
            ... )
            >>> spec2 = FormalSpecification(
            ...     constraints={"latency": {"max": 100}}
            ... )
            >>> spec1.hash() == spec2.hash()
            True

        References:
        - Python hashlib: https://docs.python.org/3/library/hashlib.html
        - SHA-256: Industry standard cryptographic hash
        """
        # Convert to JSON with sorted keys for deterministic serialization
        spec_dict = self.to_dict()
        json_str = json.dumps(spec_dict, sort_keys=True)

        # Compute SHA-256 hash
        hash_obj = hashlib.sha256(json_str.encode('utf-8'))
        return hash_obj.hexdigest()

    def __str__(self) -> str:
        """Human-readable string representation."""
        parts = []
        if self.invariants:
            parts.append(f"{len(self.invariants)} invariant(s)")
        if self.properties:
            parts.append(f"{len(self.properties)} propertie(s)")
        if self.constraints:
            parts.append(f"{len(self.constraints)} constraint(s)")
        if self.assumptions:
            parts.append(f"{len(self.assumptions)} assumption(s)")

        if not parts:
            return "FormalSpecification(empty)"

        return f"FormalSpecification({', '.join(parts)})"

    def __repr__(self) -> str:
        """Developer-friendly representation."""
        return (
            f"FormalSpecification("
            f"invariants={len(self.invariants)}, "
            f"properties={len(self.properties)}, "
            f"constraints={len(self.constraints)}, "
            f"assumptions={len(self.assumptions)})"
        )

Functions

validate()

Validate specification consistency and well-formedness.

Checks performed: 1. No duplicate invariants (same operator and predicate) 2. No duplicate properties 3. All constraints have valid fields (min, max, or equals) 4. Constraint values are valid (min <= max if both present) 5. No empty predicates in temporal properties

Returns:

Type Description
bool

True if specification is valid

Raises:

Type Description
ValueError

If specification has consistency issues
Example

spec = FormalSpecification( ... invariants=[ ... TemporalProperty( ... operator=TemporalOperator.ALWAYS, ... predicate="data_consistent" ... ) ... ] ... ) spec.validate() True

Source code in upir/core/specification.py 
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
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
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
def validate(self) -> bool:
    """
    Validate specification consistency and well-formedness.

    Checks performed:
    1. No duplicate invariants (same operator and predicate)
    2. No duplicate properties
    3. All constraints have valid fields (min, max, or equals)
    4. Constraint values are valid (min <= max if both present)
    5. No empty predicates in temporal properties

    Returns:
        True if specification is valid

    Raises:
        ValueError: If specification has consistency issues

    Example:
        >>> spec = FormalSpecification(
        ...     invariants=[
        ...         TemporalProperty(
        ...             operator=TemporalOperator.ALWAYS,
        ...             predicate="data_consistent"
        ...         )
        ...     ]
        ... )
        >>> spec.validate()
        True
    """
    # Check for duplicate invariants
    seen_invariants = set()
    for inv in self.invariants:
        key = (inv.operator, inv.predicate)
        if key in seen_invariants:
            raise ValueError(
                f"Duplicate invariant: {inv.operator.value}({inv.predicate})"
            )
        seen_invariants.add(key)

    # Check for duplicate properties
    seen_properties = set()
    for prop in self.properties:
        key = (prop.operator, prop.predicate)
        if key in seen_properties:
            raise ValueError(
                f"Duplicate property: {prop.operator.value}({prop.predicate})"
            )
        seen_properties.add(key)

    # Validate constraints
    for resource_name, constraint in self.constraints.items():
        if not isinstance(constraint, dict):
            raise ValueError(
                f"Constraint '{resource_name}' must be a dictionary"
            )

        # Check that constraint has at least one valid field
        valid_fields = {"min", "max", "equals"}
        constraint_fields = set(constraint.keys())
        if not constraint_fields.intersection(valid_fields):
            raise ValueError(
                f"Constraint '{resource_name}' must have at least one of: "
                f"min, max, equals. Got: {constraint_fields}"
            )

        # Check that invalid fields are not present
        invalid_fields = constraint_fields - valid_fields
        if invalid_fields:
            raise ValueError(
                f"Constraint '{resource_name}' has invalid fields: {invalid_fields}. "
                f"Valid fields are: {valid_fields}"
            )

        # Validate min <= max if both present
        if "min" in constraint and "max" in constraint:
            min_val = constraint["min"]
            max_val = constraint["max"]
            if min_val > max_val:
                raise ValueError(
                    f"Constraint '{resource_name}': min ({min_val}) > max ({max_val})"
                )

        # If 'equals' is present, it should be the only field
        if "equals" in constraint and len(constraint) > 1:
            raise ValueError(
                f"Constraint '{resource_name}': 'equals' cannot be combined "
                f"with min/max"
            )

    return True

to_dict()

Serialize formal specification to JSON-compatible dictionary.

Uses deterministic ordering (sorted keys) to ensure consistent serialization for hashing.

Returns:

Type Description
Dict[str, Any]

Dictionary with all specification fields
Example

spec = FormalSpecification( ... constraints={"latency": {"max": 100}} ... ) d = spec.to_dict() d["constraints"]["latency"]["max"] 100

Source code in upir/core/specification.py 
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
def to_dict(self) -> Dict[str, Any]:
    """
    Serialize formal specification to JSON-compatible dictionary.

    Uses deterministic ordering (sorted keys) to ensure consistent
    serialization for hashing.

    Returns:
        Dictionary with all specification fields

    Example:
        >>> spec = FormalSpecification(
        ...     constraints={"latency": {"max": 100}}
        ... )
        >>> d = spec.to_dict()
        >>> d["constraints"]["latency"]["max"]
        100
    """
    return {
        "invariants": [inv.to_dict() for inv in self.invariants],
        "properties": [prop.to_dict() for prop in self.properties],
        "constraints": {
            k: dict(v) for k, v in sorted(self.constraints.items())
        },
        "assumptions": sorted(self.assumptions)
    }

to_json()

Serialize formal specification to JSON string.

Returns:

Type Description
str

JSON string representation
Example

spec = FormalSpecification( ... constraints={"latency": {"max": 100}} ... ) json_str = spec.to_json() isinstance(json_str, str) True

Source code in upir/core/specification.py 
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
def to_json(self) -> str:
    """
    Serialize formal specification to JSON string.

    Returns:
        JSON string representation

    Example:
        >>> spec = FormalSpecification(
        ...     constraints={"latency": {"max": 100}}
        ... )
        >>> json_str = spec.to_json()
        >>> isinstance(json_str, str)
        True
    """
    return json.dumps(self.to_dict(), indent=2)

from_dict(data) classmethod

Deserialize formal specification from dictionary.

Parameters:

Name Type Description Default
data Dict[str, Any]

Dictionary containing specification fields

 required

Returns:

Type Description
FormalSpecification

FormalSpecification instance
Example

data = { ... "invariants": [], ... "properties": [], ... "constraints": {"latency": {"max": 100}}, ... "assumptions": ["network_reliable"] ... } spec = FormalSpecification.from_dict(data) spec.constraints["latency"]["max"] 100

Source code in upir/core/specification.py 
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
@classmethod
def from_dict(cls, data: Dict[str, Any]) -> "FormalSpecification":
    """
    Deserialize formal specification from dictionary.

    Args:
        data: Dictionary containing specification fields

    Returns:
        FormalSpecification instance

    Example:
        >>> data = {
        ...     "invariants": [],
        ...     "properties": [],
        ...     "constraints": {"latency": {"max": 100}},
        ...     "assumptions": ["network_reliable"]
        ... }
        >>> spec = FormalSpecification.from_dict(data)
        >>> spec.constraints["latency"]["max"]
        100
    """
    return cls(
        invariants=[
            TemporalProperty.from_dict(inv)
            for inv in data.get("invariants", [])
        ],
        properties=[
            TemporalProperty.from_dict(prop)
            for prop in data.get("properties", [])
        ],
        constraints=data.get("constraints", {}),
        assumptions=data.get("assumptions", [])
    )

hash()

Compute SHA-256 hash of specification for integrity verification.

Uses deterministic JSON serialization (sorted keys) to ensure consistent hashes across runs and platforms.

Returns:

Type Description
str

Hexadecimal string of SHA-256 hash
Example

spec1 = FormalSpecification( ... constraints={"latency": {"max": 100}} ... ) spec2 = FormalSpecification( ... constraints={"latency": {"max": 100}} ... ) spec1.hash() == spec2.hash() True

References: - Python hashlib: https://docs.python.org/3/library/hashlib.html - SHA-256: Industry standard cryptographic hash

Source code in upir/core/specification.py 
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
def hash(self) -> str:
    """
    Compute SHA-256 hash of specification for integrity verification.

    Uses deterministic JSON serialization (sorted keys) to ensure
    consistent hashes across runs and platforms.

    Returns:
        Hexadecimal string of SHA-256 hash

    Example:
        >>> spec1 = FormalSpecification(
        ...     constraints={"latency": {"max": 100}}
        ... )
        >>> spec2 = FormalSpecification(
        ...     constraints={"latency": {"max": 100}}
        ... )
        >>> spec1.hash() == spec2.hash()
        True

    References:
    - Python hashlib: https://docs.python.org/3/library/hashlib.html
    - SHA-256: Industry standard cryptographic hash
    """
    # Convert to JSON with sorted keys for deterministic serialization
    spec_dict = self.to_dict()
    json_str = json.dumps(spec_dict, sort_keys=True)

    # Compute SHA-256 hash
    hash_obj = hashlib.sha256(json_str.encode('utf-8'))
    return hash_obj.hexdigest()

__str__()

Human-readable string representation.

Source code in upir/core/specification.py 
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
def __str__(self) -> str:
    """Human-readable string representation."""
    parts = []
    if self.invariants:
        parts.append(f"{len(self.invariants)} invariant(s)")
    if self.properties:
        parts.append(f"{len(self.properties)} propertie(s)")
    if self.constraints:
        parts.append(f"{len(self.constraints)} constraint(s)")
    if self.assumptions:
        parts.append(f"{len(self.assumptions)} assumption(s)")

    if not parts:
        return "FormalSpecification(empty)"

    return f"FormalSpecification({', '.join(parts)})"

__repr__()

Developer-friendly representation.

Source code in upir/core/specification.py 
292
293
294
295
296
297
298
299
300
def __repr__(self) -> str:
    """Developer-friendly representation."""
    return (
        f"FormalSpecification("
        f"invariants={len(self.invariants)}, "
        f"properties={len(self.properties)}, "
        f"constraints={len(self.constraints)}, "
        f"assumptions={len(self.assumptions)})"
    )

Usage Example

from upir.core.specification import FormalSpecification
from upir.core.temporal import TemporalProperty, TemporalOperator

# Define temporal properties
properties = [
    TemporalProperty(
        operator=TemporalOperator.EVENTUALLY,
        predicate="task_complete",
        time_bound=60000  # 60 seconds
    ),
    TemporalProperty(
        operator=TemporalOperator.WITHIN,
        predicate="respond",
        time_bound=100  # 100ms
    )
]

# Define invariants
invariants = [
    TemporalProperty(
        operator=TemporalOperator.ALWAYS,
        predicate="data_consistent"
    ),
    TemporalProperty(
        operator=TemporalOperator.ALWAYS,
        predicate="no_data_loss"
    )
]

# Define constraints
constraints = {
    "latency_p99": {"max": 100.0},
    "monthly_cost": {"max": 5000.0},
    "throughput_qps": {"min": 10000.0}
}

# Create specification
spec = FormalSpecification(
    properties=properties,
    invariants=invariants,
    constraints=constraints
)

# Serialize
spec_json = spec.to_json()

Constraint Schema

Constraints are dictionaries with min/max bounds:

constraints = {
    "metric_name": {
        "min": 100.0,  # Minimum value (optional)
        "max": 1000.0  # Maximum value (optional)
    }
}

See Also