Skip to content

graflo.architecture.profile

Named conformance profiles over a manifest.

A profile is a versioned list of mechanically checkable assertions. It reads only fields the contract already has, introduces no semantics and touches no backend -- so it can run against manifests this package did not author, which is the point.

Eager re-export: the subpackage imports nothing heavier than the contract models, so there is no lazy-facade cost to avoid here.

Modules:

Name Description
context

What a profile check reads, and the vocabularies it recognises.
inverses

Audit of declared inverses across the whole manifest, and its inverses profile.
model

The conformance report, and the waiver document that can excuse part of it.
runner

Profile definition, registry, and the entry points every surface calls.
world_model

The World Model Profile: six mechanically checkable assertions.

Attributes

INVERSES_PROFILE = Profile(name='inverses', version=PROFILE_VERSION, assertions=(Assertion(_CONSISTENT, 'Declared inverses are realized without contradiction', True, check_inverses_consistent), Assertion(_COMPLETE, 'Every realization of an inverse is complete', True, check_inverses_complete))) module-attribute

Severity = Literal['error', 'warning', 'info'] module-attribute

Status = Literal['pass', 'fail', 'warn', 'waived', 'not_applicable'] module-attribute

VocabularyStatus = Literal['live', 'unknown', 'malformed'] module-attribute

WORLD_MODEL_PROFILE = Profile(name='world-model', version=PROFILE_VERSION, assertions=(Assertion(_A1, 'Types are grounded in an external vocabulary', True, check_grounded_types), Assertion(_A2, 'Every vertex declares an identity mode', True, check_declared_identity), Assertion(_A3, 'Every edge declares its directionality', True, check_declared_directionality), Assertion(_A4, 'Every measured property carries a unit', True, check_declared_units), Assertion(_A5, 'Temporal validity is declared or waived', True, check_temporal), Assertion(_A6, 'Provenance is expressible and attached', True, check_provenance))) module-attribute

__all__ = ['INVERSES_PROFILE', 'WORLD_MODEL_PROFILE', 'Assertion', 'AssertionResult', 'CheckContext', 'Finding', 'InverseReport', 'PairStatus', 'PrefixAllowListResolver', 'Profile', 'ProfileReport', 'ProfileWaivers', 'Severity', 'Status', 'VocabularyResolver', 'VocabularyStatus', 'Waiver', 'audit_inverses', 'check_manifest', 'check_manifest_config', 'get_profile', 'list_profiles', 'run_profile'] module-attribute

Classes

Assertion dataclass

One mechanically checkable claim about a manifest.

Source code in graflo/architecture/profile/runner.py 
29
30
31
32
33
34
35
36
@dataclass(frozen=True, slots=True)
class Assertion:
    """One mechanically checkable claim about a manifest."""

    id: str
    title: str
    required: bool
    run: Callable[[CheckContext], AssertionResult]

Attributes

id instance-attribute
required instance-attribute
run instance-attribute
title instance-attribute

Methods:

__init__(id, title, required, run)

AssertionResult

Bases: ConfigBaseModel

One assertion's outcome over the whole manifest.

Source code in graflo/architecture/profile/model.py 
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
class AssertionResult(ConfigBaseModel):
    """One assertion's outcome over the whole manifest."""

    id: str = PydanticField(...)
    title: str = PydanticField(...)
    required: bool = PydanticField(default=True)
    status: Status = PydanticField(...)
    checked: int = PydanticField(
        default=0,
        description=(
            "Elements examined. Zero means the assertion had nothing to say "
            "about this manifest, which is reported as ``not_applicable`` "
            "rather than as a pass."
        ),
    )
    findings: list[Finding] = PydanticField(default_factory=list)
    waiver: Waiver | None = PydanticField(default=None)

Attributes

checked = PydanticField(default=0, description='Elements examined. Zero means the assertion had nothing to say about this manifest, which is reported as ``not_applicable`` rather than as a pass.') class-attribute instance-attribute
findings = PydanticField(default_factory=list) class-attribute instance-attribute
id = PydanticField(...) class-attribute instance-attribute
required = PydanticField(default=True) class-attribute instance-attribute
status = PydanticField(...) class-attribute instance-attribute
title = PydanticField(...) class-attribute instance-attribute
waiver = PydanticField(default=None) class-attribute instance-attribute

CheckContext dataclass

Everything an assertion may read.

Attributes:

Name Type Description
manifest GraphManifest

The parsed, finish_init-ed manifest.
authored Mapping[str, Any] | None

The document the author wrote, when the caller has it. None degrades the two declaration assertions to warnings instead of letting them pass on a normalized model.
waivers ProfileWaivers | None

Operator waivers to apply, if any.
resolver VocabularyResolver

Vocabulary liveness backend.
Source code in graflo/architecture/profile/context.py 
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
@dataclass(slots=True)
class CheckContext:
    """Everything an assertion may read.

    Attributes:
        manifest: The parsed, ``finish_init``-ed manifest.
        authored: The document the author wrote, when the caller has it.
            ``None`` degrades the two declaration assertions to warnings
            instead of letting them pass on a normalized model.
        waivers: Operator waivers to apply, if any.
        resolver: Vocabulary liveness backend.
    """

    manifest: GraphManifest
    authored: Mapping[str, Any] | None = None
    waivers: ProfileWaivers | None = None
    resolver: VocabularyResolver = field(default_factory=PrefixAllowListResolver)

    @property
    def has_authored(self) -> bool:
        """Whether declaration-sensitive assertions can be decided at all."""
        return self.authored is not None

    def authored_vertices(self) -> dict[str, Mapping[str, Any]]:
        """Authored vertex blocks by name; empty when the document is absent."""
        return {
            str(vertex["name"]): vertex
            for vertex in self._authored_path("vertex_config", "vertices")
            if isinstance(vertex, Mapping) and "name" in vertex
        }

    def authored_edges(self) -> list[Mapping[str, Any]]:
        """Authored edge blocks; empty when the document is absent."""
        return [
            edge
            for edge in self._authored_path("edge_config", "edges")
            if isinstance(edge, Mapping)
        ]

    def _authored_path(self, holder: str, key: str) -> list[Any]:
        """``schema.graph.<holder>.<key>`` out of the authored document.

        Two levels carry a serialized alias beside the field name --
        ``schema``/``graph_schema`` on the manifest and ``graph``/``core_schema``
        on the schema -- and authored documents use the aliases while a dumped
        model may use either. Reading only one spelling silently finds nothing,
        which would make the declaration assertions pass on every real file
        instead of failing loudly.
        """
        if self.authored is None:
            return []
        schema = self.authored.get("schema") or self.authored.get("graph_schema")
        if not isinstance(schema, Mapping):
            return []
        core = schema.get("graph") or schema.get("core_schema")
        if not isinstance(core, Mapping):
            return []
        block = core.get(holder)
        if not isinstance(block, Mapping):
            return []
        value = block.get(key)
        return list(value) if isinstance(value, list) else []

Attributes

authored = None class-attribute instance-attribute
has_authored property

Whether declaration-sensitive assertions can be decided at all.

manifest instance-attribute
resolver = field(default_factory=PrefixAllowListResolver) class-attribute instance-attribute
waivers = None class-attribute instance-attribute

Methods:

__init__(manifest, authored=None, waivers=None, resolver=PrefixAllowListResolver())
authored_edges()

Authored edge blocks; empty when the document is absent.

Source code in graflo/architecture/profile/context.py 
173
174
175
176
177
178
179
def authored_edges(self) -> list[Mapping[str, Any]]:
    """Authored edge blocks; empty when the document is absent."""
    return [
        edge
        for edge in self._authored_path("edge_config", "edges")
        if isinstance(edge, Mapping)
    ]
authored_vertices()

Authored vertex blocks by name; empty when the document is absent.

Source code in graflo/architecture/profile/context.py 
165
166
167
168
169
170
171
def authored_vertices(self) -> dict[str, Mapping[str, Any]]:
    """Authored vertex blocks by name; empty when the document is absent."""
    return {
        str(vertex["name"]): vertex
        for vertex in self._authored_path("vertex_config", "vertices")
        if isinstance(vertex, Mapping) and "name" in vertex
    }

Finding

Bases: ConfigBaseModel

One assertion's verdict about one element.

Source code in graflo/architecture/profile/model.py 
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
class Finding(ConfigBaseModel):
    """One assertion's verdict about one element."""

    assertion: str = PydanticField(...)
    status: Status = PydanticField(...)
    severity: Severity = PydanticField(...)
    target: str | None = PydanticField(
        default=None,
        description=(
            "The element this is about, in a stable address form: "
            "``vertex:Observation``, ``vertex:Observation.result_value``, "
            "``edge:Observation-hasFeatureOfInterest->Asset``, or ``manifest``."
        ),
    )
    message: str = PydanticField(
        ..., description="One line, already readable without the detail payload."
    )
    detail: dict[str, Any] = PydanticField(
        default_factory=dict,
        description="Machine payload -- the IRI, the unit token, the identity mode.",
    )

Attributes

assertion = PydanticField(...) class-attribute instance-attribute
detail = PydanticField(default_factory=dict, description='Machine payload -- the IRI, the unit token, the identity mode.') class-attribute instance-attribute
message = PydanticField(..., description='One line, already readable without the detail payload.') class-attribute instance-attribute
severity = PydanticField(...) class-attribute instance-attribute
status = PydanticField(...) class-attribute instance-attribute
target = PydanticField(default=None, description='The element this is about, in a stable address form: ``vertex:Observation``, ``vertex:Observation.result_value``, ``edge:Observation-hasFeatureOfInterest->Asset``, or ``manifest``.') class-attribute instance-attribute

InverseReport

Bases: ConfigBaseModel

Everything known about how a manifest realizes its declared inverses.

Source code in graflo/architecture/profile/inverses.py 
 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
class InverseReport(ConfigBaseModel):
    """Everything known about how a manifest realizes its declared inverses."""

    pairs: list[PairStatus] = PydanticField(default_factory=list)
    symmetric: list[str] = PydanticField(default_factory=list)
    findings: list[InverseFinding] = PydanticField(default_factory=list)

    def repairable(self) -> list[InverseFinding]:
        """Findings a planner may fix by propagation."""
        return [f for f in self.findings if f.severity == "repairable"]

    def conflicts(self) -> list[InverseFinding]:
        """Findings only the author can settle."""
        return [f for f in self.findings if f.severity == "conflict"]

    def notes(self) -> list[InverseFinding]:
        return [f for f in self.findings if f.severity == "note"]

    def pair(self, relation: str) -> PairStatus | None:
        """The status of the pair that ``relation`` belongs to, either side."""
        return next(
            (p for p in self.pairs if relation in (p.relation, p.inverse)), None
        )

    def introduced_since(self, before: InverseReport) -> list[InverseFinding]:
        """Findings of this report that *before* did not have.

        ``audit_inverses(after).introduced_since(audit_inverses(before))`` is what
        a change did to the inverses, whatever produced the change.
        """
        known = {finding.key for finding in before.findings}
        return [finding for finding in self.findings if finding.key not in known]

    def to_lines(self) -> list[str]:
        """The report as text. The one renderer."""
        lines: list[str] = []
        for pair in self.pairs:
            # A mirror count only says something once the inverse is stored as edges.
            stored = pair.state in ("materialized", "partial")
            counts = f" ({pair.mirrored}/{pair.total} mirrored)" if stored else ""
            lines.append(f"{pair.relation} <-> {pair.inverse}: {pair.state}{counts}")
            for resource, feeding in sorted(pair.feeding.items()):
                lines.append(f"    fed in {resource}: {feeding}")
        for name in self.symmetric:
            lines.append(f"{name}: symmetric")
        if not self.pairs and not self.symmetric:
            lines.append("no declared inverses")
        for severity in ("conflict", "repairable", "note"):
            group = [f for f in self.findings if f.severity == severity]
            if not group:
                continue
            lines.append("")
            lines.append(f"{severity} ({len(group)}):")
            for finding in group:
                lines.append(f"  - [{finding.kind}] {finding.message}")
                for step in finding.steps:
                    lines.append(f"      at {step}")
        return lines

Attributes

findings = PydanticField(default_factory=list) class-attribute instance-attribute
pairs = PydanticField(default_factory=list) class-attribute instance-attribute
symmetric = PydanticField(default_factory=list) class-attribute instance-attribute

Methods:

conflicts()

Findings only the author can settle.

Source code in graflo/architecture/profile/inverses.py 
109
110
111
def conflicts(self) -> list[InverseFinding]:
    """Findings only the author can settle."""
    return [f for f in self.findings if f.severity == "conflict"]
introduced_since(before)

Findings of this report that before did not have.

audit_inverses(after).introduced_since(audit_inverses(before)) is what a change did to the inverses, whatever produced the change.

Source code in graflo/architecture/profile/inverses.py 
122
123
124
125
126
127
128
129
def introduced_since(self, before: InverseReport) -> list[InverseFinding]:
    """Findings of this report that *before* did not have.

    ``audit_inverses(after).introduced_since(audit_inverses(before))`` is what
    a change did to the inverses, whatever produced the change.
    """
    known = {finding.key for finding in before.findings}
    return [finding for finding in self.findings if finding.key not in known]
notes()
Source code in graflo/architecture/profile/inverses.py 
113
114
def notes(self) -> list[InverseFinding]:
    return [f for f in self.findings if f.severity == "note"]
pair(relation)

The status of the pair that relation belongs to, either side.

Source code in graflo/architecture/profile/inverses.py 
116
117
118
119
120
def pair(self, relation: str) -> PairStatus | None:
    """The status of the pair that ``relation`` belongs to, either side."""
    return next(
        (p for p in self.pairs if relation in (p.relation, p.inverse)), None
    )
repairable()

Findings a planner may fix by propagation.

Source code in graflo/architecture/profile/inverses.py 
105
106
107
def repairable(self) -> list[InverseFinding]:
    """Findings a planner may fix by propagation."""
    return [f for f in self.findings if f.severity == "repairable"]
to_lines()

The report as text. The one renderer.

Source code in graflo/architecture/profile/inverses.py 
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
def to_lines(self) -> list[str]:
    """The report as text. The one renderer."""
    lines: list[str] = []
    for pair in self.pairs:
        # A mirror count only says something once the inverse is stored as edges.
        stored = pair.state in ("materialized", "partial")
        counts = f" ({pair.mirrored}/{pair.total} mirrored)" if stored else ""
        lines.append(f"{pair.relation} <-> {pair.inverse}: {pair.state}{counts}")
        for resource, feeding in sorted(pair.feeding.items()):
            lines.append(f"    fed in {resource}: {feeding}")
    for name in self.symmetric:
        lines.append(f"{name}: symmetric")
    if not self.pairs and not self.symmetric:
        lines.append("no declared inverses")
    for severity in ("conflict", "repairable", "note"):
        group = [f for f in self.findings if f.severity == severity]
        if not group:
            continue
        lines.append("")
        lines.append(f"{severity} ({len(group)}):")
        for finding in group:
            lines.append(f"  - [{finding.kind}] {finding.message}")
            for step in finding.steps:
                lines.append(f"      at {step}")
    return lines

PairStatus

Bases: PairRealization

A declared pair across schema and ingestion.

Source code in graflo/architecture/profile/inverses.py 
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
class PairStatus(PairRealization):
    """A declared pair across schema and ingestion."""

    feeding: dict[str, Feeding] = PydanticField(
        default_factory=dict,
        description=(
            "Per resource that writes either relation of the pair: how it feeds "
            "the materialized inverse. Empty when nothing is materialized."
        ),
    )
    native_eligibility: list[NativeInverseViolation] | None = PydanticField(
        default=None,
        description=(
            "Rules that would be broken if the database maintained this pair; "
            "empty means eligible. None when the pair is already native."
        ),
    )
    native_candidate: str | None = PydanticField(
        default=None,
        description=(
            "The relation that would be listed in `native_inverses`: the one "
            "that has edges while its inverse has none."
        ),
    )

Attributes

feeding = PydanticField(default_factory=dict, description='Per resource that writes either relation of the pair: how it feeds the materialized inverse. Empty when nothing is materialized.') class-attribute instance-attribute
native_candidate = PydanticField(default=None, description='The relation that would be listed in `native_inverses`: the one that has edges while its inverse has none.') class-attribute instance-attribute
native_eligibility = PydanticField(default=None, description='Rules that would be broken if the database maintained this pair; empty means eligible. None when the pair is already native.') class-attribute instance-attribute

PrefixAllowListResolver dataclass

Recognises :data:KNOWN_NAMESPACES; everything else is unknown.

The honest v0.1 answer to "does this resolve to a live vocabulary": it checks the shape and the namespace and says so, rather than dereferencing anything. unknown is reported as a warning, never as a failure.

Source code in graflo/architecture/profile/context.py 
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
@dataclass(frozen=True, slots=True)
class PrefixAllowListResolver:
    """Recognises :data:`KNOWN_NAMESPACES`; everything else is ``unknown``.

    The honest v0.1 answer to "does this resolve to a *live* vocabulary": it
    checks the shape and the namespace and says so, rather than dereferencing
    anything. ``unknown`` is reported as a warning, never as a failure.
    """

    namespaces: tuple[str, ...] = KNOWN_NAMESPACES

    def resolve(self, iri: str) -> VocabularyStatus:
        if "://" not in iri or iri.startswith("://"):
            return "malformed"
        scheme = iri.split("://", 1)[0]
        if not scheme or not scheme.isascii() or " " in iri:
            return "malformed"
        if any(iri.startswith(ns) for ns in self.namespaces):
            return "live"
        return "unknown"

Attributes

namespaces = KNOWN_NAMESPACES class-attribute instance-attribute

Methods:

__init__(namespaces=KNOWN_NAMESPACES)
resolve(iri)
Source code in graflo/architecture/profile/context.py 
128
129
130
131
132
133
134
135
136
def resolve(self, iri: str) -> VocabularyStatus:
    if "://" not in iri or iri.startswith("://"):
        return "malformed"
    scheme = iri.split("://", 1)[0]
    if not scheme or not scheme.isascii() or " " in iri:
        return "malformed"
    if any(iri.startswith(ns) for ns in self.namespaces):
        return "live"
    return "unknown"

Profile dataclass

A named conformance level.

Source code in graflo/architecture/profile/runner.py 
39
40
41
42
43
44
45
@dataclass(frozen=True, slots=True)
class Profile:
    """A named conformance level."""

    name: str
    version: str
    assertions: tuple[Assertion, ...]

Attributes

assertions instance-attribute
name instance-attribute
version instance-attribute

Methods:

__init__(name, version, assertions)

ProfileReport

Bases: ConfigBaseModel

The result of checking one manifest against one named profile.

Source code in graflo/architecture/profile/model.py 
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
class ProfileReport(ConfigBaseModel):
    """The result of checking one manifest against one named profile."""

    profile: str = PydanticField(...)
    profile_version: str = PydanticField(...)
    graflo_version: str | None = PydanticField(default=None)
    subject: str | None = PydanticField(
        default=None,
        description="What was checked -- a path, or ``uuid@version``. Set by the caller.",
    )
    manifest_hash: str | None = PydanticField(
        default=None,
        description=(
            "Content address of the manifest checked, so a stored report can be "
            "matched back to the artifact that produced it."
        ),
    )
    status: Status = PydanticField(...)
    assertions: list[AssertionResult] = PydanticField(default_factory=list)

    @property
    def ok(self) -> bool:
        """No error-severity finding survived waivers."""
        return not self.errors()

    def errors(self) -> list[Finding]:
        """Every finding that makes the manifest non-conformant."""
        return [
            finding
            for result in self.assertions
            if result.waiver is None
            for finding in result.findings
            if finding.severity == "error"
        ]

    def warnings(self) -> list[Finding]:
        """Every advisory finding, waived assertions included."""
        return [
            finding
            for result in self.assertions
            for finding in result.findings
            if finding.severity == "warning"
        ]

    def to_lines(self) -> list[str]:
        """The report as text. The one renderer -- every surface calls this."""
        marks = {
            "pass": "PASS",
            "fail": "FAIL",
            "warn": "WARN",
            "waived": "WAIVED",
            "not_applicable": "N/A",
        }
        subject = self.subject or "<manifest>"
        lines = [
            f"profile {self.profile} v{self.profile_version} -- {subject}",
            f"  overall: {marks[self.status]}",
            "",
        ]
        for result in self.assertions:
            lines.append(
                f"  [{marks[result.status]:>6}] {result.id}: {result.title}"
                f"  ({result.checked} checked)"
            )
            if result.waiver is not None:
                lines.append(f"           waived: {result.waiver.reason}")
            for finding in result.findings:
                where = f"{finding.target}: " if finding.target else ""
                lines.append(f"           - {where}{finding.message}")
        return lines

Attributes

assertions = PydanticField(default_factory=list) class-attribute instance-attribute
graflo_version = PydanticField(default=None) class-attribute instance-attribute
manifest_hash = PydanticField(default=None, description='Content address of the manifest checked, so a stored report can be matched back to the artifact that produced it.') class-attribute instance-attribute
ok property

No error-severity finding survived waivers.

profile = PydanticField(...) class-attribute instance-attribute
profile_version = PydanticField(...) class-attribute instance-attribute
status = PydanticField(...) class-attribute instance-attribute
subject = PydanticField(default=None, description='What was checked -- a path, or ``uuid@version``. Set by the caller.') class-attribute instance-attribute

Methods:

errors()

Every finding that makes the manifest non-conformant.

Source code in graflo/architecture/profile/model.py 
154
155
156
157
158
159
160
161
162
def errors(self) -> list[Finding]:
    """Every finding that makes the manifest non-conformant."""
    return [
        finding
        for result in self.assertions
        if result.waiver is None
        for finding in result.findings
        if finding.severity == "error"
    ]
to_lines()

The report as text. The one renderer -- every surface calls this.

Source code in graflo/architecture/profile/model.py 
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
def to_lines(self) -> list[str]:
    """The report as text. The one renderer -- every surface calls this."""
    marks = {
        "pass": "PASS",
        "fail": "FAIL",
        "warn": "WARN",
        "waived": "WAIVED",
        "not_applicable": "N/A",
    }
    subject = self.subject or "<manifest>"
    lines = [
        f"profile {self.profile} v{self.profile_version} -- {subject}",
        f"  overall: {marks[self.status]}",
        "",
    ]
    for result in self.assertions:
        lines.append(
            f"  [{marks[result.status]:>6}] {result.id}: {result.title}"
            f"  ({result.checked} checked)"
        )
        if result.waiver is not None:
            lines.append(f"           waived: {result.waiver.reason}")
        for finding in result.findings:
            where = f"{finding.target}: " if finding.target else ""
            lines.append(f"           - {where}{finding.message}")
    return lines
warnings()

Every advisory finding, waived assertions included.

Source code in graflo/architecture/profile/model.py 
164
165
166
167
168
169
170
171
def warnings(self) -> list[Finding]:
    """Every advisory finding, waived assertions included."""
    return [
        finding
        for result in self.assertions
        for finding in result.findings
        if finding.severity == "warning"
    ]

ProfileWaivers

Bases: ConfigBaseModel

A sidecar document of waivers granted against one profile.

Source code in graflo/architecture/profile/model.py 
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
class ProfileWaivers(ConfigBaseModel):
    """A sidecar document of waivers granted against one profile."""

    profile: str = PydanticField(default="world-model")
    subject: str | None = PydanticField(
        default=None,
        description=(
            "What these waivers were granted against -- a manifest name or "
            "content hash. Advisory: nothing enforces the match."
        ),
    )
    waivers: list[Waiver] = PydanticField(default_factory=list)

    def for_assertion(self, assertion_id: str) -> Waiver | None:
        """The waiver covering *assertion_id*, or ``None``."""
        for waiver in self.waivers:
            if waiver.assertion == assertion_id:
                return waiver
        return None

Attributes

profile = PydanticField(default='world-model') class-attribute instance-attribute
subject = PydanticField(default=None, description='What these waivers were granted against -- a manifest name or content hash. Advisory: nothing enforces the match.') class-attribute instance-attribute
waivers = PydanticField(default_factory=list) class-attribute instance-attribute

Methods:

for_assertion(assertion_id)

The waiver covering assertion_id, or None.

Source code in graflo/architecture/profile/model.py 
79
80
81
82
83
84
def for_assertion(self, assertion_id: str) -> Waiver | None:
    """The waiver covering *assertion_id*, or ``None``."""
    for waiver in self.waivers:
        if waiver.assertion == assertion_id:
            return waiver
    return None

VocabularyResolver

Bases: Protocol

Decides whether an IRI resolves to a vocabulary worth grounding in.

A protocol rather than a class so the bundled prefix check can be replaced by a real registry lookup without touching the report model or any of the surfaces that render it.

Source code in graflo/architecture/profile/context.py 
106
107
108
109
110
111
112
113
114
class VocabularyResolver(Protocol):
    """Decides whether an IRI resolves to a vocabulary worth grounding in.

    A protocol rather than a class so the bundled prefix check can be replaced
    by a real registry lookup without touching the report model or any of the
    surfaces that render it.
    """

    def resolve(self, iri: str) -> VocabularyStatus: ...

Methods:

resolve(iri)
Source code in graflo/architecture/profile/context.py 
114
def resolve(self, iri: str) -> VocabularyStatus: ...

Waiver

Bases: ConfigBaseModel

An operator's decision to excuse one assertion, with its reason.

Source code in graflo/architecture/profile/model.py 
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
class Waiver(ConfigBaseModel):
    """An operator's decision to excuse one assertion, with its reason."""

    assertion: str = PydanticField(
        ...,
        description="Identifier of the assertion this waiver excuses, e.g. ``temporal``.",
    )
    reason: str = PydanticField(
        ...,
        description=(
            "Why the assertion does not apply to this deployment. Required: a "
            "waiver without a reason is a silent pass with extra steps."
        ),
    )
    granted_by: str | None = PydanticField(default=None)
    granted_at: datetime | None = PydanticField(default=None)
    expires: datetime | None = PydanticField(default=None)

Attributes

assertion = PydanticField(..., description='Identifier of the assertion this waiver excuses, e.g. ``temporal``.') class-attribute instance-attribute
expires = PydanticField(default=None) class-attribute instance-attribute
granted_at = PydanticField(default=None) class-attribute instance-attribute
granted_by = PydanticField(default=None) class-attribute instance-attribute
reason = PydanticField(..., description='Why the assertion does not apply to this deployment. Required: a waiver without a reason is a silent pass with extra steps.') class-attribute instance-attribute

Functions:

audit_inverses(manifest)

Report how manifest realizes every declared inverse, and what is wrong with it.

Reads the schema, the physical profile and the ingestion model; changes nothing, raises nothing, and does not need the manifest to have loaded.

Source code in graflo/architecture/profile/inverses.py 
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
def audit_inverses(manifest: GraphManifest) -> InverseReport:
    """Report how *manifest* realizes every declared inverse, and what is wrong with it.

    Reads the schema, the physical profile and the ingestion model; changes
    nothing, raises nothing, and does not need the manifest to have loaded.
    """
    schema = manifest.graph_schema
    if schema is None:
        return InverseReport()
    edge_config = schema.core_schema.edge_config
    profile = schema.db_profile
    vertex_names = {vertex.name for vertex in schema.core_schema.vertex_config.vertices}

    findings = schema_inverse_findings(schema)
    ingestion_findings, feeding_table = _ingestion_findings(manifest, schema)
    findings = [*findings, *ingestion_findings]

    pairs: list[PairStatus] = []
    for pair in pair_realizations(schema, findings=findings):
        candidate = _native_candidate(pair)
        eligibility = (
            None
            if pair.native_side is not None
            else native_inverse_violations(
                profile, edge_config, vertex_names, candidates=[candidate]
            )
        )
        if eligibility is not None:
            eligibility = [v for v in eligibility if v.relation == candidate]
        pairs.append(
            PairStatus(
                **pair.model_dump(),
                feeding=feeding_table.get(frozenset({pair.relation, pair.inverse}), {}),
                native_eligibility=eligibility,
                native_candidate=None if pair.native_side else candidate,
            )
        )
    return InverseReport(
        pairs=pairs, symmetric=list(edge_config.symmetric), findings=findings
    )

check_manifest(manifest, *, profile='world-model', authored=None, waivers=None, subject=None, resolver=None)

Check an already-parsed manifest.

Prefer :func:check_manifest_config when the authored document is available: without it the declaration assertions can only warn.

Source code in graflo/architecture/profile/runner.py 
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
def check_manifest(
    manifest: GraphManifest,
    *,
    profile: str = "world-model",
    authored: Mapping[str, Any] | None = None,
    waivers: ProfileWaivers | None = None,
    subject: str | None = None,
    resolver: VocabularyResolver | None = None,
) -> ProfileReport:
    """Check an already-parsed *manifest*.

    Prefer :func:`check_manifest_config` when the authored document is
    available: without it the declaration assertions can only warn.
    """
    context = CheckContext(manifest=manifest, authored=authored, waivers=waivers)
    if resolver is not None:
        context.resolver = resolver
    report = run_profile(get_profile(profile), context)
    return report.model_copy(update={"subject": subject})

check_manifest_config(config, *, profile='world-model', waivers=None, subject=None, resolver=None)

Check the manifest config as authored.

The primary entry point. Parses config into a manifest and keeps the original mapping alongside it, so an assertion can tell "the author did not declare this" from "the author declared the value that is also the default".

Source code in graflo/architecture/profile/runner.py 
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
def check_manifest_config(
    config: Mapping[str, Any],
    *,
    profile: str = "world-model",
    waivers: ProfileWaivers | None = None,
    subject: str | None = None,
    resolver: VocabularyResolver | None = None,
) -> ProfileReport:
    """Check the manifest *config* as authored.

    The primary entry point. Parses *config* into a manifest and keeps the
    original mapping alongside it, so an assertion can tell "the author did not
    declare this" from "the author declared the value that is also the default".
    """
    manifest = GraphManifest.from_config(dict(config))
    manifest.finish_init()
    return check_manifest(
        manifest,
        profile=profile,
        authored=config,
        waivers=waivers,
        subject=subject,
        resolver=resolver,
    )

get_profile(name)

The profile called name.

Raises:

Type Description
KeyError

no such profile, naming the ones that exist.
Source code in graflo/architecture/profile/runner.py 
78
79
80
81
82
83
84
85
86
87
88
def get_profile(name: str) -> Profile:
    """The profile called *name*.

    Raises:
        KeyError: no such profile, naming the ones that exist.
    """
    profiles = _registry()
    if name not in profiles:
        known = ", ".join(sorted(profiles))
        raise KeyError(f"unknown profile {name!r}; known profiles: {known}")
    return profiles[name]

list_profiles()

(name, version) for every known profile.

Source code in graflo/architecture/profile/runner.py 
73
74
75
def list_profiles() -> list[tuple[str, str]]:
    """``(name, version)`` for every known profile."""
    return sorted((p.name, p.version) for p in _registry().values())

run_profile(profile, context)

Run every assertion of profile against context.

Source code in graflo/architecture/profile/runner.py 
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
def run_profile(profile: Profile, context: CheckContext) -> ProfileReport:
    """Run every assertion of *profile* against *context*."""
    results: list[AssertionResult] = []
    for assertion in profile.assertions:
        result = assertion.run(context)
        waiver = (
            context.waivers.for_assertion(assertion.id)
            if context.waivers is not None
            else None
        )
        if waiver is not None and result.status in ("fail", "warn"):
            result = result.model_copy(update={"status": "waived", "waiver": waiver})
        results.append(result)
    return ProfileReport(
        profile=profile.name,
        profile_version=profile.version,
        graflo_version=_graflo_version(),
        status=roll_up([r.status for r in results]),
        assertions=results,
    )