Skip to content

ontocast.stategraph.helpers

all_unit_patch_source_iris(state)

Sorted unique ontology IRIs appearing in any unit's patch source list.

Source code in ontocast/stategraph/helpers.py 
14
15
16
17
18
19
20
21
22
23
def all_unit_patch_source_iris(state: AgentState) -> list[str]:
    """Sorted unique ontology IRIs appearing in any unit's patch source list."""
    seen: set[str] = set()
    ordered: list[str] = []
    for sources in state.unit_patch_sources.values():
        for iri in sources:
            if iri not in seen:
                seen.add(iri)
                ordered.append(iri)
    return sorted(ordered)

build_document_excerpt(state)

Create a representative excerpt from sampled source units.

Source code in ontocast/stategraph/helpers.py 
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
def build_document_excerpt(state: AgentState) -> str:
    """Create a representative excerpt from sampled source units."""
    excerpt_parts: list[str] = []

    if state.content_units:
        unit_count = len(state.content_units)
        if unit_count == 1:
            sample_indices = [0]
        elif unit_count == 2:
            sample_indices = [0, 1]
        else:
            sample_indices = [0, 1, unit_count // 2, unit_count - 1]

        visited_indices: set[int] = set()
        for index in sample_indices:
            if index in visited_indices or index < 0 or index >= unit_count:
                continue
            visited_indices.add(index)
            unit_text = state.content_units[index].text.strip()
            if not unit_text:
                continue
            excerpt_parts.append(unit_text)

    if excerpt_parts:
        return "\n\n[...]\n\n".join(excerpt_parts)
    if state.docling_doc is not None:
        return state.docling_doc.export_to_markdown()
    return ""

enforce_redeclared_deletes(delta)

Drop merged deletes whose subject the merged inserts do not redeclare.

Applied only under vector-retrieval context: each unit judged its deletes on a retrieved subset, so a delete is trusted only as part of a redeclaration — the corrected statement must reappear in the inserts. A bare delete of catalog content, judged on partial evidence, would propagate onto shared catalog terminals cross-document. A safety invariant like the insert-wins veto in :func:merge_unit_deltas, not a knob; the per-unit foreign_delete finding teaches the loop, this backstops the write.

Returns:

Type Description
int

Number of delete triples dropped.
Source code in ontocast/stategraph/helpers.py 
 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
def enforce_redeclared_deletes(delta: OntologyDelta) -> int:
    """Drop merged deletes whose subject the merged inserts do not redeclare.

    Applied only under vector-retrieval context: each unit judged its deletes
    on a *retrieved subset*, so a delete is trusted only as part of a
    redeclaration — the corrected statement must reappear in the inserts. A
    bare delete of catalog content, judged on partial evidence, would
    propagate onto shared catalog terminals cross-document. A safety invariant
    like the insert-wins veto in :func:`merge_unit_deltas`, not a knob; the
    per-unit ``foreign_delete`` finding teaches the loop, this backstops the
    write.

    Returns:
        Number of delete triples dropped.
    """
    if len(delta.deletes) == 0:
        return 0
    redeclared = {
        subject for subject in delta.inserts.subjects() if isinstance(subject, URIRef)
    }
    dropped = 0
    for triple in list(delta.deletes):
        if triple[0] not in redeclared:
            delta.deletes.remove(triple)
            dropped += 1
    if dropped:
        logger.warning(
            "Dropped %d delete triple(s) whose subject the merged inserts do "
            "not redeclare — under retrieved (partial) context an "
            "unredeclared delete is judged on partial evidence and would "
            "propagate to shared catalog terminals",
            dropped,
        )
    return dropped

merge_unit_deltas(deltas)

Union per-unit deltas into one document-level delta.

Delete consensus across parallel units is conservative: a triple inserted by any unit wins over another unit's delete of the same triple, so the merged delta stays monotone-safe under parallel map/reduce.

Source code in ontocast/stategraph/helpers.py 
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
def merge_unit_deltas(deltas: list[OntologyDelta]) -> OntologyDelta:
    """Union per-unit deltas into one document-level delta.

    Delete consensus across parallel units is conservative: a triple inserted
    by any unit wins over another unit's delete of the same triple, so the
    merged delta stays monotone-safe under parallel map/reduce.
    """
    inserts = RDFGraph()
    deletes = RDFGraph()
    for delta in deltas:
        for graph, merged in ((delta.inserts, inserts), (delta.deletes, deletes)):
            for triple in graph:
                merged.add(triple)
            for prefix, namespace_uri in graph.namespaces():
                if prefix:
                    merged.bind(prefix, namespace_uri)
    insert_set = set(inserts)
    reconciled_deletes = RDFGraph()
    for prefix, namespace_uri in deletes.namespaces():
        if prefix:
            reconciled_deletes.bind(prefix, namespace_uri)
    vetoed = 0
    for triple in deletes:
        if triple in insert_set:
            vetoed += 1
            continue
        reconciled_deletes.add(triple)
    if vetoed:
        logger.info(
            "merge_unit_deltas: %d delete triple(s) vetoed by parallel-unit "
            "inserts (insert wins on conflict).",
            vetoed,
        )
    return OntologyDelta(inserts=inserts, deletes=reconciled_deletes)

reconcile_fresh_ontologies(fresh_ontologies)

Reconcile the fresh-create fan-out's N independent ontologies.

Units on the fresh path each return a whole ontology. Same-IRI artifacts are union-merged (the reduce used to keep whichever indexed last, silently dropping the others' content); across different fresh IRIs, minted duplicates are detected pairwise so the overlap is at least visible — no LLM harmonization is attempted.

Returns:

Type Description
list[Ontology]

The reconciled artifact list and the metrics to record
dict[str, int]

(fresh_ontologies_merged, fresh_minted_duplicates; keys
tuple[list[Ontology], dict[str, int]]

present only when non-zero).
Source code in ontocast/stategraph/helpers.py 
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
def reconcile_fresh_ontologies(
    fresh_ontologies: list[Ontology],
) -> tuple[list[Ontology], dict[str, int]]:
    """Reconcile the fresh-create fan-out's N independent ontologies.

    Units on the fresh path each return a whole ontology. Same-IRI artifacts
    are union-merged (the reduce used to keep whichever indexed last,
    silently dropping the others' content); across *different* fresh IRIs,
    minted duplicates are detected pairwise so the overlap is at least
    visible — no LLM harmonization is attempted.

    Returns:
        The reconciled artifact list and the metrics to record
        (``fresh_ontologies_merged``, ``fresh_minted_duplicates``; keys
        present only when non-zero).
    """
    metrics: dict[str, int] = {}
    if len(fresh_ontologies) <= 1:
        return fresh_ontologies, metrics

    by_iri: dict[str, list[Ontology]] = {}
    for fresh in fresh_ontologies:
        by_iri.setdefault(fresh.iri, []).append(fresh)
    merged: list[Ontology] = []
    same_iri_merges = 0
    for group in by_iri.values():
        if len(group) == 1:
            merged.append(group[0])
            continue
        same_iri_merges += len(group) - 1
        logger.warning(
            "%d unit(s) produced fresh ontologies under the same IRI %s; "
            "union-merging instead of keeping the last one",
            len(group),
            group[0].iri,
        )
        merged.append(group[0].union_fresh(group[1:]))
    if same_iri_merges:
        metrics["fresh_ontologies_merged"] = same_iri_merges

    if len(merged) > 1:
        cross_duplicates = 0
        for index, fresh in enumerate(merged):
            other_graphs = {other.iri: other.graph for other in merged[index + 1 :]}
            cross_duplicates += len(detect_minted_duplicates(fresh.graph, other_graphs))
        if cross_duplicates:
            metrics["fresh_minted_duplicates"] = cross_duplicates
            logger.warning(
                "%d term(s) overlap across %d distinct fresh ontologies — "
                "consider a shared seed ontology",
                cross_duplicates,
                len(merged),
            )
    return merged, metrics