ontocast.stategraph.atomic

Reusable per-unit render/critic retry loops.

These loops are designed for map/reduce execution where each content unit is processed independently. They deep-copy the incoming unit state, then run render -> critic until success or retry exhaustion. After the last allowed render succeeds, the critic is skipped: no further extract exists for feedback to inform.

Ontology context assembly (resolve_unit_ontology_context) runs at the start of both ontology_loop and facts_loop so each unit chooses its own ontology context according to mode/policy.

facts_loop(state, tools, document_context, max_visits_per_node=None, pre_resolved_context=None) async

Run facts render/critic loop for one content unit.

Parameters:

Name	Type	Description	Default
state	UnitFactsState	Unit facts state to run the loop over.	required
tools	ToolBox	Tool container.	required
document_context	UnitLoopContext	Document-level inputs, shared read-only.	required
max_visits_per_node	int | None	Override for the render/critic bound.	None
pre_resolved_context	UnitOntologyContext | None	Ontology context resolved once by the caller. The merged document ontology depends only on document-level state, so the fan-out builds it once and hands the same object to every unit; resolving it here instead cost one full rdflib merge and two graph copies per unit. Falls back to per-unit resolution when None.	None

Source code in ontocast/stategraph/atomic.py

async def facts_loop(
    state: UnitFactsState,
    tools: ToolBox,
    document_context: UnitLoopContext,
    max_visits_per_node: int | None = None,
    pre_resolved_context: UnitOntologyContext | None = None,
) -> UnitFactsState:
    """Run facts render/critic loop for one content unit.

    Args:
        state: Unit facts state to run the loop over.
        tools: Tool container.
        document_context: Document-level inputs, shared read-only.
        max_visits_per_node: Override for the render/critic bound.
        pre_resolved_context: Ontology context resolved once by the caller.
            The merged document ontology depends only on document-level state,
            so the fan-out builds it once and hands the *same object* to every
            unit; resolving it here instead cost one full rdflib merge and two
            graph copies per unit. Falls back to per-unit resolution when None.
    """
    atomic = tools.get_atomic_tools()
    unit_state = state.model_copy(deep=True)
    # Charge resolver LLM calls (e.g. ontology selection) to this unit's
    # tracker — the copy that survives the loop and is merged by the caller.
    # Shallow copy: retrieval_metrics stays shared with the caller's context.
    document_context = document_context.model_copy(
        update={"budget_tracker": unit_state.budget_tracker}
    )
    # The stage the loop is currently in, so an unhandled exception is
    # attributed to where it happened. Hardcoding the critique stage reported
    # a render or context-resolution crash as a failed critique.
    stage = FailureStage.GENERATE_GRAPH_UPDATE_FOR_FACTS
    try:
        if pre_resolved_context is not None:
            _apply_unit_ontology_context(unit_state, pre_resolved_context)
        else:
            unit_state = await _apply_facts_ontology_context(
                unit_state, document_context, tools
            )
        max_visits = _resolve_max_visits_limit(
            unit_state.max_visits_per_node, max_visits_per_node
        )
        unit_state.max_visits_per_node = max_visits

        for render_attempt in range(1, max_visits + 1):
            stage = FailureStage.GENERATE_GRAPH_UPDATE_FOR_FACTS
            unit_state.node_visits[WorkflowNode.TEXT_TO_FACTS] += 1
            _reset_node_evidence_context(unit_state, WorkflowNode.TEXT_TO_FACTS)
            supplemental = _supplemental_ontologies_for_unit(
                document_context, unit_state, tools
            )
            unit_state = await render_facts(
                unit_state, atomic, supplemental_ontologies=supplemental
            )
            _record_facts_attempt(
                unit_state, kind="render", render_attempt=render_attempt
            )
            if unit_state.status != Status.SUCCESS:
                render_request = unit_state.get_external_evidence_request(
                    WorkflowNode.TEXT_TO_FACTS
                )
                if render_request.initiate_search:
                    unit_state = await plan_external_evidence_for_node(
                        unit_state, atomic, WorkflowNode.TEXT_TO_FACTS
                    )
                    unit_state = await fetch_external_evidence_for_node(
                        unit_state, atomic, WorkflowNode.TEXT_TO_FACTS
                    )
                    unit_state = await render_facts(
                        unit_state, atomic, supplemental_ontologies=supplemental
                    )
                    _record_facts_attempt(
                        unit_state, kind="render", render_attempt=render_attempt
                    )
                    if unit_state.status == Status.SUCCESS:
                        logger.info(
                            "Unit facts render recovered with search at attempt %s/%s",
                            render_attempt,
                            max_visits,
                        )
                    else:
                        logger.info(
                            "Unit facts render failed at attempt %s/%s (with search)",
                            render_attempt,
                            max_visits,
                        )
                        continue
                else:
                    logger.info(
                        "Unit facts render failed at attempt %s/%s (no search request)",
                        render_attempt,
                        max_visits,
                    )
                    continue

            if _skip_critic_after_final_render(render_attempt, max_visits):
                logger.info(
                    "Unit facts loop finishing on final render attempt %s/%s "
                    "(skipping LLM critic; finding-driven repair renders may "
                    "still run)",
                    render_attempt,
                    max_visits,
                )
                return await _run_finding_driven_repair(
                    unit_state,
                    atomic,
                    supplemental,
                    render_attempt=render_attempt,
                )

            stage = FailureStage.FACTS_CRITIQUE
            for critic_attempt in range(1, _resolve_critic_visits(unit_state) + 1):
                unit_state.node_visits[WorkflowNode.CRITICISE_FACTS] += 1
                _reset_node_evidence_context(unit_state, WorkflowNode.CRITICISE_FACTS)
                unit_state.deterministic_findings = _collect_facts_findings(
                    unit_state, atomic
                )
                unit_state = await criticise_facts(unit_state, atomic)
                if unit_state.status == Status.SUCCESS:
                    logger.info(
                        "Unit facts loop converged at render %s/%s critic %s/%s",
                        render_attempt,
                        max_visits,
                        critic_attempt,
                        max_visits,
                    )
                    return await _run_finding_driven_repair(
                        unit_state,
                        atomic,
                        supplemental,
                        render_attempt=render_attempt,
                    )

                critic_request = unit_state.get_external_evidence_request(
                    WorkflowNode.CRITICISE_FACTS
                )
                if not critic_request.initiate_search:
                    logger.info(
                        "Unit facts critic rejected at render %s/%s critic %s/%s "
                        "without search request; repairing in place",
                        render_attempt,
                        max_visits,
                        critic_attempt,
                        max_visits,
                    )
                    break

                unit_state = await plan_external_evidence_for_node(
                    unit_state, atomic, WorkflowNode.CRITICISE_FACTS
                )
                unit_state = await fetch_external_evidence_for_node(
                    unit_state, atomic, WorkflowNode.CRITICISE_FACTS
                )
                unit_state = await criticise_facts(unit_state, atomic)
                if unit_state.status == Status.SUCCESS:
                    logger.info(
                        "Unit facts loop converged with critic search at "
                        "render %s/%s critic %s/%s",
                        render_attempt,
                        max_visits,
                        critic_attempt,
                        max_visits,
                    )
                    return await _run_finding_driven_repair(
                        unit_state,
                        atomic,
                        supplemental,
                        render_attempt=render_attempt,
                    )

            # A rejecting critic no longer escalates to another full render.
            # It used to fall through to the next `render_attempt`, which
            # re-extracted the unit from scratch under a prompt that invited
            # unrequested rewriting -- the expensive, open-ended answer to a
            # signal that is now a list of specific defects. Its blocking fixes
            # go through the same bounded rewrite-in-place pass the
            # deterministic findings use. The outer loop therefore retries only
            # on *render failure*, and a unit's worst-case call count no longer
            # grows with MAX_VISITS.
            return await _run_finding_driven_repair(
                unit_state,
                atomic,
                supplemental,
                render_attempt=render_attempt,
                critic_fixes=unit_state.suggestions.actionable_fixes,
            )

        logger.info("Unit facts loop exhausted retries")
        return unit_state
    except Exception as exc:
        logger.exception("Unhandled exception in facts_loop")
        unit_state.set_failure(stage, str(exc))
        return unit_state

ontology_loop(state, tools, document_context, max_visits_per_node=None) async

Run ontology render/critic loop for one content unit.

Per-unit ontology context is assembled via resolve_unit_ontology_context before the first render.

Source code in ontocast/stategraph/atomic.py

async def ontology_loop(
    state: UnitOntologyState,
    tools: ToolBox,
    document_context: UnitLoopContext,
    max_visits_per_node: int | None = None,
) -> UnitOntologyState:
    """Run ontology render/critic loop for one content unit.

    Per-unit ontology context is assembled via ``resolve_unit_ontology_context``
    before the first render.
    """
    atomic = tools.get_atomic_tools()
    unit_state = state.model_copy(deep=True)
    # Charge resolver LLM calls to this unit's surviving tracker; shallow copy
    # keeps retrieval_metrics shared with the caller's context.
    document_context = document_context.model_copy(
        update={"budget_tracker": unit_state.budget_tracker}
    )
    # See facts_loop: the stage an unhandled exception is attributed to tracks
    # where the loop actually is, rather than always naming the critique.
    stage = FailureStage.GENERATE_GRAPH_UPDATE_FOR_ONTOLOGY
    try:
        ctx = await resolve_unit_ontology_context(
            document_context, tools, unit_state.content_unit
        )
        _apply_unit_ontology_context(unit_state, ctx)
        working_copy_start = time.perf_counter()
        unit_state.working_graph = unit_state.ontology_snapshot.graph.copy()
        unit_state.budget_tracker.add_duration(
            "ctx/working_graph_copy", time.perf_counter() - working_copy_start
        )

        max_visits = _resolve_max_visits_limit(
            unit_state.max_visits_per_node, max_visits_per_node
        )
        unit_state.max_visits_per_node = max_visits

        for render_attempt in range(1, max_visits + 1):
            stage = FailureStage.GENERATE_GRAPH_UPDATE_FOR_ONTOLOGY
            unit_state.node_visits[WorkflowNode.TEXT_TO_ONTOLOGY] += 1
            _reset_node_evidence_context(unit_state, WorkflowNode.TEXT_TO_ONTOLOGY)
            supplemental = _supplemental_ontologies_for_unit(
                document_context, unit_state, tools
            )
            unit_state = await render_ontology(
                unit_state, atomic, supplemental_ontologies=supplemental
            )
            if unit_state.status != Status.SUCCESS:
                render_request = unit_state.get_external_evidence_request(
                    WorkflowNode.TEXT_TO_ONTOLOGY
                )
                if render_request.initiate_search:
                    unit_state = await plan_external_evidence_for_node(
                        unit_state, atomic, WorkflowNode.TEXT_TO_ONTOLOGY
                    )
                    unit_state = await fetch_external_evidence_for_node(
                        unit_state, atomic, WorkflowNode.TEXT_TO_ONTOLOGY
                    )
                    unit_state = await render_ontology(
                        unit_state, atomic, supplemental_ontologies=supplemental
                    )
                    if unit_state.status == Status.SUCCESS:
                        logger.info(
                            "Unit ontology render recovered with search at attempt %s/%s",
                            render_attempt,
                            max_visits,
                        )
                    else:
                        logger.info(
                            "Unit ontology render failed at attempt %s/%s (with search)",
                            render_attempt,
                            max_visits,
                        )
                        continue
                else:
                    logger.info(
                        "Unit ontology render failed at attempt %s/%s (no search request)",
                        render_attempt,
                        max_visits,
                    )
                    continue

            if _skip_critic_after_final_render(render_attempt, max_visits):
                logger.info(
                    "Unit ontology loop finishing on final render attempt %s/%s "
                    "(no further extract; skipping critic)",
                    render_attempt,
                    max_visits,
                )
                # The residual metric needs findings even when no critic runs
                # (the MAX_VISITS=1 default).
                unit_state.deterministic_findings = _collect_ontology_findings(
                    unit_state, atomic
                )
                return unit_state

            stage = FailureStage.ONTOLOGY_CRITIQUE
            for critic_attempt in range(1, _resolve_critic_visits(unit_state) + 1):
                unit_state.node_visits[WorkflowNode.CRITICISE_ONTOLOGY] += 1
                _reset_node_evidence_context(
                    unit_state, WorkflowNode.CRITICISE_ONTOLOGY
                )
                unit_state.deterministic_findings = _collect_ontology_findings(
                    unit_state, atomic
                )
                unit_state = await criticise_ontology(unit_state, atomic)
                if unit_state.status == Status.SUCCESS:
                    logger.info(
                        "Unit ontology loop converged at render %s/%s critic %s/%s",
                        render_attempt,
                        max_visits,
                        critic_attempt,
                        max_visits,
                    )
                    return unit_state

                critic_request = unit_state.get_external_evidence_request(
                    WorkflowNode.CRITICISE_ONTOLOGY
                )
                if not critic_request.initiate_search:
                    logger.info(
                        "Unit ontology critic failed at render %s/%s critic %s/%s "
                        "without search request",
                        render_attempt,
                        max_visits,
                        critic_attempt,
                        max_visits,
                    )
                    break

                unit_state = await plan_external_evidence_for_node(
                    unit_state, atomic, WorkflowNode.CRITICISE_ONTOLOGY
                )
                unit_state = await fetch_external_evidence_for_node(
                    unit_state, atomic, WorkflowNode.CRITICISE_ONTOLOGY
                )
                unit_state = await criticise_ontology(unit_state, atomic)
                if unit_state.status == Status.SUCCESS:
                    logger.info(
                        "Unit ontology loop converged with critic search at "
                        "render %s/%s critic %s/%s",
                        render_attempt,
                        max_visits,
                        critic_attempt,
                        max_visits,
                    )
                    return unit_state

        logger.info("Unit ontology loop exhausted retries")
        unit_state.deterministic_findings = _collect_ontology_findings(
            unit_state, atomic
        )
        return unit_state
    except Exception as exc:
        logger.exception("Unhandled exception in ontology_loop")
        unit_state.set_failure(stage, str(exc))
        return unit_state