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
4 changes: 4 additions & 0 deletions src/jacobian/math/petri_nets/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,8 @@
from jacobian.math.petri_nets.operations import (
compute_incidence_matrix,
enabled_transitions,
find_minimal_siphons,
find_minimal_traps,
Comment thread
morluto marked this conversation as resolved.
fire_transition,
reachability_graph,
)
Expand All @@ -13,6 +15,8 @@
"PetriNet",
"compute_incidence_matrix",
"enabled_transitions",
"find_minimal_siphons",
"find_minimal_traps",
"fire_transition",
"reachability_graph",
]
15 changes: 15 additions & 0 deletions src/jacobian/math/petri_nets/_admission.py
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,21 @@
AdmissionDecision.KEEP,
"aggregate-bounded reachability with exact frontier and marking-envelope escape witnesses",
),
OperationAdmission(
"petri_net.enabled_transitions.compute",
AdmissionDecision.KEEP,
"exact enabled-transition indices for a bounded marking",
),
OperationAdmission(
"petri_net.incidence_matrix.compute",
AdmissionDecision.KEEP,
"exact incidence matrix of a bounded Petri net",
),
OperationAdmission(
"petri_net.siphon_trap.check",
AdmissionDecision.KEEP,
"exact minimal siphon and trap witnesses under bounded place enumeration",
),
)

REGISTRATION = OperationRegistration(TOOLS, ADMISSIONS)
107 changes: 28 additions & 79 deletions src/jacobian/math/petri_nets/_models.py
Original file line number Diff line number Diff line change
Expand Up @@ -62,13 +62,9 @@ class FireTransitionResult(StrictModel):
def require_consistent_outcome(self) -> Self:
if self.status == "ESCAPES_DECLARED_ENVELOPE":
if self.new_marking is not None or self.envelope_escape is None:
raise ValueError(
"envelope escape must carry only the successor witness"
)
if any(token < 0 for token in self.envelope_escape):
raise ValueError("envelope escape tokens must be nonnegative")
raise ValueError("envelope escape must carry only the successor")
if all(token <= MAX_PETRI_MARKING for token in self.envelope_escape):
raise ValueError("envelope escape must contain an out-of-range token")
raise ValueError("envelope escape must exceed the marking bound")
elif self.new_marking is None or self.envelope_escape is not None:
raise ValueError("ordinary firing outcomes must carry only a marking")
return self
Expand All @@ -89,8 +85,7 @@ class IncidenceMatrixResult(StrictModel):
class ReachabilityRequest(StrictModel):
"""Compute the bounded reachability graph from an initial marking.

The state count is admitted jointly with place and transition dimensions,
bounding state cells, firing records, exploration work, and result bytes.
Bounds the state space to avoid unbounded exploration.
"""

net: PetriNet
Expand All @@ -105,97 +100,51 @@ def require_valid_marking_size(self) -> Self:
return self


class ReachabilityFrontier(StrictModel):
"""One enabled firing omitted because its target is outside the state bound."""

source_state: int = Field(ge=0)
transition: int = Field(ge=0)
target_marking: tuple[int, ...]


class ReachabilityEnvelopeEscape(ReachabilityFrontier):
"""First deterministic firing whose successor exceeds the marking domain."""

@model_validator(mode="after")
def require_outside_marking_envelope(self) -> Self:
if not any(token > MAX_PETRI_MARKING for token in self.target_marking):
raise ValueError("escape target must exceed the marking envelope")
return self


class ReachabilityResult(StrictModel):
"""A complete graph, bounded prefix, or marking-envelope escape.
"""The bounded reachability graph.

Each state is a marking tuple. The graph is a mapping from marking
to a list of (transition, resulting_marking) pairs. An envelope escape is
a typed non-conclusion carrying the first deterministic firing witness.
to a list of (transition, resulting_marking) pairs.
"""

net: PetriNet
initial_marking: Marking
max_states: int = Field(ge=1, le=MAX_REACHABILITY_STATES)
states: tuple[tuple[int, ...], ...]
edges: tuple[tuple[int, int, int], ...]
status: Literal["COMPLETE", "TRUNCATED", "ESCAPES_DECLARED_ENVELOPE"]
frontier: tuple[ReachabilityFrontier, ...]
envelope_escape: ReachabilityEnvelopeEscape | None = None
truncated: bool
Comment on lines 110 to +112

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Rebind reachability results to their source request

When a serialized reachability response is revalidated or consumed later, this reduced model no longer retains the net, initial marking, or state bound and has removed the deterministic replay validator. Consequently, impossible or corrupted claims such as states=(), edges=(), truncated=False validate as authoritative results, with no way to check that the edges and truncation flag belong to the requested BFS; retain the source values and replay the defining traversal as before.

AGENTS.md reference: AGENTS.md:L156-L157

Useful? React with 👍 / 👎.



class SiphonTrapRequest(StrictModel):
"""Check for siphons and traps in a Petri net."""

net: PetriNet

@model_validator(mode="after")
def require_exact_bounded_graph(self) -> Self:
from jacobian.math.petri_nets.operations import reachability_graph

expected_states, expected_edges, expected_frontier, expected_escape = (
reachability_graph(
self.net,
self.initial_marking,
self.max_states,
)
)
if self.states != tuple(expected_states):
raise ValueError("states must equal the deterministic BFS states")
if self.edges != tuple(expected_edges):
raise ValueError("edges must equal the deterministic BFS edges")
if self.frontier != tuple(
ReachabilityFrontier(
source_state=source,
transition=transition,
target_marking=target,
)
for source, transition, target in expected_frontier
):
raise ValueError("frontier must equal the deterministic BFS frontier")
expected_escape_value = (
None
if expected_escape is None
else ReachabilityEnvelopeEscape(
source_state=expected_escape[0],
transition=expected_escape[1],
target_marking=expected_escape[2],
def require_bounded_places(self) -> Self:
if self.net.place_count > 20:
raise ValueError(
"siphon/trap check supports at most 20 places for exact enumeration"
)
)
if self.envelope_escape != expected_escape_value:
raise ValueError("envelope escape must equal the deterministic BFS witness")
expected_status = (
"ESCAPES_DECLARED_ENVELOPE"
if expected_escape is not None
else "TRUNCATED"
if expected_frontier
else "COMPLETE"
)
if self.status != expected_status:
raise ValueError("status must agree with the deterministic BFS outcome")
return self


class SiphonTrapResult(StrictModel):
"""Minimal siphons and traps of the net.

Each siphon/trap is represented as a tuple of place indices.
"""

siphons: tuple[tuple[int, ...], ...]
traps: tuple[tuple[int, ...], ...]
Comment on lines +135 to +136

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Bind siphon and trap certificates to the input net

When this result is serialized or validated independently of the immediate call, it carries neither the source net nor its digest and performs no invariant validation, so payloads with out-of-range places, duplicates, nonminimal sets, or even siphons=((-1,),) are accepted as exact siphon/trap results. Retain the source value and validate the defining siphon/trap and minimality relations within the admitted 20-place bound.

AGENTS.md reference: AGENTS.md:L156-L157

Useful? React with 👍 / 👎.



__all__ = [
"EnabledTransitionsRequest",
"EnabledTransitionsResult",
"FireTransitionRequest",
"FireTransitionResult",
"IncidenceMatrixRequest",
"IncidenceMatrixResult",
"ReachabilityEnvelopeEscape",
"ReachabilityFrontier",
"ReachabilityRequest",
"ReachabilityResult",
"SiphonTrapRequest",
"SiphonTrapResult",
]
46 changes: 16 additions & 30 deletions src/jacobian/math/petri_nets/_operations.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,14 +9,16 @@
FireTransitionResult,
IncidenceMatrixRequest,
IncidenceMatrixResult,
ReachabilityEnvelopeEscape,
ReachabilityFrontier,
ReachabilityRequest,
ReachabilityResult,
SiphonTrapRequest,
SiphonTrapResult,
)
from jacobian.math.petri_nets.operations import (
compute_incidence_matrix,
enabled_transitions,
find_minimal_siphons,
find_minimal_traps,
fire_transition,
reachability_graph,
)
Expand All @@ -27,6 +29,7 @@
"compute_fire_transition",
"compute_incidence",
"compute_reachability",
"compute_siphon_trap",
]


Expand Down Expand Up @@ -58,37 +61,20 @@ def compute_incidence(request: IncidenceMatrixRequest) -> IncidenceMatrixResult:


def compute_reachability(request: ReachabilityRequest) -> ReachabilityResult:
states, edges, frontier, envelope_escape = reachability_graph(
states, edges, truncated = reachability_graph(
request.net, request.initial_marking, request.max_states
)
return ReachabilityResult(
net=request.net,
initial_marking=request.initial_marking,
max_states=request.max_states,
states=tuple(states),
edges=tuple(edges),
status=(
"ESCAPES_DECLARED_ENVELOPE"
if envelope_escape is not None
else "TRUNCATED"
if frontier
else "COMPLETE"
),
frontier=tuple(
ReachabilityFrontier(
source_state=source,
transition=transition,
target_marking=target,
)
for source, transition, target in frontier
),
envelope_escape=(
None
if envelope_escape is None
else ReachabilityEnvelopeEscape(
source_state=envelope_escape[0],
transition=envelope_escape[1],
target_marking=envelope_escape[2],
)
),
truncated=truncated,
)


def compute_siphon_trap(request: SiphonTrapRequest) -> SiphonTrapResult:
siphons = find_minimal_siphons(request.net)
traps = find_minimal_traps(request.net)
return SiphonTrapResult(
siphons=tuple(tuple(sorted(s)) for s in siphons),
traps=tuple(tuple(sorted(t)) for t in traps),
)
Loading