ontocast.api.process_helpers

Shared helpers for local batch processing and HTTP response assembly.

Attributes

GRAPH_RECURSION_LIMIT = 1000 module-attribute

logger = logging.getLogger(__name__) module-attribute

Classes

Functions:

calculate_recursion_limit(head_chunks, server_config, *, max_visits_per_node=None)

Recursion limit for one document run.

Arguments are accepted and ignored: the graph's depth is a property of its topology, not of the document or the visit budget. See :data:GRAPH_RECURSION_LIMIT.

Source code in ontocast/api/process_helpers.py

def calculate_recursion_limit(
    head_chunks: int | None,
    server_config: ServerConfig,
    *,
    max_visits_per_node: int | None = None,
) -> int:
    """Recursion limit for one document run.

    Arguments are accepted and ignored: the graph's depth is a property of its
    topology, not of the document or the visit budget. See
    :data:`GRAPH_RECURSION_LIMIT`.
    """
    return GRAPH_RECURSION_LIMIT

dump_facts_ttl(state, file_path, *, line_number=None, output_dir=None, strip_provenance=True)

Write the facts Turtle when facts exist.

Parameters:

Name	Type	Description	Default
state	AgentState	Document state carrying aggregated_facts.	required
file_path	Path	Source file the facts were extracted from.	required
line_number	int | None	Record number for JSONL inputs.	None
output_dir	Path | None	Destination directory; defaults to the source's directory.	None
strip_provenance	bool	Drop chunk-level provenance from the dump. Keeping it is what lets a statement be traced back to its source span and re-verified against the document; stripping it stays the default so existing outputs are unchanged. Same meaning as the HTTP strip_provenance parameter.	True

Returns:

Type	Description
Path | None	The path written, or None when there are no facts.

Source code in ontocast/api/process_helpers.py

def dump_facts_ttl(
    state: AgentState,
    file_path: pathlib.Path,
    *,
    line_number: int | None = None,
    output_dir: pathlib.Path | None = None,
    strip_provenance: bool = True,
) -> pathlib.Path | None:
    """Write the facts Turtle when facts exist.

    Args:
        state: Document state carrying ``aggregated_facts``.
        file_path: Source file the facts were extracted from.
        line_number: Record number for JSONL inputs.
        output_dir: Destination directory; defaults to the source's directory.
        strip_provenance: Drop chunk-level provenance from the dump. Keeping it
            is what lets a statement be traced back to its source span and
            re-verified against the document; stripping it stays the default so
            existing outputs are unchanged. Same meaning as the HTTP
            ``strip_provenance`` parameter.

    Returns:
        The path written, or None when there are no facts.
    """
    if state.aggregated_facts is None or len(state.aggregated_facts) == 0:
        return None
    ttl_content = turtle_from_graph(
        state.aggregated_facts, strip_provenance=strip_provenance
    )
    if output_dir is not None:
        output_dir.mkdir(parents=True, exist_ok=True)
    output_path = facts_ttl_output_path(
        file_path, line_number=line_number, output_dir=output_dir
    )
    output_path.write_text(ttl_content, encoding="utf-8")
    logger.info(
        "Dumped facts graph with chunk-level provenance %s to %s",
        "stripped" if strip_provenance else "retained",
        output_path,
    )
    return output_path

dump_ontology_ttls(state, file_path, *, line_number=None, output_dir=None)

Write provenance-stripped ontology Turtle dumps when artifacts exist.

Source code in ontocast/api/process_helpers.py

def dump_ontology_ttls(
    state: AgentState,
    file_path: pathlib.Path,
    *,
    line_number: int | None = None,
    output_dir: pathlib.Path | None = None,
) -> list[pathlib.Path]:
    """Write provenance-stripped ontology Turtle dumps when artifacts exist."""
    artifacts = _ontology_artifacts_for_dump(state)
    if not artifacts:
        return []
    if output_dir is not None:
        output_dir.mkdir(parents=True, exist_ok=True)
    include_ids = len(artifacts) > 1
    written: list[pathlib.Path] = []
    for artifact in artifacts:
        ontology_id = safe_ontology_filename_id(artifact) if include_ids else None
        ttl_content = turtle_from_graph(artifact.graph, strip_provenance=True)
        output_path = ontology_ttl_output_path(
            file_path,
            line_number=line_number,
            output_dir=output_dir,
            ontology_id=ontology_id,
        )
        output_path.write_text(ttl_content, encoding="utf-8")
        logger.info(
            "Dumped ontology graph with provenance stripped to %s",
            output_path,
        )
        written.append(output_path)
    return written

dump_run_manifest(state, file_path, *, config, line_number=None, output_dir=None, shapes_triples=None, shapes_prompt_selection=None, fanout_settings_apply=True)

Write the run's cost and configuration beside the facts TTL.

BudgetTracker is returned over HTTP and logged at INFO, then discarded, so a batch run left no record of the model, the settings, or the tokens behind its own output -- and no way to compare two dumps except by rerunning them. One small JSON per document closes that.

Source code in ontocast/api/process_helpers.py

def dump_run_manifest(
    state: AgentState,
    file_path: pathlib.Path,
    *,
    config: Config,
    line_number: int | None = None,
    output_dir: pathlib.Path | None = None,
    shapes_triples: int | None = None,
    shapes_prompt_selection: bool | None = None,
    fanout_settings_apply: bool = True,
) -> pathlib.Path | None:
    """Write the run's cost and configuration beside the facts TTL.

    ``BudgetTracker`` is returned over HTTP and logged at INFO, then discarded,
    so a batch run left no record of the model, the settings, or the tokens
    behind its own output -- and no way to compare two dumps except by rerunning
    them. One small JSON per document closes that.
    """
    llm_config = config.tool_config.llm_config
    tool_config = config.get_tool_config()
    facts_validation = tool_config.facts_validation
    facts_critic_passes = facts_validation.critic_passes
    # The .facts.ttl dump strips provenance; count what the file will actually
    # hold, or the manifest is not comparable to its own TTL.
    serialized_facts = (
        TripleStoreManager.strip_provenance(state.aggregated_facts)
        if state.aggregated_facts is not None
        else None
    )
    manifest = RunManifest(
        source=file_path.name,
        line_number=line_number,
        ontocast_version=__version__,
        render_mode=str(state.render_mode),
        loops=RunManifestLoops(
            max_visits=state.max_visits,
            facts_critic_passes=facts_critic_passes,
            ontology_critic_passes=tool_config.ontology_validation.critic_passes,
        ),
        critic=summarize_loop(state.facts_loop_telemetry),
        ontology_critic=summarize_loop(state.ontology_loop_telemetry),
        completion=summarize_completion(state.facts_loop_telemetry),
        ontology_reduce_metrics=dict(state.ontology_reduce_metrics),
        selection=_selection_manifest(state, config),
        validation_config=RunManifestValidationConfig(
            context_from_units=facts_validation.context_from_units,
            json_mode=llm_config.json_mode,
            shapes_prompt_contract=facts_validation.shapes_prompt_contract,
            shapes_prompt_selection=shapes_prompt_selection,
            shapes_triples=shapes_triples,
            shacl_inference=str(facts_validation.shacl_inference),
            numeric_coverage_mandatory=facts_validation.numeric_coverage_mandatory,
            facts_user_instruction_chars=len(state.facts_user_instruction or ""),
        ),
        graph_metrics=(
            facts_graph_shape_metrics(
                serialized_facts,
                [DEFAULT_IRI, str(state.doc_iri), state.doc_namespace or ""],
            )
            if serialized_facts is not None
            else None
        ),
        current_domain=state.current_domain,
        doc_iri=str(state.doc_iri) if state.doc_hid else None,
        tenant=state.tenant,
        project=state.project,
        llm=RunManifestLLM(
            provider=str(llm_config.provider),
            model_name=str(llm_config.model_name),
            temperature=llm_config.temperature,
            think=llm_config.think,
            num_ctx=llm_config.num_ctx,
            num_predict=llm_config.num_predict,
            reasoning_effort=llm_config.reasoning_effort,
            thinking_budget=llm_config.thinking_budget,
            requests_per_second=llm_config.requests_per_second,
            max_retries=llm_config.max_retries,
            prompt_cache_key=llm_config.prompt_cache_key,
            max_inflight=llm_config.llm_max_inflight,
        ),
        prompting=RunManifestPrompting(
            llm_graph_format=str(config.server.llm_graph_format),
            llm_output_layout=str(config.server.llm_output_layout),
            ontology_chapter_format=str(config.server.ontology_chapter_format),
            # Null on the single-unit path rather than echoed back: both
            # settings are read only by the document fan-out node, so recording
            # a scope that never applied would make a unit-path manifest assert
            # a prompt-sharing regime it did not run under.
            ontology_context_scope=(
                str(config.server.ontology_context_scope)
                if fanout_settings_apply
                else None
            ),
            fanout_warmup_units=(
                config.server.fanout_warmup_units if fanout_settings_apply else None
            ),
            parallel_workers=config.server.parallel_workers,
            embedding_model_name=tool_config.embedding.model_name,
        ),
        budget=state.budget_tracker,
        ontology_triples=sum(
            len(artifact.graph) for artifact in _ontology_artifacts_for_dump(state)
        ),
        facts_triples=(
            len(state.aggregated_facts) if state.aggregated_facts is not None else 0
        ),
        facts_triples_serialized=(
            len(serialized_facts) if serialized_facts is not None else 0
        ),
        retrieval_metrics=dict(state.retrieval_metrics),
    )
    if output_dir is not None:
        output_dir.mkdir(parents=True, exist_ok=True)
    facts_path = facts_ttl_output_path(
        file_path, line_number=line_number, output_dir=output_dir
    )
    output_path = facts_path.with_name(
        f"{facts_path.name.removesuffix('.facts.ttl')}.run.json"
    )
    output_path.write_text(
        manifest.model_dump_json(indent=2, exclude_none=True), encoding="utf-8"
    )
    logger.info("Dumped run manifest to %s", output_path)
    return output_path

dump_validation_report(state, file_path, *, line_number=None, output_dir=None)

Write the conformance summary and residual findings beside the facts TTL.

A batch run otherwise leaves no record of why a graph is non-conformant: the findings live on the state and are logged, and every downstream reader ends up re-running a validator to rebuild what the gate already computed.

unit_repairs and unit_failures are the per-unit counterparts of gate_repairs: what the deterministic passes rewrote in each render before aggregation, and which units produced nothing. A predicate the machine substituted is otherwise indistinguishable in the TTL from one the model asserted, and a unit that failed from one that found nothing.

Source code in ontocast/api/process_helpers.py

def dump_validation_report(
    state: AgentState,
    file_path: pathlib.Path,
    *,
    line_number: int | None = None,
    output_dir: pathlib.Path | None = None,
) -> pathlib.Path | None:
    """Write the conformance summary and residual findings beside the facts TTL.

    A batch run otherwise leaves no record of *why* a graph is non-conformant:
    the findings live on the state and are logged, and every downstream reader
    ends up re-running a validator to rebuild what the gate already computed.

    ``unit_repairs`` and ``unit_failures`` are the per-unit counterparts of
    ``gate_repairs``: what the deterministic passes rewrote in each render
    before aggregation, and which units produced nothing. A predicate the
    machine substituted is otherwise indistinguishable in the TTL from one the
    model asserted, and a unit that failed from one that found nothing.
    """
    if not (
        state.facts_conformance
        or state.facts_validation_findings
        or state.facts_repairs_applied
        or state.unit_failures
    ):
        return None
    payload = {
        "source": file_path.name,
        "conformance": state.facts_conformance,
        "gate_repairs": [
            record.model_dump(mode="json") for record in state.facts_gate_repairs
        ],
        "findings": [
            finding.model_dump(mode="json")
            for finding in state.facts_validation_findings
        ],
        # Keyed by unit index, as facts_repairs is in the HTTP response.
        "unit_repairs": {
            str(unit_index): [record.model_dump(mode="json") for record in records]
            for unit_index, records in sorted(state.facts_repairs_applied.items())
        },
        "unit_failures": [
            failure.model_dump(mode="json") for failure in state.unit_failures
        ],
    }
    if output_dir is not None:
        output_dir.mkdir(parents=True, exist_ok=True)
    facts_path = facts_ttl_output_path(
        file_path, line_number=line_number, output_dir=output_dir
    )
    output_path = facts_path.with_name(
        f"{facts_path.name.removesuffix('.ttl')}.validation.json"
    )
    output_path.write_text(json.dumps(payload, indent=2), encoding="utf-8")
    logger.info("Dumped facts validation report to %s", output_path)
    return output_path

expand_input_to_states(file_path, *, config, head_chunks, ontology_context_mode_value, tenant, project, target_sections=None, exclude_sections=None, summarize_sections=None, summary_max_sentences=5, document_type_hint=None, section_schema_id=None, max_visits=None, document_metadata=None, facts_user_instruction='')

Expand a local input file into one AgentState per logical record.

Source code in ontocast/api/process_helpers.py

def expand_input_to_states(
    file_path: pathlib.Path,
    *,
    config: Config,
    head_chunks: int | None,
    ontology_context_mode_value: OntologyContextMode,
    tenant: str | None,
    project: str | None,
    target_sections: list[str] | None = None,
    exclude_sections: list[str] | None = None,
    summarize_sections: list[str] | None = None,
    summary_max_sentences: int = 5,
    document_type_hint: str | None = None,
    section_schema_id: str | None = None,
    max_visits: int | None = None,
    document_metadata: dict[str, object] | None = None,
    facts_user_instruction: str = "",
) -> list[AgentState]:
    """Expand a local input file into one ``AgentState`` per logical record."""
    file_bytes = file_path.read_bytes()
    resolved_max_visits = (
        max_visits if max_visits is not None else config.server.max_visits_per_node
    )
    # Explicitly typed: the splat below is only checkable if the mapping's
    # value type is known, and a heterogeneous literal infers as a union that
    # matches no single field.
    base_state_kwargs: dict[str, Any] = {
        "max_visits": resolved_max_visits,
        "max_chunks": head_chunks,
        "render_mode": config.server.render_mode,
        "llm_graph_format": config.server.llm_graph_format,
        "ontology_context_mode": ontology_context_mode_value,
        "ontology_context_fixed_ontology_id": (
            config.server.ontology_context_fixed_ontology_id
        ),
        "tenant": tenant,
        "project": project,
        "target_sections": target_sections,
        "exclude_sections": exclude_sections,
        "summarize_sections": summarize_sections,
        "summary_max_sentences": summary_max_sentences,
        "document_type_hint": document_type_hint,
        "section_schema_id": section_schema_id,
        "facts_user_instruction": facts_user_instruction,
    }

    if file_path.suffix.lower() != ".jsonl":
        return [
            AgentState(
                raw_input={file_path.as_posix(): file_bytes},
                document_metadata=_resolve_document_metadata(
                    file_path, document_metadata
                ),
                **base_state_kwargs,
            )
        ]

    states: list[AgentState] = []
    for line_number, line in enumerate(
        file_bytes.decode("utf-8").splitlines(), start=1
    ):
        if not line.strip():
            continue
        virtual_path = f"{file_path.as_posix()}:{line_number}.json"
        states.append(
            AgentState(
                raw_input={virtual_path: line.encode("utf-8")},
                document_metadata=_resolve_document_metadata(
                    file_path,
                    document_metadata,
                    line_number=line_number,
                ),
                **base_state_kwargs,
            )
        )
    return states

facts_ttl_output_path(file_path, *, line_number=None, output_dir=None)

Return the .facts.ttl path for a processed input file.

Source code in ontocast/api/process_helpers.py

def facts_ttl_output_path(
    file_path: pathlib.Path,
    *,
    line_number: int | None = None,
    output_dir: pathlib.Path | None = None,
) -> pathlib.Path:
    """Return the ``.facts.ttl`` path for a processed input file."""
    name = _ttl_basename(file_path, line_number=line_number, kind="facts")
    if output_dir is not None:
        return output_dir / name
    return file_path.with_name(name)

flush_triple_configured_scope(tools) async

Match POST /flush without tenant/project: triple store only, current scope.

Source code in ontocast/api/process_helpers.py

async def flush_triple_configured_scope(tools: ToolBox) -> None:
    """Match POST /flush without tenant/project: triple store only, current scope."""
    if tools.triple_store_manager is not None:
        await tools.triple_store_manager.clean()

get_batch_input_extensions(tools)

Return the suffixes ontocast process reads: conversion's plus JSONL.

A JSONL file is fanned out into one document per line before conversion, which only the batch path does.

Source code in ontocast/api/process_helpers.py

def get_batch_input_extensions(tools: ToolBox) -> tuple[str, ...]:
    """Return the suffixes ``ontocast process`` reads: conversion's plus JSONL.

    A JSONL file is fanned out into one document per line before conversion,
    which only the batch path does.
    """
    return tuple(sorted({*get_supported_input_extensions(tools), ".jsonl"}))

get_supported_input_extensions(tools)

Return the file suffixes document conversion accepts (one document per file).

Source code in ontocast/api/process_helpers.py

def get_supported_input_extensions(tools: ToolBox) -> tuple[str, ...]:
    """Return the file suffixes document conversion accepts (one document per file)."""
    built_in_suffixes = {".json", ".txt"}
    converter_suffixes = set(tools.converter.supported_extensions)
    return tuple(sorted(built_in_suffixes | converter_suffixes))

ontology_ttl_output_path(file_path, *, line_number=None, output_dir=None, ontology_id=None)

Return the .ontology.ttl path for a processed input file.

Source code in ontocast/api/process_helpers.py

def ontology_ttl_output_path(
    file_path: pathlib.Path,
    *,
    line_number: int | None = None,
    output_dir: pathlib.Path | None = None,
    ontology_id: str | None = None,
) -> pathlib.Path:
    """Return the ``.ontology.ttl`` path for a processed input file."""
    name = _ttl_basename(
        file_path,
        line_number=line_number,
        kind="ontology",
        ontology_id=ontology_id,
    )
    if output_dir is not None:
        return output_dir / name
    return file_path.with_name(name)

persist_unit_pipeline_outputs(state, onto_result, facts_result, tools) async

Serialize unit-pipeline outputs using the standard document serializer.

Source code in ontocast/api/process_helpers.py

async def persist_unit_pipeline_outputs(
    state: AgentState,
    onto_result,
    facts_result,
    tools: ToolBox,
) -> None:
    """Serialize unit-pipeline outputs using the standard document serializer."""
    if onto_result is not None:
        if (
            onto_result.fresh_ontology is not None
            and not onto_result.fresh_ontology.is_null()
        ):
            state.reduced_ontology_artifacts = [onto_result.fresh_ontology]
    if facts_result is not None:
        ontology_graph = select_unit_facts_ontology_graph(onto_result, facts_result)
        state.aggregated_facts = tools.aggregator.postprocess_facts_units(
            units=[facts_result.content_unit],
            ontology_graph=ontology_graph,
            doc_iri=state.doc_iri,
            document_metadata=_effective_document_metadata(state),
            doc_namespace=state.doc_namespace,
        ).graph
        validate_unit_pipeline_facts(state, ontology_graph, tools)
    await asyncio.to_thread(serialize_agent_state, state, tools)

process_files_input(files, *, config, head_chunks, use_unit_pipeline, tools, workflow, ontology_context_mode_value, tenant, project, target_sections=None, exclude_sections=None, summarize_sections=None, summary_max_sentences=5, document_type_hint=None, section_schema_id=None, max_visits=None, document_metadata=None, facts_user_instruction='', output_dir=None, facts_output_dir=None, ontology_output_dir=None, strip_provenance=True) async

Process each input file, isolating per-file failures.

Returns:

Type	Description
list[Path]	The files that failed, in input order. Empty on full success. Callers
list[Path]	use this to set a non-zero exit code -- previously every failure was
list[Path]	logged and swallowed, so ontocast process exited 0 even when no
list[Path]	file produced any output.

Source code in ontocast/api/process_helpers.py

async def process_files_input(
    files: list[pathlib.Path],
    *,
    config: Config,
    head_chunks: int | None,
    use_unit_pipeline: bool,
    tools: ToolBox,
    workflow: CompiledStateGraph,
    ontology_context_mode_value: OntologyContextMode,
    tenant: str | None,
    project: str | None,
    target_sections: list[str] | None = None,
    exclude_sections: list[str] | None = None,
    summarize_sections: list[str] | None = None,
    summary_max_sentences: int = 5,
    document_type_hint: str | None = None,
    section_schema_id: str | None = None,
    max_visits: int | None = None,
    document_metadata: dict[str, object] | None = None,
    facts_user_instruction: str = "",
    output_dir: pathlib.Path | None = None,
    facts_output_dir: pathlib.Path | None = None,
    ontology_output_dir: pathlib.Path | None = None,
    strip_provenance: bool = True,
) -> list[pathlib.Path]:
    """Process each input file, isolating per-file failures.

    Returns:
        The files that failed, in input order. Empty on full success. Callers
        use this to set a non-zero exit code -- previously every failure was
        logged and swallowed, so ``ontocast process`` exited 0 even when no
        file produced any output.
    """
    failed_files: list[pathlib.Path] = []
    resolved_max_visits = (
        max_visits if max_visits is not None else config.server.max_visits_per_node
    )
    recursion_limit = calculate_recursion_limit(
        head_chunks,
        config.server,
        max_visits_per_node=resolved_max_visits,
    )
    facts_dir, ontology_dir = resolve_batch_output_dirs(
        output_dir, facts_output_dir, ontology_output_dir
    )
    for file_path in files:
        try:
            states = expand_input_to_states(
                file_path,
                config=config,
                head_chunks=head_chunks,
                ontology_context_mode_value=ontology_context_mode_value,
                tenant=tenant,
                project=project,
                target_sections=target_sections,
                exclude_sections=exclude_sections,
                summarize_sections=summarize_sections,
                summary_max_sentences=summary_max_sentences,
                document_type_hint=document_type_hint,
                section_schema_id=section_schema_id,
                max_visits=resolved_max_visits,
                document_metadata=document_metadata,
                facts_user_instruction=facts_user_instruction,
            )
            for state_index, state in enumerate(states):
                if use_unit_pipeline:
                    try:
                        onto_result, facts_result = await run_unit_pipeline(
                            state, tools
                        )
                    except DocumentConversionError as exc:
                        logger.error("Error processing %s: %s", file_path, exc)
                        if file_path not in failed_files:
                            failed_files.append(file_path)
                        continue
                    await persist_unit_pipeline_outputs(
                        state, onto_result, facts_result, tools
                    )
                else:
                    workflow_state: AgentState | dict | None = None
                    try:
                        async for chunk in workflow.astream(
                            state,
                            stream_mode="values",
                            config=RunnableConfig(recursion_limit=recursion_limit),
                        ):
                            workflow_state = chunk
                    except SectionSelectionEmptyError as exc:
                        # Batch semantics: one unmatched selection must not kill
                        # the other files. cli/server.py turns a non-empty
                        # failed_files into a non-zero exit, so the `error` mode
                        # is scriptable without new CLI code.
                        logger.error("Error processing %s: %s", file_path, exc)
                        if file_path not in failed_files:
                            failed_files.append(file_path)
                        continue
                    if workflow_state is not None:
                        state = _merge_workflow_state_into_agent_state(
                            state, workflow_state
                        )
                if state.status == Status.FAILED:
                    # Every unit failed. The dumps below still run -- an empty
                    # facts graph next to its manifest is the diagnostic -- but
                    # the file is on the record as failed, which cli/server.py
                    # turns into a non-zero exit. Without this a run that
                    # extracted nothing was indistinguishable from a clean one.
                    logger.error(
                        "No unit of %s produced output; recording it as failed",
                        file_path,
                    )
                    if file_path not in failed_files:
                        failed_files.append(file_path)
                line_number: int | None = None
                if file_path.suffix.lower() == ".jsonl" and len(states) > 1:
                    # Recover line from virtual raw_input key "...:N.json"
                    raw_key = next(iter(state.raw_input), "")
                    marker = f"{file_path.as_posix()}:"
                    if raw_key.startswith(marker) and raw_key.endswith(".json"):
                        try:
                            line_number = int(raw_key[len(marker) : -len(".json")])
                        except ValueError:
                            line_number = state_index + 1
                    else:
                        line_number = state_index + 1
                dump_facts_ttl(
                    state,
                    file_path,
                    line_number=line_number,
                    output_dir=facts_dir,
                    strip_provenance=strip_provenance,
                )
                dump_ontology_ttls(
                    state,
                    file_path,
                    line_number=line_number,
                    output_dir=ontology_dir,
                )
                dump_validation_report(
                    state,
                    file_path,
                    line_number=line_number,
                    output_dir=facts_dir,
                )
                shapes_graph = tools.shapes_catalog.graph()
                _, _, selection_pending = tools.shapes_prompt_contract()
                dump_run_manifest(
                    state,
                    file_path,
                    config=config,
                    line_number=line_number,
                    output_dir=facts_dir,
                    shapes_triples=(
                        len(shapes_graph) if shapes_graph is not None else 0
                    ),
                    shapes_prompt_selection=selection_pending,
                    fanout_settings_apply=not use_unit_pipeline,
                )
        except LLMConfigurationError:
            # Batch semantics stop here: the provider refuses the request as
            # configured, so every remaining file would burn its conversion and
            # ontology sync to reach the same rejection. Files already finished
            # keep their dumps; this one gets none, and that absence is the
            # signal.
            logger.error(
                "Aborting the batch at %s: the provider rejects every request",
                file_path,
            )
            raise
        except Exception:
            logger.exception("Error processing %s", file_path)
            if file_path not in failed_files:
                failed_files.append(file_path)

    return failed_files

resolve_batch_output_dirs(output_dir, facts_output_dir, ontology_output_dir)

Resolve facts/ontology dump dirs from shared and override flags.

Returns:

Type	Description
tuple[Path | None, Path | None]	(facts_dir, ontology_dir). None means sibling-of-input.

Source code in ontocast/api/process_helpers.py

def resolve_batch_output_dirs(
    output_dir: pathlib.Path | None,
    facts_output_dir: pathlib.Path | None,
    ontology_output_dir: pathlib.Path | None,
) -> tuple[pathlib.Path | None, pathlib.Path | None]:
    """Resolve facts/ontology dump dirs from shared and override flags.

    Returns:
        ``(facts_dir, ontology_dir)``. ``None`` means sibling-of-input.
    """
    facts_dir = facts_output_dir or output_dir
    ontology_dir = ontology_output_dir or output_dir
    return facts_dir, ontology_dir

safe_ontology_filename_id(ontology)

Return a filesystem-safe ontology id fragment, or None if unavailable.

Source code in ontocast/api/process_helpers.py

def safe_ontology_filename_id(ontology: Ontology) -> str | None:
    """Return a filesystem-safe ontology id fragment, or None if unavailable."""
    raw = ontology.ontology_id
    if not raw and ontology.iri:
        raw = ontology.iri.rstrip("/").rsplit("/", 1)[-1]
    if not raw:
        return None
    cleaned = _SAFE_ONTOLOGY_ID_RE.sub("_", raw).strip("._-")
    return cleaned or None

select_unit_facts_ontology_graph(onto_result, facts_result)

Return ontology graph for facts post-processing in unit pipeline flows.

Source code in ontocast/api/process_helpers.py

def select_unit_facts_ontology_graph(onto_result, facts_result) -> RDFGraph:
    """Return ontology graph for facts post-processing in unit pipeline flows."""
    if facts_result is not None:
        return facts_result.ontology_snapshot.graph
    if onto_result is not None:
        if (
            onto_result.fresh_ontology is not None
            and not onto_result.fresh_ontology.is_null()
            and len(onto_result.fresh_ontology.graph) > 0
        ):
            return onto_result.fresh_ontology.graph
        if len(onto_result.working_graph) > 0:
            return onto_result.working_graph
        if not onto_result.ontology_snapshot.is_empty():
            return onto_result.ontology_snapshot.graph
    return RDFGraph()

turtle_from_graph(graph, *, strip_provenance)

Serialize graph to Turtle, optionally stripping reification/provenance.

Source code in ontocast/api/process_helpers.py

def turtle_from_graph(graph: RDFGraph, *, strip_provenance: bool) -> str:
    """Serialize ``graph`` to Turtle, optionally stripping reification/provenance."""
    out: RDFGraph = (
        TripleStoreManager.strip_provenance(graph) if strip_provenance else graph
    )
    return out.serialize_canonical_turtle()

validate_unit_pipeline_facts(state, ontology_graph, tools)

Run the post-aggregation invariant gate for the single-unit path.

The document graph reaches this gate at VALIDATE_FACTS; the unit pipeline does not run the graph, so both single-unit callers -- the CLI --use-unit-pipeline batch path and the /process_unit route -- invoke it here after aggregation. Without it they would ship facts with no functional-violation, coreference, or SHACL check at all.

merge_repair=False: un-merging re-aggregates retained units against each other, which has no meaning for a single unit. Everything else is the document path verbatim, so batch dumps stay comparable across the two entry paths.

Source code in ontocast/api/process_helpers.py

def validate_unit_pipeline_facts(
    state: AgentState,
    ontology_graph: RDFGraph,
    tools: ToolBox,
) -> None:
    """Run the post-aggregation invariant gate for the single-unit path.

    The document graph reaches this gate at VALIDATE_FACTS; the unit pipeline
    does not run the graph, so both single-unit callers -- the CLI
    ``--use-unit-pipeline`` batch path and the ``/process_unit`` route -- invoke
    it here after aggregation. Without it they would ship facts with no
    functional-violation, coreference, or SHACL check at all.

    ``merge_repair=False``: un-merging re-aggregates *retained units against
    each other*, which has no meaning for a single unit. Everything else is the
    document path verbatim, so batch dumps stay comparable across the two entry
    paths.
    """
    run_facts_gate(state, ontology_graph, tools, merge_repair=False)