Skip to content

Inquiry API

Status: Generated from current Python docstrings and type hints.

Semantic-inquiry state machine, focus tracking, obligations, hypotheses, rejections, review reports, and structured diagnostics for the gaia inquiry CLI sub-app. Five of the 45 public symbols moved here from gaia.cli.* in alpha 0 (KnowledgeBreakdown, HoleEntry, analyze_knowledge_breakdown, find_possible_duplicate_claims, load_or_generate_review_manifest).

gaia.engine.inquiry

gaia.engine.inquiry — spec §10 public surface.

Thin wrapper over Gaia. This module does not run its own compiler, validator, or inference engine; it composes the ones already in Gaia.

SourceAnchor dataclass 

SourceAnchor(file: str, line: int, column: int)

Source location for a DSL declaration inside a package.

to_dict 

to_dict() -> dict[str, Any]

Return the JSON-compatible anchor payload.

Source code in gaia/engine/inquiry/anchor.py 
65
66
67
def to_dict(self) -> dict[str, Any]:
    """Return the JSON-compatible anchor payload."""
    return {"file": self.file, "line": self.line, "column": self.column}

HoleEntry dataclass 

HoleEntry(cid: str, label: str, content: str, prior: float | None, prior_justification: str = '')

One independent claim, with or without a prior set.

is_hole property 

is_hole: bool

Return whether this independent claim is missing a prior.

KnowledgeBreakdown dataclass 

KnowledgeBreakdown(settings: list[str] = list(), questions: list[str] = list(), independent: list[HoleEntry] = list(), derived: list[str] = list(), structural: list[str] = list(), background_only: list[str] = list(), orphaned: list[str] = list(), classification: KnowledgeClassification = KnowledgeClassification())

Role-based breakdown of all knowledge nodes in an IR.

holes property 

holes: list[HoleEntry]

Return independent claims that still need priors.

covered property 

covered: list[HoleEntry]

Return independent claims that already have priors.

Diagnostic dataclass 

Diagnostic(severity: Severity, kind: DiagnosticKind, target: str, label: str, message: str, suggested_edit: str = '', data: dict[str, Any] = dict(), source_anchor: SourceAnchor | None = None)

Spec §15 Diagnostic record. Uniform across all detection sources.

to_dict 

to_dict() -> dict[str, Any]

Return the diagnostic as a JSON-compatible dictionary.

Source code in gaia/engine/inquiry/diagnostics.py 
67
68
69
70
71
72
73
74
def to_dict(self) -> dict[str, Any]:
    """Return the diagnostic as a JSON-compatible dictionary."""
    d = asdict(self)
    if not d["data"]:
        d.pop("data")
    if d.get("source_anchor") is None:
        d.pop("source_anchor", None)
    return d

NextEdit dataclass 

NextEdit(text: str, kind: str, severity: Severity, target: str, label: str, source_anchor: SourceAnchor | None = None)

Spec §8.8 / Round A2 structured edit suggestion.

text 是渲染给人看的 imperative 一行; source_anchor 在可定位时指向 需要修改的源位置。其余字段复制自产生该 edit 的 Diagnostic。

to_dict 

to_dict() -> dict[str, Any]

Return the structured edit as a JSON-compatible dictionary.

Source code in gaia/engine/inquiry/diagnostics.py 
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
def to_dict(self) -> dict[str, Any]:
    """Return the structured edit as a JSON-compatible dictionary."""
    d: dict[str, Any] = {
        "text": self.text,
        "kind": self.kind,
        "severity": self.severity,
        "target": self.target,
        "label": self.label,
    }
    if self.source_anchor is not None:
        d["source_anchor"] = self.source_anchor.to_dict()
    return d

ClaimDelta dataclass 

ClaimDelta(label: str, field: str, before: str, after: str)

Per-field change record for a knowledge / strategy / operator / prior.

to_dict 

to_dict() -> dict[str, str]

Return the delta record as a JSON-compatible dictionary.

Source code in gaia/engine/inquiry/diff.py 
36
37
38
39
40
41
42
43
def to_dict(self) -> dict[str, str]:
    """Return the delta record as a JSON-compatible dictionary."""
    return {
        "label": self.label,
        "field": self.field,
        "before": self.before,
        "after": self.after,
    }

SemanticDiff dataclass 

SemanticDiff(baseline_review_id: str | None = None, added_claims: list[str] = list(), removed_claims: list[str] = list(), changed_claims: list[ClaimDelta] = list(), added_questions: list[str] = list(), removed_questions: list[str] = list(), added_settings: list[str] = list(), removed_settings: list[str] = list(), added_strategies: list[str] = list(), removed_strategies: list[str] = list(), changed_strategies: list[ClaimDelta] = list(), added_operators: list[str] = list(), removed_operators: list[str] = list(), changed_operators: list[ClaimDelta] = list(), changed_priors: list[ClaimDelta] = list(), changed_exports: list[ClaimDelta] = list())

Spec §9.1 semantic_diff object — covers all 16 §14.2 categories.

is_empty property 

is_empty: bool

Return whether the diff contains no semantic deltas.

to_dict 

to_dict() -> dict[str, Any]

Return the full semantic diff as a JSON-compatible dictionary.

Source code in gaia/engine/inquiry/diff.py 
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
def to_dict(self) -> dict[str, Any]:
    """Return the full semantic diff as a JSON-compatible dictionary."""
    return {
        "baseline_review_id": self.baseline_review_id,
        "added_claims": list(self.added_claims),
        "removed_claims": list(self.removed_claims),
        "changed_claims": [d.to_dict() for d in self.changed_claims],
        "added_questions": list(self.added_questions),
        "removed_questions": list(self.removed_questions),
        "added_settings": list(self.added_settings),
        "removed_settings": list(self.removed_settings),
        "added_strategies": list(self.added_strategies),
        "removed_strategies": list(self.removed_strategies),
        "changed_strategies": [d.to_dict() for d in self.changed_strategies],
        "added_operators": list(self.added_operators),
        "removed_operators": list(self.removed_operators),
        "changed_operators": [d.to_dict() for d in self.changed_operators],
        "changed_priors": [d.to_dict() for d in self.changed_priors],
        "changed_exports": [d.to_dict() for d in self.changed_exports],
    }

FocusBinding dataclass 

FocusBinding(raw: str | None, resolved_id: str | None = None, resolved_label: str | None = None, kind: str = 'freeform')

Resolved inquiry focus target for review and rendering.

to_dict 

to_dict() -> dict[str, str | None]

Return the focus binding as a JSON-compatible dictionary.

Source code in gaia/engine/inquiry/focus.py 
18
19
20
21
22
23
24
25
def to_dict(self) -> dict[str, str | None]:
    """Return the focus binding as a JSON-compatible dictionary."""
    return {
        "raw": self.raw,
        "resolved_id": self.resolved_id,
        "resolved_label": self.resolved_label,
        "kind": self.kind,
    }

HypothesisView dataclass 

HypothesisView(qid: str, content: str, scope_qid: str | None, origin: str)

Display record for an IR or synthetic proof hypothesis.

ObligationView dataclass 

ObligationView(qid: str, target_qid: str | None, content: str, diagnostic_kind: str, origin: str, anchor: dict[str, Any] = dict())

Display record for an IR or synthetic proof obligation.

ProofContext dataclass 

ProofContext(obligations: list[ObligationView] = list(), hypotheses: list[HypothesisView] = list(), rejections: list[RejectionView] = list())

Merged proof-state view shown in inquiry review reports.

to_dict 

to_dict() -> dict[str, Any]

Return the proof context as a JSON-compatible dictionary.

Source code in gaia/engine/inquiry/proof_state.py 
52
53
54
55
56
57
58
def to_dict(self) -> dict[str, Any]:
    """Return the proof context as a JSON-compatible dictionary."""
    return {
        "obligations": [asdict(o) for o in self.obligations],
        "hypotheses": [asdict(h) for h in self.hypotheses],
        "rejections": [asdict(r) for r in self.rejections],
    }

RejectionView dataclass 

RejectionView(qid: str, target_strategy: str, content: str)

Display record for a closed or rejected strategy branch.

ReviewReport dataclass 

ReviewReport(review_id: str, created_at: str, path: str, focus: FocusBinding, mode: str, compile_status: str, ir_hash: str | None, counts: dict[str, int], semantic_diff: SemanticDiff = empty_diff(), graph_health: dict[str, Any] = dict(), inquiry_tree: dict[str, Any] = dict(), prior_holes: list[dict[str, Any]] = list(), belief_report: dict[str, Any] = dict(), diagnostics: list[Diagnostic] = list(), next_edits: list[str] = list(), next_edits_structured: list[NextEdit] = list(), proof_context: ProofContext | None = None)

Passive container for the eight-section Gaia inquiry review report.

to_json_dict 

to_json_dict() -> dict[str, Any]

Return the report in the shared JSON dictionary shape.

Source code in gaia/engine/inquiry/review.py 
114
115
116
def to_json_dict(self) -> dict[str, Any]:
    """Return the report in the shared JSON dictionary shape."""
    return _to_json_dict(self)

InquiryState dataclass 

InquiryState(version: int = STATE_SCHEMA_VERSION, focus: str | None = None, focus_kind: str | None = None, focus_resolved_id: str | None = None, mode: str = 'auto', last_review_id: str | None = None, baseline_review_id: str | None = None, focus_stack: list[dict[str, Any]] = list(), synthetic_obligations: list[SyntheticObligation] = list(), synthetic_hypotheses: list[SyntheticHypothesis] = list(), synthetic_rejections: list[SyntheticRejection] = list())

Mutable Lean-style inquiry state stored under .gaia/inquiry.

to_dict 

to_dict() -> dict[str, Any]

Return the persisted state payload as a JSON-compatible dictionary.

Source code in gaia/engine/inquiry/state.py 
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
def to_dict(self) -> dict[str, Any]:
    """Return the persisted state payload as a JSON-compatible dictionary."""
    return {
        "version": self.version,
        "focus": self.focus,
        "focus_kind": self.focus_kind,
        "focus_resolved_id": self.focus_resolved_id,
        "mode": self.mode,
        "last_review_id": self.last_review_id,
        "baseline_review_id": self.baseline_review_id,
        "focus_stack": list(self.focus_stack),
        "synthetic_obligations": [asdict(o) for o in self.synthetic_obligations],
        "synthetic_hypotheses": [asdict(h) for h in self.synthetic_hypotheses],
        "synthetic_rejections": [asdict(r) for r in self.synthetic_rejections],
    }

SyntheticHypothesis dataclass 

SyntheticHypothesis(qid: str, content: str, scope_qid: str | None = None, created_at: str | None = None)

Synthetic hypothesis tracked outside compiled IR.

SyntheticObligation dataclass 

SyntheticObligation(qid: str, target_qid: str, content: str, diagnostic_kind: str = 'other', anchor: dict[str, Any] = dict(), created_at: str | None = None)

Synthetic obligation tracked outside compiled IR.

SyntheticRejection dataclass 

SyntheticRejection(qid: str, target_strategy: str, content: str, created_at: str | None = None)

Synthetic record for a rejected strategy branch.

find_anchors 

find_anchors(pkg_path: str | Path) -> dict[str, SourceAnchor]

Scan package Python files and return a label-to-anchor map.

重复 label 取首次出现; 排除 .gaia/ 与隐藏目录。

Source code in gaia/engine/inquiry/anchor.py 
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
def find_anchors(pkg_path: str | Path) -> dict[str, SourceAnchor]:
    """Scan package Python files and return a label-to-anchor map.

    重复 label 取首次出现; 排除 .gaia/ 与隐藏目录。
    """
    root = Path(pkg_path).resolve()
    out: dict[str, SourceAnchor] = {}
    if not root.is_dir():
        return out
    for py_file in sorted(root.rglob("*.py")):
        # 跳过 .gaia/ / .venv / __pycache__ / 隐藏目录
        if any(
            part.startswith(".") or part == "__pycache__"
            for part in py_file.relative_to(root).parts
        ):
            continue
        rel = py_file.relative_to(root).as_posix()
        for label, anchor in _scan_module(py_file, rel).items():
            out.setdefault(label, anchor)
    return out

analyze_knowledge_breakdown 

analyze_knowledge_breakdown(ir: dict[str, Any]) -> KnowledgeBreakdown

Walk the IR once and classify every knowledge node by structural role.

Source code in gaia/engine/inquiry/check_core.py 
 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
def analyze_knowledge_breakdown(ir: dict[str, Any]) -> KnowledgeBreakdown:
    """Walk the IR once and classify every knowledge node by structural role."""
    c = classify_ir(ir)
    out = KnowledgeBreakdown(classification=c)

    for k in ir.get("knowledges", []):
        ktype = k.get("type")
        kid = k["id"]
        label = k.get("label") or kid.split("::")[-1]
        if ktype == "setting":
            out.settings.append(label)
            continue
        if ktype == "question":
            out.questions.append(label)
            continue
        if ktype != "claim":
            continue
        role = node_role(kid, "claim", c)
        if role == "structural":
            out.structural.append(label)
        elif role == "derived":
            out.derived.append(label)
        elif role == "independent":
            meta = k.get("metadata") or {}
            out.independent.append(
                HoleEntry(
                    cid=kid,
                    label=label,
                    content=k.get("content", ""),
                    prior=get_prior(k),
                    prior_justification=meta.get("prior_justification", ""),
                )
            )
        elif role == "background":
            out.background_only.append(label)
        else:
            out.orphaned.append(label)
    return out

find_possible_duplicate_claims 

find_possible_duplicate_claims(ir: dict[str, Any]) -> list[tuple[str, str]]

Heuristic: pairs of claims with identical normalized content.

Conservative — only exact-match after whitespace collapse. Per spec §8 possible_duplicate_claims is a graph-health hint, not an error.

Source code in gaia/engine/inquiry/check_core.py 
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
def find_possible_duplicate_claims(ir: dict[str, Any]) -> list[tuple[str, str]]:
    """Heuristic: pairs of claims with identical normalized content.

    Conservative — only exact-match after whitespace collapse. Per spec §8
    `possible_duplicate_claims` is a graph-health hint, not an error.
    """
    claims = [k for k in ir.get("knowledges", []) if k.get("type") == "claim"]
    by_norm: dict[str, list[str]] = {}
    for k in claims:
        content = " ".join((k.get("content") or "").split()).lower()
        if not content:
            continue
        by_norm.setdefault(content, []).append(k.get("label", k["id"].split("::")[-1]))
    pairs: list[tuple[str, str]] = []
    for labels in by_norm.values():
        if len(labels) < 2:
            continue
        for i in range(len(labels)):
            for j in range(i + 1, len(labels)):
                pairs.append((labels[i], labels[j]))
    return pairs

format_diagnostics_as_next_edits 

format_diagnostics_as_next_edits(diags: list[Diagnostic]) -> list[str]

Format diagnostics as the spec §8 text Next edits list.

若 diagnostic 带 source_anchor, 追加 (file:line) 到末尾, 便于人眼直接定位源行。

Source code in gaia/engine/inquiry/diagnostics.py 
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
def format_diagnostics_as_next_edits(diags: list[Diagnostic]) -> list[str]:
    """Format diagnostics as the spec §8 text ``Next edits`` list.

    若 diagnostic 带 ``source_anchor``, 追加 ``(file:line)`` 到末尾,
    便于人眼直接定位源行。
    """
    seen: set[str] = set()
    ordered: list[str] = []
    for d in sorted(diags, key=lambda d: _PRIO.get(d.severity, 9)):
        edit = d.suggested_edit.strip()
        if not edit or edit in seen:
            continue
        seen.add(edit)
        if d.source_anchor is not None:
            ordered.append(f"{edit} ({d.source_anchor.file}:{d.source_anchor.line})")
        else:
            ordered.append(edit)
    return ordered

format_diagnostics_as_structured_edits 

format_diagnostics_as_structured_edits(diags: list[Diagnostic]) -> list[NextEdit]

Format diagnostics as Round A2 structured NextEdit records.

Source code in gaia/engine/inquiry/diagnostics.py 
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
def format_diagnostics_as_structured_edits(diags: list[Diagnostic]) -> list[NextEdit]:
    """Format diagnostics as Round A2 structured ``NextEdit`` records."""
    seen: set[str] = set()
    out: list[NextEdit] = []
    for d in sorted(diags, key=lambda d: _PRIO.get(d.severity, 9)):
        edit = d.suggested_edit.strip()
        if not edit or edit in seen:
            continue
        seen.add(edit)
        out.append(
            NextEdit(
                text=edit,
                kind=d.kind,
                severity=d.severity,
                target=d.target,
                label=d.label,
                source_anchor=d.source_anchor,
            )
        )
    return out

from_knowledge_breakdown 

from_knowledge_breakdown(kb: KnowledgeBreakdown, ir: dict[str, Any], focus: FocusBinding | None, anchors: dict[str, SourceAnchor] | None = None) -> list[Diagnostic]

Emit diagnostics for prior holes, orphans, background-only, duplicates.

Source code in gaia/engine/inquiry/diagnostics.py 
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
def from_knowledge_breakdown(
    kb: KnowledgeBreakdown,
    ir: dict[str, Any],
    focus: FocusBinding | None,
    anchors: dict[str, SourceAnchor] | None = None,
) -> list[Diagnostic]:
    """Emit diagnostics for prior holes, orphans, background-only, duplicates."""
    _ = focus
    out: list[Diagnostic] = []
    for entry in kb.holes:
        out.append(_attach_anchor(_prior_hole_diag(entry), anchors))
    for label in kb.orphaned:
        out.append(
            _attach_anchor(
                Diagnostic(
                    severity="warning",
                    kind="orphaned_claim",
                    target=label,
                    label=label,
                    message="Claim is not referenced by any strategy or operator.",
                    suggested_edit=(
                        f"Either connect `{label}` to a strategy/operator, or remove it."
                    ),
                ),
                anchors,
            )
        )
    for label in kb.background_only:
        out.append(
            _attach_anchor(
                Diagnostic(
                    severity="info",
                    kind="background_only_claim",
                    target=label,
                    label=label,
                    message="Claim appears only in strategy background; not part of the BP graph.",
                    suggested_edit=(
                        f"If `{label}` should affect beliefs, move it to premises; "
                        "otherwise leave as background."
                    ),
                ),
                anchors,
            )
        )
    for a, b in find_possible_duplicate_claims(ir):
        out.append(
            Diagnostic(
                severity="warning",
                kind="possible_duplicate_claim",
                target=f"{a}|{b}",
                label=f"{a} / {b}",
                message=f"Claims `{a}` and `{b}` have identical content.",
                suggested_edit=f"Merge `{a}` and `{b}`, or differentiate their content.",
            )
        )
    return out

from_validation 

from_validation(warnings: list[str], errors: list[str]) -> list[Diagnostic]

Lift strings from ValidationResult into Diagnostic records.

Source code in gaia/engine/inquiry/diagnostics.py 
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
def from_validation(warnings: list[str], errors: list[str]) -> list[Diagnostic]:
    """Lift strings from ``ValidationResult`` into ``Diagnostic`` records."""
    out: list[Diagnostic] = []
    for msg in errors:
        out.append(
            Diagnostic(
                severity="error",
                kind="validation_error",
                target="graph",
                label="graph",
                message=msg,
            )
        )
    for msg in warnings:
        out.append(
            Diagnostic(
                severity="warning",
                kind="validation_warning",
                target="graph",
                label="graph",
                message=msg,
            )
        )
    return out

empty_diff 

empty_diff() -> SemanticDiff

Return an empty semantic diff for report defaults.

Source code in gaia/engine/inquiry/diff.py 
130
131
132
def empty_diff() -> SemanticDiff:
    """Return an empty semantic diff for report defaults."""
    return SemanticDiff()

resolve_focus_target 

resolve_focus_target(target: str | None, graph: Any) -> FocusBinding

Resolve a raw focus selector against graph IDs and labels.

Source code in gaia/engine/inquiry/focus.py 
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
def resolve_focus_target(target: str | None, graph: Any) -> FocusBinding:
    """Resolve a raw focus selector against graph IDs and labels."""
    if target is None:
        return FocusBinding(raw=None, kind="none")
    t = str(target).strip()
    if not t:
        return FocusBinding(raw=None, kind="none")

    if graph is None:
        return FocusBinding(raw=t, kind="freeform")

    knowledges = getattr(graph, "knowledges", None) or []
    by_id: dict[str, object] = {}
    by_label: dict[str, object] = {}
    for k in knowledges:
        kid = getattr(k, "id", None)
        klabel = getattr(k, "label", None)
        if kid:
            by_id[kid] = k
        if klabel:
            by_label[klabel] = k

    hit = by_id.get(t) or by_label.get(t)
    if hit is None:
        return FocusBinding(raw=t, kind="freeform")

    ktype = getattr(hit, "type", None)
    kind = "claim"
    if str(ktype).endswith("question") or str(ktype) == "question":
        kind = "question"
    elif str(ktype).endswith("setting") or str(ktype) == "setting":
        kind = "setting"
    return FocusBinding(
        raw=t,
        resolved_id=getattr(hit, "id", None),
        resolved_label=getattr(hit, "label", None),
        kind=kind,
    )

build_proof_context 

build_proof_context(graph: Any, state: InquiryState) -> ProofContext

Build the merged IR and synthetic proof context for a package.

Source code in gaia/engine/inquiry/proof_state.py 
 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
def build_proof_context(graph: Any, state: InquiryState) -> ProofContext:
    """Build the merged IR and synthetic proof context for a package."""
    ctx = ProofContext()

    # IR side — question() becomes obligation view, setting() becomes hypothesis view.
    if graph is not None:
        for k in getattr(graph, "knowledges", []) or []:
            ktype = str(getattr(k, "type", ""))
            qid = getattr(k, "id", None) or getattr(k, "label", None) or ""
            content = getattr(k, "content", "") or ""
            if ktype.endswith("question") or ktype == "question":
                ctx.obligations.append(
                    ObligationView(
                        qid=qid,
                        target_qid=None,
                        content=content,
                        diagnostic_kind="other",
                        origin="ir",
                    )
                )
            elif ktype.endswith("setting") or ktype == "setting":
                ctx.hypotheses.append(
                    HypothesisView(
                        qid=qid,
                        content=content,
                        scope_qid=None,
                        origin="ir",
                    )
                )

    # Synthetic state.
    for o in state.synthetic_obligations:
        ctx.obligations.append(
            ObligationView(
                qid=o.qid,
                target_qid=o.target_qid,
                content=o.content,
                diagnostic_kind=o.diagnostic_kind,
                origin="synthetic",
                anchor=dict(o.anchor),
            )
        )
    for h in state.synthetic_hypotheses:
        ctx.hypotheses.append(
            HypothesisView(
                qid=h.qid,
                content=h.content,
                scope_qid=h.scope_qid,
                origin="synthetic",
            )
        )
    for r in state.synthetic_rejections:
        ctx.rejections.append(
            RejectionView(
                qid=r.qid,
                target_strategy=r.target_strategy,
                content=r.content,
            )
        )
    return ctx

render_json 

render_json(report: ReviewReport) -> str

Render a review report as pretty JSON without ASCII escaping.

Source code in gaia/engine/inquiry/render.py 
534
535
536
def render_json(report: ReviewReport) -> str:
    """Render a review report as pretty JSON without ASCII escaping."""
    return json.dumps(to_json_dict(report), ensure_ascii=False, indent=2)

render_markdown 

render_markdown(report: ReviewReport) -> str

Spec §17.2 Markdown renderer.

Mirrors the eight-section text layout but uses Markdown headings, bullet lists, and fenced code blocks for IDs/source anchors. The section names match render_text exactly so agents can diff outputs.

Source code in gaia/engine/inquiry/render.py 
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
def render_markdown(report: ReviewReport) -> str:
    """Spec §17.2 Markdown renderer.

    Mirrors the eight-section text layout but uses Markdown headings, bullet
    lists, and fenced code blocks for IDs/source anchors. The section names
    match render_text exactly so agents can diff outputs.
    """
    md: list[str] = []
    _append_markdown_header(md, report)
    _append_markdown_focus(md, report)
    _append_markdown_compile(md, report)
    _append_markdown_semantic_diff(md, report)
    _append_markdown_graph_health(md, report)
    _append_markdown_inquiry_tree(md, report)
    _append_markdown_prior_holes(md, report)
    _append_markdown_belief_report(md, report)
    _append_markdown_proof_state(md, report)
    _append_markdown_next_edits(md, report)

    return "\n".join(md)

to_json_dict 

to_json_dict(report: ReviewReport) -> dict[str, Any]

Serialize a review report to the spec §9.1 JSON dictionary shape.

Source code in gaia/engine/inquiry/render.py 
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
def to_json_dict(report: ReviewReport) -> dict[str, Any]:
    """Serialize a review report to the spec §9.1 JSON dictionary shape."""
    return {
        "review_id": report.review_id,
        "created_at": report.created_at,
        "path": report.path,
        "focus": _focus_to_dict(report.focus),
        "mode": report.mode,
        "compile": {
            "status": report.compile_status,
            "ir_hash": report.ir_hash,
            "counts": dict(report.counts),
        },
        "semantic_diff": report.semantic_diff.to_dict(),
        "graph_health": report.graph_health,
        "inquiry_tree": report.inquiry_tree,
        "prior_holes": list(report.prior_holes),
        "belief_report": report.belief_report,
        "diagnostics": [d.to_dict() for d in report.diagnostics],
        "next_edits": list(report.next_edits),
        "next_edits_structured": [e.to_dict() for e in report.next_edits_structured],
        "proof_context": _proof_context_to_dict(report.proof_context),
    }

render_text 

render_text(report: ReviewReport) -> str

Spec §8 eight-section text renderer.

Source code in gaia/engine/inquiry/review.py 
138
139
140
def render_text(report: ReviewReport) -> str:
    """Spec §8 eight-section text renderer."""
    return _render_text(report)

resolve_graph 

resolve_graph(path: str | Path) -> Any

Compile a package and return its LocalCanonicalGraph (or None on failure).

Source code in gaia/engine/inquiry/review.py 
124
125
126
127
128
129
130
131
132
133
134
135
def resolve_graph(path: str | Path) -> Any:
    """Compile a package and return its LocalCanonicalGraph (or None on failure)."""
    try:
        ensure_package_env(Path(path).resolve())
        loaded = load_gaia_package(str(path))
        apply_package_priors(loaded)
        compiled = compile_loaded_package_artifact(loaded)
    except GaiaPackagingError:
        return None
    except Exception:
        return None
    return compiled.graph

run_review 

run_review(path: str | Path, *, focus_override: str | None = None, mode: str = 'auto', no_infer: bool = False, depth: int = 0, since: str | None = None, strict: bool = False) -> ReviewReport

Run the inquiry review pipeline and persist a review snapshot.

Source code in gaia/engine/inquiry/review.py 
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
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
def run_review(
    path: str | Path,
    *,
    focus_override: str | None = None,
    mode: str = "auto",
    no_infer: bool = False,
    depth: int = 0,
    since: str | None = None,
    strict: bool = False,
) -> ReviewReport:
    """Run the inquiry review pipeline and persist a review snapshot."""
    del strict
    pkg_path = Path(path).resolve()
    state = load_state(pkg_path)
    focus_raw = focus_override if focus_override is not None else state.focus

    warnings: list[str] = []
    errors: list[str] = []
    graph = None
    loaded = None
    compiled = None
    review_manifest: ReviewManifest | None = None
    compile_status = "error"
    ir_hash: str | None = None
    counts = {"knowledge": 0, "strategies": 0, "operators": 0}

    # Step 1: compile via Gaia.
    try:
        ensure_package_env(pkg_path)
        loaded = load_gaia_package(str(pkg_path))
        apply_package_priors(loaded)
        compiled = compile_loaded_package_artifact(loaded)
        graph = compiled.graph
        compile_status = "ok"
    except GaiaPackagingError as exc:
        errors.append(f"compile: {exc}")
    except Exception as exc:  # surfaced as report error, not raised
        errors.append(f"compile: {exc}")

    if graph is not None:
        counts["knowledge"] = len(getattr(graph, "knowledges", []) or [])
        counts["strategies"] = len(getattr(graph, "strategies", []) or [])
        counts["operators"] = len(getattr(graph, "operators", []) or [])
        ir_hash = getattr(graph, "ir_hash", None)

        # Step 2: validate via Gaia.
        validation = validate_local_graph(graph)
        warnings.extend(validation.warnings)
        errors.extend(validation.errors)

    if loaded is not None and compiled is not None:
        try:
            review_manifest = load_or_generate_review_manifest(loaded.pkg_path, compiled)
        except GaiaPackagingError as exc:
            errors.append(f"review_manifest: {exc}")

    focus = resolve_focus_target(focus_raw, graph)

    # Step 3: knowledge breakdown via check_core (single source of truth).
    ir_dict = _graph_to_ir_dict(graph)
    kb = analyze_knowledge_breakdown(ir_dict) if ir_dict is not None else KnowledgeBreakdown()

    prior_holes = _build_prior_holes(kb)
    inquiry_tree = _build_inquiry_tree(kb, graph, review_manifest)

    # Step 4: semantic diff against baseline snapshot.
    baseline_id = resolve_baseline(pkg_path, since, state.last_review_id)
    baseline_snap = load_snapshot(pkg_path, baseline_id) if baseline_id else None
    semantic_diff = compute_semantic_diff(ir_dict, baseline_snap)

    # Step 5: inference via gaia.engine.bp; enrich with baseline belief deltas.
    belief_report = _build_belief_report(
        graph,
        pkg_path,
        no_infer,
        errors,
        focus,
        loaded=loaded,
        compiled=compiled,
        review_manifest=review_manifest,
        depth=depth,
    )
    if belief_report["ran_inference"] and baseline_snap is not None:
        _annotate_belief_deltas(belief_report, baseline_snap)

    graph_health = _build_graph_health(kb, ir_dict, warnings, errors)

    # Step 6: diagnostics — translate validator + breakdown into one stream.
    anchors = find_anchors(pkg_path)
    diagnostics: list[Diagnostic] = []
    diagnostics.extend(from_validation(warnings, errors))
    if ir_dict is not None:
        diagnostics.extend(from_knowledge_breakdown(kb, ir_dict, focus, anchors))
        diagnostics.extend(detect_prior_without_justification(kb, anchors))
        diagnostics.extend(detect_prior_dissent(ir_dict, anchors=anchors))
        diagnostics.extend(detect_prior_overridden(ir_dict, anchors=anchors))
    diagnostics.extend(detect_stale_artifact(pkg_path, ir_hash))
    diagnostics.extend(detect_focus_low_posterior(belief_report))
    rejected_targets = {r.target_strategy for r in getattr(state, "synthetic_rejections", []) or []}
    diagnostics.extend(
        detect_warrant_status(
            graph,
            rejected_targets,
            anchors,
            review_manifest=review_manifest,
        )
    )
    if graph is not None:
        if ir_dict is not None:
            diagnostics.extend(detect_blocked_warrant_path(graph, kb, anchors))
        diagnostics.extend(detect_focus_unsupported(graph, focus, anchors))
        diagnostics.extend(detect_overstrong_strategy_without_provenance(graph, anchors=anchors))
        diagnostics.extend(
            detect_claim_with_evidence_but_no_focus_connection(graph, focus, anchors)
        )
    diagnostics.extend(detect_large_belief_drop(belief_report))
    diagnostics = rank_diagnostics(diagnostics, mode)
    next_edits_structured = rank_next_edits(
        format_diagnostics_as_structured_edits(diagnostics), mode
    )
    next_edits = [
        f"{e.text} ({e.source_anchor.file}:{e.source_anchor.line})"
        if e.source_anchor is not None
        else e.text
        for e in next_edits_structured
    ]

    # Step 7: ProofContext (Round A1).
    proof_ctx = build_proof_context(graph, state)

    review_id = mint_review_id(ir_hash, mode)
    created_at = _utcnow_iso()
    report = ReviewReport(
        review_id=review_id,
        created_at=created_at,
        path=str(pkg_path),
        focus=focus,
        mode=mode,
        compile_status=compile_status,
        ir_hash=ir_hash,
        counts=counts,
        semantic_diff=semantic_diff,
        graph_health=graph_health,
        inquiry_tree=inquiry_tree,
        prior_holes=prior_holes,
        belief_report=belief_report,
        diagnostics=diagnostics,
        next_edits=next_edits,
        next_edits_structured=next_edits_structured,
        proof_context=proof_ctx,
    )

    # Persist snapshot for future diffs.
    snapshot_path = save_snapshot(
        pkg_path,
        review_id=review_id,
        created_at=created_at,
        ir_hash=ir_hash,
        ir_dict=ir_dict,
        beliefs=belief_report.get("beliefs", []),
    )
    actual_review_id = snapshot_path.stem
    if actual_review_id != review_id:
        review_id = actual_review_id
        report.review_id = actual_review_id

    state.last_review_id = review_id
    if state.baseline_review_id is None:
        state.baseline_review_id = review_id
    state.mode = mode
    save_state(pkg_path, state)

    return report

load_or_generate_review_manifest 

load_or_generate_review_manifest(pkg_path: str | Path, compiled: Any) -> ReviewManifest

Load .gaia/review_manifest.json if present, else generate one from the compiled IR.

Source code in gaia/engine/inquiry/review_manifest.py 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
def load_or_generate_review_manifest(pkg_path: str | Path, compiled: Any) -> ReviewManifest:
    """Load `.gaia/review_manifest.json` if present, else generate one from the compiled IR."""
    generated = _generated_manifest(compiled)
    path = Path(pkg_path) / REVIEW_MANIFEST_REL_PATH
    if not path.exists():
        return generated

    try:
        data = json.loads(path.read_text())
        persisted = ReviewManifest.model_validate(data)
    except (OSError, json.JSONDecodeError, ValidationError) as exc:
        raise GaiaPackagingError(f"Error: {path} is not a valid ReviewManifest: {exc}") from exc
    return merge_review_manifests(generated, persisted)

append_tactic_event 

append_tactic_event(pkg_path: str | Path, event: str, payload: dict[str, Any] | None = None) -> None

Append one tactic event to the inquiry audit log.

Source code in gaia/engine/inquiry/state.py 
186
187
188
189
190
191
192
193
194
195
196
197
def append_tactic_event(
    pkg_path: str | Path, event: str, payload: dict[str, Any] | None = None
) -> None:
    """Append one tactic event to the inquiry audit log."""
    rec = {
        "timestamp": _utcnow(),
        "event": event,
        "payload": payload or {},
    }
    p = _tactics_path(pkg_path)
    with p.open("a", encoding="utf-8") as f:
        f.write(json.dumps(rec, ensure_ascii=False) + "\n")

inquiry_dir 

inquiry_dir(pkg_path: str | Path) -> Path

Return the package inquiry directory, creating it if needed.

Source code in gaia/engine/inquiry/state.py 
132
133
134
135
136
def inquiry_dir(pkg_path: str | Path) -> Path:
    """Return the package inquiry directory, creating it if needed."""
    d = Path(pkg_path).resolve() / ".gaia" / "inquiry"
    d.mkdir(parents=True, exist_ok=True)
    return d

load_state 

load_state(pkg_path: str | Path) -> InquiryState

Load the persisted inquiry state, or return the default empty state.

Source code in gaia/engine/inquiry/state.py 
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
def load_state(pkg_path: str | Path) -> InquiryState:
    """Load the persisted inquiry state, or return the default empty state."""
    p = _state_path(pkg_path)
    if not p.exists():
        return InquiryState()
    raw = json.loads(p.read_text(encoding="utf-8"))
    version = int(raw.get("version", 1))
    if version > STATE_SCHEMA_VERSION:
        raise ValueError(
            f"state.json version {version} is newer than supported {STATE_SCHEMA_VERSION}"
        )
    mode = raw.get("mode", "auto")
    if mode not in VALID_MODES:
        raise ValueError(f"invalid mode {mode!r}; allowed: {sorted(VALID_MODES)}")
    obligations = [SyntheticObligation(**o) for o in raw.get("synthetic_obligations", [])]
    hypotheses = [SyntheticHypothesis(**h) for h in raw.get("synthetic_hypotheses", [])]
    rejections = [SyntheticRejection(**r) for r in raw.get("synthetic_rejections", [])]
    return InquiryState(
        version=STATE_SCHEMA_VERSION,
        focus=raw.get("focus"),
        focus_kind=raw.get("focus_kind"),
        focus_resolved_id=raw.get("focus_resolved_id"),
        mode=mode,
        last_review_id=raw.get("last_review_id"),
        baseline_review_id=raw.get("baseline_review_id"),
        focus_stack=list(raw.get("focus_stack", [])),
        synthetic_obligations=obligations,
        synthetic_hypotheses=hypotheses,
        synthetic_rejections=rejections,
    )

mint_qid 

mint_qid(prefix: str) -> str

Mint a short synthetic inquiry identifier with the given prefix.

Source code in gaia/engine/inquiry/state.py 
42
43
44
def mint_qid(prefix: str) -> str:
    """Mint a short synthetic inquiry identifier with the given prefix."""
    return f"{prefix}_{uuid.uuid4().hex[:8]}"

pop_focus_frame 

pop_focus_frame(state: InquiryState) -> dict[str, Any] | None

Pop the top frame and restore it. Returns old current frame for logging.

Source code in gaia/engine/inquiry/state.py 
224
225
226
227
228
229
230
231
232
233
234
235
236
237
def pop_focus_frame(state: InquiryState) -> dict[str, Any] | None:
    """Pop the top frame and restore it. Returns old current frame for logging."""
    if not state.focus_stack:
        return None
    old = {
        "focus": state.focus,
        "focus_kind": state.focus_kind,
        "focus_resolved_id": state.focus_resolved_id,
    }
    restored = state.focus_stack.pop()
    state.focus = restored.get("focus")
    state.focus_kind = restored.get("focus_kind")
    state.focus_resolved_id = restored.get("focus_resolved_id")
    return old

push_focus_frame 

push_focus_frame(state: InquiryState) -> None

Push current focus onto focus_stack. Caller sets the new focus after.

Source code in gaia/engine/inquiry/state.py 
214
215
216
217
218
219
220
221
def push_focus_frame(state: InquiryState) -> None:
    """Push current focus onto focus_stack. Caller sets the new focus after."""
    frame = {
        "focus": state.focus,
        "focus_kind": state.focus_kind,
        "focus_resolved_id": state.focus_resolved_id,
    }
    state.focus_stack.append(frame)

read_tactic_log 

read_tactic_log(pkg_path: str | Path) -> list[dict[str, Any]]

Read the inquiry tactic log as JSON records.

Source code in gaia/engine/inquiry/state.py 
200
201
202
203
204
205
206
207
208
209
210
211
def read_tactic_log(pkg_path: str | Path) -> list[dict[str, Any]]:
    """Read the inquiry tactic log as JSON records."""
    p = _tactics_path(pkg_path)
    if not p.exists():
        return []
    out: list[dict[str, Any]] = []
    for line in p.read_text(encoding="utf-8").splitlines():
        line = line.strip()
        if not line:
            continue
        out.append(json.loads(line))
    return out

save_state 

save_state(pkg_path: str | Path, state: InquiryState) -> None

Persist inquiry state to .gaia/inquiry/state.json.

Source code in gaia/engine/inquiry/state.py 
179
180
181
182
183
def save_state(pkg_path: str | Path, state: InquiryState) -> None:
    """Persist inquiry state to ``.gaia/inquiry/state.json``."""
    p = _state_path(pkg_path)
    payload = state.to_dict()
    p.write_text(json.dumps(payload, indent=2, ensure_ascii=False), encoding="utf-8")