Skip to content
Closed
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
47 changes: 46 additions & 1 deletion src/jacobian/contracts/graph_invariant_operations.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,8 +4,9 @@

from typing import Annotated, Literal, Self

from pydantic import Field, StrictBool, StrictInt, model_validator
from pydantic import Field, StrictBool, StrictInt, StrictStr, model_validator

from jacobian.contracts.common import Sha256Digest
from jacobian.contracts.graph_coloring import ChromaticGraph, GraphVertex
from jacobian.contracts.graph_optimization import (
OptimizationSearchStep,
Expand All @@ -19,6 +20,50 @@ class GraphInvariantRequest(ContractModel):
graph: ChromaticGraph


class Graph6DecodeRequest(ContractModel):
graph6: StrictStr = Field(min_length=1, max_length=352)


class Graph6Edge(ContractModel):
first: StrictInt = Field(ge=0, le=62)
second: StrictInt = Field(ge=0, le=62)

@model_validator(mode="after")
def require_canonical_endpoints(self) -> Self:
if self.first >= self.second:
raise ValueError("graph6 edge endpoints must be strictly increasing")
return self


class Graph6DecodeResult(ContractModel):
graph6: StrictStr = Field(min_length=1, max_length=352)
order: StrictInt = Field(ge=0, le=62)
edges: tuple[Graph6Edge, ...] = Field(max_length=1891)
degrees: tuple[StrictInt, ...] = Field(max_length=62)
graph_digest: Sha256Digest
format: Literal["GRAPH6_SMALL_ORDER"] = "GRAPH6_SMALL_ORDER"
bit_order: Literal["COLUMN_MAJOR_UPPER_TRIANGLE"] = "COLUMN_MAJOR_UPPER_TRIANGLE"
exactness: Literal["EXACT_BINARY_DECODE"] = "EXACT_BINARY_DECODE"
verification: Literal["UNVERIFIED"] = "UNVERIFIED"

@model_validator(mode="after")
def bind_dimensions(self) -> Self:
if len(self.degrees) != self.order:
raise ValueError("graph6 degree sequence must match graph order")
pairs = tuple((edge.first, edge.second) for edge in self.edges)
if pairs != tuple(sorted(pairs)) or len(pairs) != len(set(pairs)):
raise ValueError("graph6 edges must be unique and sorted")
if any(edge.second >= self.order for edge in self.edges):
raise ValueError("graph6 edge endpoint exceeds graph order")
expected = [0] * self.order
for edge in self.edges:
expected[edge.first] += 1
expected[edge.second] += 1
if tuple(expected) != self.degrees:
raise ValueError("graph6 degree sequence does not match edges")
return self


class GraphMaximumMatchingGraph(ChromaticGraph):
"""A simple graph bounded for the polynomial-time matching capability."""

Expand Down
26 changes: 26 additions & 0 deletions src/jacobian/domains/graph_optimization/checkers.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@

from jacobian.checker_operations import ExactReplayCheckerDeclaration
from jacobian.contracts.graph_invariant_operations import (
Graph6DecodeRequest,
GraphInvariantRequest,
GraphMaximumMatchingRequest,
)
Expand Down Expand Up @@ -185,6 +186,31 @@
"tutte-berge",
),
),
ExactReplayCheckerDeclaration(
"graph.encoding.graph6.decode.compute",
Graph6DecodeRequest,
"check_graph6_decode",
"graph.graph6-decode.standard-library-v1",
entrypoint_module=_GRAPH_ENTRYPOINT,
replay_method="standard-library graph6 bitstream replay",
reason=(
"operator-authorized standard-library checker independently decodes "
"the graph6 bitstream without importing the producer"
),
verification_capability_id="graph.encoding.graph6.decode.verify",
verification_title="Verify a canonical graph6 decode",
verification_description=(
"Independently replay the small-order graph6 header, upper-triangle "
"bits, padding, sorted edges, degrees, and canonical graph digest."
),
verification_tags=(
"verification",
"exact",
"graph",
"encoding",
"graph6",
),
),
)

GRAPH_SEARCH_EXACT_REPLAY_CHECKERS = GRAPH_OPTIMIZATION_EXACT_REPLAY_CHECKERS[:3]
Expand Down
4 changes: 3 additions & 1 deletion src/jacobian/domains/graph_optimization/invariant_bundle.py
Original file line number Diff line number Diff line change
Expand Up @@ -25,11 +25,12 @@ def build_graph_invariant_bundle() -> DomainBundle:
schema_namespace="jacobian.graph-invariants",
semantics=DomainSemantics(
name="jacobian.finite-simple-graph-invariants",
version="1",
version="2",
definition={
"graph_class": "finite simple undirected",
"maximum_order": 32,
"maximum_edges": 496,
"graph6_decode_maximum_order": 62,
"exact_computations": [
"girth",
"diameter",
Expand All @@ -38,6 +39,7 @@ def build_graph_invariant_bundle() -> DomainBundle:
"is_eulerian",
"spanning_tree_count",
"maximum_matching",
"small-order graph6 decoding",
],
"spanning_tree_arithmetic": "exact SymPy integer determinant",
"assurance": "computed; no producer result is independently verified",
Expand Down
94 changes: 94 additions & 0 deletions src/jacobian/domains/graph_optimization/invariants.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,13 +5,18 @@
import math
import time
from collections.abc import Callable
from hashlib import sha256
from typing import Any, cast

from jacobian.canonical import canonicalize_json
from jacobian.contracts.capabilities import (
CapabilityDiagnostic,
CapabilityInvocationExample,
)
from jacobian.contracts.graph_invariant_operations import (
Graph6DecodeRequest,
Graph6DecodeResult,
Graph6Edge,
GraphCardinalityMaximumObligation,
GraphCliqueNumberResult,
GraphCoreRequest,
Expand Down Expand Up @@ -68,6 +73,74 @@
)


def _decode_graph6(request: Graph6DecodeRequest) -> ComputedOutcome[Graph6DecodeResult]:
value = request.graph6
if value.startswith(">>graph6<<"):
value = value[10:]
if not value or value[0] in {":", "&"}:
return ComputedNotApplicable(
CapabilityDiagnostic(
code="GRAPH6_FORMAT_UNSUPPORTED",
stage="graph6_decoding",
message="Only graph6 is supported; sparse6 and digraph6 are rejected.",
)
)
codes = [ord(character) - 63 for character in value]
if any(code < 0 or code > 63 for code in codes) or codes[0] == 63:
return ComputedNotApplicable(
CapabilityDiagnostic(
code="GRAPH6_ENCODING_INVALID",
stage="graph6_decoding",
message="The graph6 payload is malformed or uses an extended header.",
)
)
order = codes[0]
bit_count = order * (order - 1) // 2
expected_characters = (bit_count + 5) // 6
if len(codes) != 1 + expected_characters:
return ComputedNotApplicable(
CapabilityDiagnostic(
code="GRAPH6_LENGTH_INVALID",
stage="graph6_decoding",
message="The graph6 payload length does not match its order header.",
)
)
bits = [(code >> shift) & 1 for code in codes[1:] for shift in range(5, -1, -1)]
if any(bits[bit_count:]):
return ComputedNotApplicable(
CapabilityDiagnostic(
code="GRAPH6_PADDING_INVALID",
stage="graph6_decoding",
message="Unused graph6 padding bits must be zero.",
)
)
pairs = [(first, second) for second in range(1, order) for first in range(second)]
edges = tuple(
Graph6Edge(first=first, second=second)
for first, second in sorted(
pair for pair, bit in zip(pairs, bits, strict=False) if bit
)
)
degrees = [0] * order
for edge in edges:
degrees[edge.first] += 1
degrees[edge.second] += 1
canonical_graph = {
"order": order,
"edges": [[edge.first, edge.second] for edge in edges],
}
return ComputedSuccess(
Graph6DecodeResult(
graph6=value,
order=order,
edges=edges,
degrees=tuple(degrees),
graph_digest="sha256:"
+ sha256(canonicalize_json(canonical_graph)).hexdigest(),
)
)


def _computed[
ResultT: ContractModel,
](
Expand Down Expand Up @@ -550,6 +623,27 @@ def _obligation(
)

EXACT_GRAPH_INVARIANT_CAPABILITIES = (
ComputedOperation(
capability_id="graph.encoding.graph6.decode.compute",
title="Decode canonical small-order graph6",
description=(
"Decode a headerless or standard-header graph6 string of order at "
"most 62 using the column-major upper-triangle bit convention, "
"returning sorted edges, degrees, and a canonical graph digest."
),
request_model=Graph6DecodeRequest,
result_model=Graph6DecodeResult,
implementation=_decode_graph6,
relation_id="graph.encoding.graph6.relation",
tags=("graph", "encoding", "graph6", "deterministic", "exact"),
invocation_examples=(
example(
"triangle_graph6",
"Decode the graph6 representation of the triangle graph.",
{"graph6": "Bw"},
),
),
),
_computed(
"graph.distance_matrix.compute",
"All-pairs distance matrix",
Expand Down
2 changes: 1 addition & 1 deletion src/jacobian/domains/number_theory/discrete_logarithm.py
Original file line number Diff line number Diff line change
Expand Up @@ -168,7 +168,7 @@ def _obligation(
"base": 2,
"target": 1,
"modulus": 3,
"resource_budget": {"wall_seconds": 1},
"resource_budget": {"wall_seconds": 5},
},
),
),
Expand Down
57 changes: 57 additions & 0 deletions src/jacobian_checkers/graph_exact_operations.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,8 @@

from __future__ import annotations

import hashlib
import json
import re
from collections import deque
from collections.abc import Callable
Expand All @@ -13,6 +15,60 @@
from jacobian_checkers.bound_artifacts import bound_request


def _graph6_decode(source: dict[str, Any], result: dict[str, Any]) -> bool:
if set(source) != {"graph6"} or not isinstance(source["graph6"], str):
return False
value = source["graph6"]
if value.startswith(">>graph6<<"):
value = value[10:]
if not value or value[0] in {":", "&"}:
raise ValueError("unsupported graph encoding")
codes = [ord(character) - 63 for character in value]
if any(code < 0 or code > 63 for code in codes) or codes[0] == 63:
raise ValueError("malformed or extended graph6 encoding")
order = codes[0]
bit_count = order * (order - 1) // 2
if len(codes) != 1 + (bit_count + 5) // 6:
raise ValueError("graph6 length does not match order")
bits = [(code >> shift) & 1 for code in codes[1:] for shift in range(5, -1, -1)]
if any(bits[bit_count:]):
raise ValueError("graph6 padding bits are nonzero")
pairs = [(first, second) for second in range(1, order) for first in range(second)]
edges = sorted(pair for pair, bit in zip(pairs, bits, strict=False) if bit)
degrees = [0] * order
for first, second in edges:
degrees[first] += 1
degrees[second] += 1
digest_payload = json.dumps(
{"edges": [list(edge) for edge in edges], "order": order},
allow_nan=False,
ensure_ascii=False,
separators=(",", ":"),
sort_keys=True,
).encode()
return result == {
"graph6": value,
"order": order,
"edges": [{"first": first, "second": second} for first, second in edges],
"degrees": degrees,
"graph_digest": "sha256:" + hashlib.sha256(digest_payload).hexdigest(),
"format": "GRAPH6_SMALL_ORDER",
"bit_order": "COLUMN_MAJOR_UPPER_TRIANGLE",
"exactness": "EXACT_BINARY_DECODE",
"verification": "UNVERIFIED",
}


def check_graph6_decode(request: dict[str, Any]) -> dict[str, Any]:
return _run(
request,
operation_id="graph.encoding.graph6.decode.compute",
witness_format="graph.graph6-decode.standard-library-v1",
replay=_graph6_decode,
replay_method="graph6 bitstream replay",
)


def _reject(detail: str) -> dict[str, Any]:
return {
"accepted": False,
Expand Down Expand Up @@ -1299,6 +1355,7 @@ def check_graph_maximum_matching(request: dict[str, Any]) -> dict[str, Any]:


__all__ = [
"check_graph6_decode",
"check_graph_diameter",
"check_graph_distance_matrix",
"check_graph_hamiltonian_path",
Expand Down
56 changes: 56 additions & 0 deletions tests/domain/graph/test_graph_verification.py
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,62 @@ def graph_verification_services(tmp_path: Path) -> Iterator[DomainTestServices]:
yield services


def test_graph6_h24_decode_is_exact_and_independently_replayed(
graph_verification_services: DomainTestServices,
) -> None:
payload = {"graph6": "W{CGW_@?Y??@?@?@_@??@??K_????G??C??B??@????_??B"}
computed = graph_verification_services.core.capabilities.invoke(
CapabilityRequest(
capability_id="graph.encoding.graph6.decode.compute",
input=payload,
)
)
decoded = computed.output["result"]
assert decoded["order"] == 24
assert len(decoded["edges"]) == 30
assert max(decoded["degrees"]) == 3
assert decoded["edges"][:4] == [
{"first": 0, "second": 1},
{"first": 0, "second": 2},
{"first": 0, "second": 3},
{"first": 1, "second": 2},
]

verified = graph_verification_services.core.capabilities.invoke(
CapabilityRequest(
capability_id="graph.encoding.graph6.decode.verify",
input={"input": payload, "candidate": decoded},
)
)
assert verified.execution.status is ExecutionStatus.COMPLETED
assert verified.output["status"] == "VERIFIED"
assert verified.assurance.level is CapabilityAssuranceLevel.VERIFIED


def test_graph6_checker_rejects_wrong_bit_order_result(
graph_verification_services: DomainTestServices,
) -> None:
payload = {"graph6": "Bw"}
computed = graph_verification_services.core.capabilities.invoke(
CapabilityRequest(
capability_id="graph.encoding.graph6.decode.compute",
input=payload,
)
)
forged = deepcopy(computed.output["result"])
forged["edges"] = forged["edges"][:-1]
forged["degrees"] = [2, 1, 1]

rejected = graph_verification_services.core.capabilities.invoke(
CapabilityRequest(
capability_id="graph.encoding.graph6.decode.verify",
input={"input": payload, "candidate": forged},
)
)
assert rejected.output["status"] == "REJECTED"
assert rejected.output["conclusion"] == "UNKNOWN"


def test_induced_tree_result_is_domain_bound_and_independently_replayed(
graph_verification_services: DomainTestServices,
) -> None:
Expand Down
Loading
Loading