Python API reference

Version 0.2.2.

The ocura_oss package exposes typed workflow functions, read-oriented state access, frozen result and record types, enums, and public exceptions.

Names listed in ocura_oss.__all__ form the documented top-level API. Internal record writers and implementation helpers remain package internals.

API status

The Python API is provisional during the 0.x series. Minor releases may refine signatures, result fields, and record semantics before 1.0.

Ocura OSS supports CPython 3.11 through 3.14 and includes a py.typed marker. The runtime uses only the Python standard library.

Optional root parameters accept a path-like object, a string, or None. None resolves to the current working directory. Relative paths and ~ are resolved without changing the process working directory.

File operations can propagate OSError when a rejected read, write, replacement, or directory operation remains an operating-system error. run_demo() wraps stage failures in DemoError.

A crafted pathway lineage that exceeds Python's recursion limit can propagate RecursionError from programmatic loading, verification, branching, or comparison. The CLI converts that condition to exit status 2.

Declared parameter mappings require nonempty string values. Keys must begin with a letter or underscore and may then contain letters, digits, underscores, periods, or hyphens. Values are stored without type conversion.

Workflow example

import sys
from pathlib import Path

from ocura_oss import ComparisonState, branch, compare, initialize, run, verify

root = Path("experiment")
initial = initialize(root, name="batch study")

baseline = run(
    [sys.executable, "-c", "print('baseline')"],
    root=root,
    pathway_id=initial.pathway.id,
    parameters={"batch": "1"},
)

child = branch(
    baseline.chokepoint.id,
    root=root,
    reason="increase batch",
    parameters={"batch": "4"},
)

run(
    [sys.executable, "-c", "print('child')"],
    root=root,
    pathway_id=child.id,
    parameters={"batch": "4"},
)

comparison = compare(baseline.chokepoint.id, root=root)

assert comparison.state is ComparisonState.READY
assert verify(root).ok

Workflow functions

ocura_oss.initialize

initialize(
    root: os.PathLike[str] | str | None = None,
    *,
    name: str = "ocura-oss",
) -> Initialization

Create one den and one default pathway under a project root.

Parameters

Returns

Initialization containing the resolved root, created den, and default pathway.

Raises

Notes

Initialization fails if .ocura-oss/ already exists, including when the directory contains incomplete state. All created state remains under the selected project root.

Example

from ocura_oss import initialize

initial = initialize("experiment", name="optimizer study")
print(initial.pathway.id)

ocura_oss.run

run(
    command: Sequence[str],
    *,
    root: os.PathLike[str] | str | None = None,
    pathway_id: str | None = None,
    parameters: Mapping[str, str] | None = None,
    mirror: bool = False,
) -> RunExecution

Run one trusted local command and record terminal evidence.

Parameters

Returns

RunExecution containing the recorded Atom and terminal Chokepoint.

Raises

Child command failures and launch failures are returned as atom outcomes rather than raised as exceptions.

Outcome mapping

Launch categories include executable_not_found, permission_denied, not_a_directory, invalid_argument, and os_error.

Notes

The child runs with shell=False and the project root as its working directory. It inherits the invoking process environment, while the inherited environment remains outside recorded state.

Stdout and stderr are stored separately under .ocura-oss/logs/. One atom and one branchable terminal chokepoint are written after the command ends. A second Ctrl+C may exit immediately before that attempt is recorded.

The host machine, network, and process tree form the execution context. The trust model is trusted, same-owner local work.

Example

import sys

from ocura_oss import Outcome, run

execution = run(
    [sys.executable, "-c", "print('recorded')"],
    root="experiment",
    parameters={"batch": "1"},
    mirror=True,
)

assert execution.atom.outcome is Outcome.PASSED

ocura_oss.branch

branch(
    source_chokepoint_id: str,
    *,
    reason: str,
    root: os.PathLike[str] | str | None = None,
    parameters: Mapping[str, str] | None = None,
) -> Pathway

Create a metadata-only child pathway from verified terminal evidence.

Parameters

Returns

The created child Pathway. It has no atom until a later run() explicitly names its ID.

Raises

Notes

Source verification covers the chokepoint, atom, pathway, their relationships, and both referenced logs before the write occurs.

Leading and trailing whitespace is removed from reason before storage. The resulting pathway contains lineage metadata only: parent, source chokepoint, reason, creation time, and effective parameters.

Example

from ocura_oss import branch

child = branch(
    "chokepoint-<id>",
    root="experiment",
    reason="increase batch",
    parameters={"batch": "4"},
)

ocura_oss.compare

compare(
    source_chokepoint_id: str | None = None,
    *,
    root: os.PathLike[str] | str | None = None,
) -> ComparisonResult

Compare a verified source run with the newest run on each child pathway created from its chokepoint.

Parameters

Returns

ComparisonResult with the source summary, ordered child comparisons, and a ComparisonState.

Raises

Notes

Explicit selection verifies the pinned source and relevant child evidence. Automatic selection requires complete state verification and selects the newest chokepoint referenced by a child pathway. A state with no child pathways selects its newest terminal chokepoint and reports NO_BRANCH. Any verification problem causes automatic selection to fail.

Comparison is a read-only operation over recorded state. For a child with more than one atom, it selects the newest atom by start time and identifier.

ComparisonState.READY means every child has evidence. PARTIAL means at least one child lacks evidence. NO_BRANCH means the source has no child pathways.

Version 0.2.2 parameter deltas contain inherited, added, and changed values.

Example

from ocura_oss import ComparisonState, compare

result = compare("chokepoint-<id>", root="experiment")
if result.state is ComparisonState.PARTIAL:
    missing = [child.pathway_id for child in result.children if child.missing_evidence]

ocura_oss.verify

verify(
    root: os.PathLike[str] | str | None = None,
) -> StateVerification

Recheck every state record and every log referenced by a recorded run.

Parameters

Returns

StateVerification. report.ok is true when report.problems is empty. The report contains record counts and the number of individual logs that passed verification.

Raises

Problems in other records and logs are normally collected in the returned report instead of raised.

Notes

Verification checks record envelopes, checksums, identifiers, filenames, required fields, semantic relationships, lineage, referenced log containment, byte counts, log digests, orphaned log files, and unexpected directories under logs/.

Verification establishes local consistency among records and logs. Authorship and execution-environment reproduction require separate evidence.

Example

from ocura_oss import verify

report = verify("experiment")
for record, problem in report.problems:
    print(record, problem)

ocura_oss.run_demo

run_demo(
    root: os.PathLike[str] | str,
) -> DemoReport

Run the complete retained demonstration in a new directory.

Parameters

Returns

DemoReport containing the retained root, completed stages, and a ready comparison.

Raises

Notes

The directory is retained on success and failure. Both demonstration commands must pass and the final state must verify before a report is returned.

Stages are initialize, baseline-run, chokepoint, branch, rerun, verify, and compare.

Example

from ocura_oss import run_demo

report = run_demo("ocura-oss-demo")
print(report.comparison.state.value)

Store

ocura_oss.Store

Store(
    root: os.PathLike[str] | str | None = None,
)

Read and verify Ocura OSS state under one project root.

The documented constructor, attributes, loading methods, listing methods, initialization method, and verification methods form the provisional low-level API. Direct record writers are internal.

Parameters

Attributes

Constructing a Store resolves paths only. exists(), require(), and initialize_state() provide the state-presence operations.

Example

from ocura_oss import Store

store = Store("experiment")
if store.exists():
    print(store.load_den().name)

Store.exists

store.exists() -> bool

Return whether the .ocura-oss/ state directory exists.

Returns

True when the state directory exists, including when its contents are incomplete. Use require() or verify_state() for stronger checks.

Store.require

store.require() -> None

Require an initialized den under the project root.

Returns

None.

Raises

This method checks den.json presence only. load_den() parses the den, and verify_state() performs complete verification.

Store.load_den

store.load_den() -> Den

Load and validate the den record.

Returns

The parsed Den.

Raises

Store.load_pathway

store.load_pathway(
    pathway_id: str,
) -> Pathway

Load a pathway and validate its den and lineage references.

Parameters

Returns

The parsed Pathway.

Raises

Child pathways must set parent and source fields together, and their source chokepoint must belong to the parent pathway.

Store.load_atom

store.load_atom(
    atom_id: str,
) -> Atom

Load an atom and validate its pathway reference.

Parameters

Returns

The parsed Atom.

Raises

This method validates the atom and its log metadata fields. verify_atom_evidence() reads and verifies the referenced files.

Store.load_chokepoint

store.load_chokepoint(
    chokepoint_id: str,
) -> Chokepoint

Load a chokepoint and validate its atom, pathway, and outcome.

Parameters

Returns

The parsed Chokepoint.

Raises

verify_atom_evidence() provides the corresponding log-file verification.

Store.list_pathways

store.list_pathways() -> list[Pathway]

Return pathway records in deterministic creation order.

Returns

A list sorted by created_at, then id.

Raises

The method fails closed instead of skipping an invalid record. Use load_pathway() or verify_state() when reference validation is required.

Store.list_atoms

store.list_atoms() -> list[Atom]

Return atom records in deterministic start order.

Returns

A list sorted by started_at, then id.

Raises

load_atom() adds pathway validation. verify_state() adds complete relationship and log verification.

Store.list_chokepoints

store.list_chokepoints() -> list[Chokepoint]

Return chokepoint records in reverse deterministic creation order.

Returns

A list sorted by created_at, then id, both descending.

Raises

load_chokepoint() adds atom relationship validation. verify_state() adds complete relationship and log verification.

Store.has_terminal_evidence

store.has_terminal_evidence(
    pathway_id: str,
) -> bool

Return whether any listed atom records the supplied pathway ID.

Parameters

Returns

True when at least one atom names the pathway, otherwise False.

Raises

StoreError if atom listing fails.

The result reflects atom membership only. Pathway loading and log verification are separate operations.

Store.resolve_log_path

store.resolve_log_path(
    atom: Atom,
    stream: Literal["stdout", "stderr"],
) -> pathlib.Path

Resolve one recorded log path after containment checks.

Parameters

Returns

The resolved pathlib.Path directly inside this store's .ocura-oss/logs/ directory.

Raises

The returned path has passed containment checks only. verify_atom_evidence() adds existence, byte-count, and digest verification before evidence is read.

Store.verify_atom_evidence

store.verify_atom_evidence(
    atom: Atom,
) -> None

Verify both log files referenced by one atom.

Parameters

Returns

None when both logs pass verification.

Raises

The checksum supports local change detection. Authorship requires separate authenticated evidence.

Store.initialize_state

store.initialize_state(
    *,
    name: str,
) -> tuple[Den, Pathway]

Create one den and one default pathway.

Parameters

Returns

A (den, pathway) tuple.

Raises

Use top-level initialize() when the resolved root should be included in the result.

Store.verify_state

store.verify_state() -> StateVerification

Verify every state record and every log referenced by a valid atom.

Returns

StateVerification. Problems outside the den are collected by record name when possible. logs_checked counts individual log files that passed verification.

Raises

Notes

The scan reports invalid record files, broken relationships, invalid lineage, source-parent disagreement, missing or changed logs, orphaned files, and unexpected directories under logs/.

Record types

Public records and results are frozen dataclasses. Their fields cannot be reassigned. Mapping fields should be treated as read-only values.

The dataclasses describe values returned by workflows and Store. Direct construction creates an in-memory value only; workflow functions and Store loading provide persistence and validation.

Identifiers use a kind prefix followed by 32 lowercase hexadecimal characters. Timestamps are normalized UTC ISO 8601 strings with offsets.

ocura_oss.Den

Den(
    id: str,
    name: str,
    created_at: str,
    default_pathway_id: str,
)

Project-local state identity and default pathway.

Attributes

ocura_oss.Pathway

Pathway(
    id: str,
    den_id: str,
    created_at: str,
    parent_pathway_id: str | None,
    source_chokepoint_id: str | None,
    reason: str,
    parameters: Mapping[str, str],
)

One lineage of declared parameters and recorded evidence.

Attributes

Parent and source fields are either both set or both None.

ocura_oss.Atom

Atom(
    id: str,
    pathway_id: str,
    started_at: str,
    finished_at: str,
    duration_seconds: float,
    outcome: Outcome,
    return_code: int | None,
    launch_error_category: str | None,
    declared_parameters: Mapping[str, str],
    command: tuple[str, ...],
    stdout_log: str,
    stderr_log: str,
    stdout_bytes: int,
    stderr_bytes: int,
    stdout_sha256: str,
    stderr_sha256: str,
)

One recorded command attempt and its referenced output logs.

Attributes

Outcome, return code, and launch category must satisfy the invariants described by run().

ocura_oss.Chokepoint

Chokepoint(
    id: str,
    pathway_id: str,
    atom_id: str,
    created_at: str,
    kind: str,
    outcome: Outcome,
    branchable: bool,
)

Terminal evidence boundary that can serve as a branch source.

Attributes

ocura_oss.Initialization

Initialization(
    root: pathlib.Path,
    den: Den,
    pathway: Pathway,
)

State created by initialize().

Attributes

ocura_oss.RunExecution

RunExecution(
    atom: Atom,
    chokepoint: Chokepoint,
)

Records produced by one command attempt.

Attributes

ocura_oss.RunSummary

RunSummary(
    id: str,
    pathway_id: str,
    outcome: Outcome,
    started_at: str,
    duration_seconds: float,
    return_code: int | None,
)

Reduced command fields used in comparisons and public summaries.

Attributes

Methods

to_dict() -> dict returns these fields with outcome converted to its string value.

ocura_oss.ParameterDelta

ParameterDelta(
    inherited: Mapping[str, str],
    added: Mapping[str, str],
    changed: Mapping[str, Mapping[str, str]],
)

Declared parameter relationship between source and child values.

Attributes

Removed source keys are outside the version 0.2.2 representation.

Methods

ocura_oss.ChildComparison

ChildComparison(
    pathway_id: str,
    reason: str,
    source_chokepoint_id: str,
    parameters: ParameterDelta,
    source_run: RunSummary,
    child_run: RunSummary | None,
    run_parameters: ParameterDelta | None,
    missing_evidence: bool,
)

Comparison of one child pathway with its source run.

Attributes

Methods

to_dict() -> dict returns a nested JSON-compatible representation.

ocura_oss.ComparisonResult

ComparisonResult(
    state: ComparisonState,
    source_chokepoint_id: str,
    source_pathway_id: str,
    source_run: RunSummary,
    children: tuple[ChildComparison, ...],
)

Comparison state and every child associated with one source chokepoint.

Attributes

Methods

to_dict() -> dict returns a nested JSON-compatible representation with enum values converted to strings.

ocura_oss.StateVerification

StateVerification(
    pathways: int,
    atoms: int,
    chokepoints: int,
    logs_checked: int,
    problems: tuple[tuple[str, str], ...],
)

Result of verifying records and referenced logs under one state directory.

Attributes

Properties

ok -> bool is True when problems is empty.

ocura_oss.DemoStep

DemoStep(
    step: str,
    detail: str,
)

One completed stage in a demonstration report.

Attributes

ocura_oss.DemoReport

DemoReport(
    root: pathlib.Path,
    steps: tuple[DemoStep, ...],
    comparison: ComparisonResult,
)

Retained demonstration location, completed stages, and comparison.

Attributes

Methods

to_dict() -> dict returns a JSON-compatible representation. root is converted to a string.

Enumerations

ocura_oss.Outcome

String enum describing one terminal command attempt.

Outcome derives from enum.StrEnum, so members also behave as strings.

ocura_oss.ComparisonState

String enum describing evidence completeness for one comparison.

ComparisonState derives from enum.StrEnum.

Exceptions

ocura_oss.StoreError

StoreError(message)

Raised for invalid workflow input and missing, malformed, checksum-mismatched, semantically inconsistent, unverifiable, or unwritable state.

run() returns normal command failures and launch failures as Outcome values. It raises StoreError when the workflow or evidence persistence fails.

ocura_oss.DemoError

DemoError(
    step: str,
    message: str,
)

Raised when a retained demonstration cannot complete.

Parameters

Attributes

The destination is retained when it was created before the failure.

Module metadata

ocura_oss.__version__

ocura_oss.__version__: str

Installed package version. Version 0.2.2 reports "0.2.2".

Typing

The wheel includes py.typed, so package annotations are visible to compatible static type checkers.

from pathlib import Path

from ocura_oss import Initialization, Store, initialize

initial: Initialization = initialize(Path("experiment"))
store: Store = Store(initial.root)

Dataclass fields use concrete record and result types. Parameter inputs use collections.abc.Sequence and Mapping so callers can supply compatible typed containers.

State and concurrency

The supported programmatic workflows write project-local .ocura-oss/ state. Legacy .ocura/ records are not compatible.

One mutating process per state root is supported at a time. Public workflow functions perform record mutation; direct record writers remain internal.

Raw atoms retain command arguments and logs retain command output. Keep secrets out of command arguments, parameter mappings, branch reasons, and output.

For command syntax and exit status, see the CLI reference. For the record layout and execution boundary, see the documentation overview.