Python API Reference

NovaFabric is a CLI-first toolkit, but every capability the nova command exposes is also reachable as a typed Python API. This reference documents the public symbols for programmatic capture, replay, diff, lineage, registry, spec validation, reporting, and signing — so you can embed NovaFabric in a test harness, a CI script, an internal tool, or a notebook without shelling out.

What you will learn

Everything below maps one-to-one onto a nova subcommand; where a CLI equivalent exists it is noted so you can cross-check behaviour with nova <command> --help.

Stability. All public symbols listed here are importable today and covered by tests. Most v0.x surfaces carry experimental maturity: they work, but interfaces may change before the v1.0 schema freeze. The Run Capsule and Evidence Bundle on-disk formats are not yet frozen — expect additive, backward-compatible schema changes until v1.0.

Table of contents

Section Import root CLI equivalent
SDK decorator novafabric.sdk.agent — (in-process)
Capture novafabric.capture.orchestrator nova capture
Extended event recording novafabric.capture.record — (in-process)
Replay novafabric.replay nova replay
Diff novafabric.diff nova diff
Lineage novafabric.lineage nova lineage
Asset Registry novafabric.registry.service nova register / nova list / nova promote
Spec validation novafabric.spec nova validate
Report generation novafabric.report.generator nova report
NovaSeal signing novafabric.trust.novaseal nova seal / nova verify
Score submission novafabric.scores nova score submit
REST client novafabric.client — (server mode)
Utility novafabric.capture —

SDK decorator

from novafabric.sdk.agent import agent

@agent(name, version, capsule_dir=None, deployment_environment=None, variant=None, session_id=None, session_sequence=None)

Wraps an agent function with capture hooks. LLM calls made inside the function are recorded into a capsule. This is the in-process alternative to nova capture — use it when you cannot (or do not want to) wrap the whole process, e.g. a single function inside a larger long-lived service.

Parameters

Parameter Type Description
name str Asset name (matches a registered asset, if any)
version str Asset version (semver)
capsule_dir Path | str | None Capsule store, like nova capture --output-dir: each call writes its own capsule to <capsule_dir>/<run_id>/. If None, OTel spans only — no capsule is written.
deployment_environment str | None Experimental (ADR-0126). Delivery-lifecycle tag for the run (e.g. production, staging), recorded verbatim as the additive optional deployment_environment manifest field with environment_source: sdk-arg. The nova capture --environment flag and the NOVAFABRIC_ENVIRONMENT env var take precedence.
variant Mapping[str, Any] | None Experimental (ADR-0116, record-only). Which A/B experiment/variant an external allocator had active: a mapping with experiment_id, variant_id, assignment_source (plus optional variant_label, assigned_at, extensions), copied verbatim into the capsule's optional variant block. NovaFabric never assigns variants; an incomplete explicit block raises before the workload runs. NOVAFABRIC_VARIANT* env vars take precedence.
session_id str | None Experimental (ADR-0122, record-only). ULID of the session this run is one turn of (create with nova session new); recorded as an additive optional back-reference — the session.json manifest stays authoritative. NOVAFABRIC_SESSION_ID takes precedence.
session_sequence int | None Experimental (ADR-0122). The run's turn number within the session; requires session_id. NOVAFABRIC_SESSION_SEQUENCE takes precedence.

All four tagging parameters are optional and additive: leave them unset and the capsule is byte-identical to previous releases. Invalid explicit values raise before the wrapped function runs; ambient env-var values that are invalid warn and are ignored rather than blocking the workload.

Returns — the decorated function's return value unchanged.

Example

from novafabric.sdk.agent import agent
import openai

client = openai.OpenAI()

@agent(name="summariser", version="0.2.0", capsule_dir="capsules/")
def summarise(text: str) -> str:
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": f"Summarise: {text}"}],
    )
    return response.choices[0].message.content

result = summarise("Long document text here...")

Every call to summarise() writes a new capsule under capsules/<run_id>/ (a fresh ULID per call, logged as sdk_capsule_started). Releases up to v0.106.0 wrote every call into capsule_dir itself, so a second call overwrote capsule.yaml while the JSONL streams appended: one capsule, several runs.

Behaviour

The capsule this produces has the same shape as one produced by nova capture — you can nova validate, nova diff, and nova replay --mode forensic|semantic it the same way. It records capture_mode: sdk-decorator and an @agent:<name>@<version> label instead of a command, so nova replay --mode mocked refuses it (CapsuleNotReplayable): there is no command to re-run.

Capture

from novafabric.capture.orchestrator import CaptureOrchestrator, CaptureResult

CaptureOrchestrator is the programmatic equivalent of nova capture <cmd>. It captures a subprocess as a Run Capsule with no changes to the target program — hooks are injected via sitecustomize.py over PYTHONPATH. A capsule is written on both success and failure (a failed run sets status: failure and records an error block).

CaptureOrchestrator(base_dir=None, runner=None, ...)

Parameter Type Default
base_dir Path | None ./.novafabric/runs/ (relative to the working directory)
runner RunnerSpec | None None — run locally; a RunnerSpec dispatches to docker/kubernetes/slurm/lsf/pbs like nova capture --runner
pii_gate Any | None None
legal_hold bool False — only acts with a pii_gate: the gate scans the manifest and, when it redacts anything, also writes the redacted copy to <base_dir>/../legal_hold/<run_id>/; a scan error exits 3. With a gate but no legal_hold, the scan is skipped. There is no nova capture flag for it
mark_provenance bool False
fast_emit bool False
emit_spool bool False
spool_dir Path | None None
masking_pipeline Any | None None
capture_media bool False

Only base_dir and runner are covered by examples here; the other keyword arguments mirror nova capture options — see nova capture --help and the constructor's docstring.

CaptureOrchestrator.run(command) → CaptureResult

Captures a command execution as a capsule.

Parameter Type Description
command list[str] Command + arguments 
from pathlib import Path
from novafabric.capture.orchestrator import CaptureOrchestrator

result = CaptureOrchestrator(base_dir=Path(".novafabric/runs")).run(
    ["python", "agent.py", "--dataset", "data.csv"]
)
print(result.run_id)      # ULID string
print(result.capsule_dir) # Path to capsule directory
print(result.exit_code)   # int

The examples on this page pass capsules written this way, so their paths start with .novafabric/runs/. That is the SDK default, not the CLI's: nova capture writes to the capsule store, ~/.novafabric/capsules/<run_id>/ by default (novafabric._paths.default_capsule_dir(), which honours NOVAFABRIC_CAPSULE_DIR and NOVAFABRIC_HOME). For a capsule from nova capture, pass that directory instead.

CaptureResult

@dataclass
class CaptureResult:
    run_id: str
    capsule_dir: Path
    exit_code: int
    runner_status: str | None = None   # e.g. "failed_setup": the workload never started
    runner_error: str | None = None    # the runner's own reason, when it gave one

exit_code alone cannot tell the workload ran and failed from the workload never started (a missing command and a program that exits 127 both give 127); only runner_status == "failed_setup" identifies the latter. None means the runner did not say, not "completed".

Extended event recording (experimental)

from novafabric.capture import record

Status: experimental (ADR-0209 P1; the API surface may change until it has survived one release unchanged). The stable public façade for emitting the extended capture events of ADR-0082 — guardrails, evaluators, state transitions, memory operations, rerankers, vector retrievals, file events — into whatever run is currently being captured (any capture mode: nova capture, adapters, @agent, in-process hooks).

Contract:

record.active() -> bool

record.file_event(operation, path, *, size_bytes=None, success=True,
                  error=None, agent_id=None)
record.state_transition(step_index, state_digest_before, state_digest_after, *,
                        agent_id=None, state_before=None, state_after=None)
record.memory_operation(operation, memory_key, *, relevance_score=None,
                        freshness_seconds=None, agent_id=None, value=None,
                        origin_run_id=None, origin_memory_key=None,
                        origin_timestamp_utc=None)
record.guardrail(guardrail_name, outcome, *, category=None, score=None,
                 agent_id=None, details=None)
record.evaluator(evaluator_name, *, score=None, label=None, passed=None,
                 dataset_id=None, agent_id=None, rationale=None)
record.reranker(reranker_model, *, input_count=None, output_count=None,
                documents=None, agent_id=None)
record.vector_retrieval(vector_store, *, phase="completed", operation="query",
                        collection=None, top_k=None, returned_count=None,
                        duration_ms=None, error=None, documents=None,
                        agent_id=None)

network_event and human_approval are deliberately not exposed: the wire-level hooks and nova seal propose own those streams, and a second producer would invite double-recording.

record.wrap_retriever(fn, *, vector_store, collection=None, operation="query")

Wraps any sync callable retriever; emits VectorRetrievalStarted before the call, …Completed (with returned_count, duration_ms) on return, …Failed (with error) on exception — then re-raises. Document payloads are recorded only at forensic/air_gapped capture level.

# LangChain retriever
search = record.wrap_retriever(retriever.invoke,
                               vector_store="qdrant", collection="docs")
docs = search("what changed in v0.63?")

# DSPy
retrieve = dspy.Retrieve(k=8)
retrieve.forward = record.wrap_retriever(retrieve.forward,
                                         vector_store="chroma")

An async variant is future design. Default-path wirings that call this API automatically (OpenAI Agents guardrail spans, LangGraph state transitions) are documented in the capture tutorial; file/memory/evaluator/reranker events have no auto-capture — they record only when you call the façade.

record.tool(fn=None, *, name=None, mutation_class="unknown", version=None, ignore=())

Status: experimental (ADR-0306 slice 1, since v0.105.0). Declares a Python function as a tool boundary that nova capture records and mocked nova replay serves, so the tool does not run again during replay.

from novafabric.capture import record

@record.tool(mutation_class="external-side-effect", ignore=("ctx",))
def send_invoice(ctx, customer_id: str, amount_cents: int) -> dict: ...

@record.tool  # name = fn.__name__, mutation_class = "unknown"
async def lookup_order(order_id: str) -> dict: ...

Replay

from novafabric.replay import ReplayEngine, ReplayFlags
from novafabric.replay._result import ReplayResult

Replay re-executes or inspects a capsule with external calls controlled. A replay writes a replay_result.yaml report, not a capsule — only intervention mode also writes a minimal counterfactual Run Capsule (capsule.yaml marked replay_mode: intervention plus the model/tool call streams) that you can nova diff against its source.

ReplayFlags

Controls replay mode and safety settings.

@dataclass
class ReplayFlags:
    mode: Literal["mocked", "forensic", "semantic", "exact", "intervention"] = "mocked"
    dry_run: bool = False
    allow_readonly: bool = False
    allow_mutating: bool = False
    allow_external_side_effects: bool = False
    allow_unknown_mutation: bool = False
    output_dir: Path | None = None
    intervention_file: Path | None = None
    required_environment: str | None = None
    permissive: bool = False   # mocked mode only (ADR-0300): warn instead of fail

Modes

Mode What it does Typical use
forensic Read-only inspection. No subprocess, no network. Audit / post-incident
mocked Re-spawns the command (Python workloads). Serves recorded responses for OpenAI chat.completions.create and responses.create / Anthropic messages.create — sync or async, streamed or not (ADR-0304) — and recorded MCP ClientSession.call_tool results, one per call, plus — experimental — results of functions declared with record.tool and of Google ADK tools run with novafabric.adapters.google_adk.make_tool_plugin() and OpenAI Agents function tools passed through novafabric.adapters.openai_agents.wrap_tools() (both shipped in v0.106.0; the latter relies on an undocumented SDK seam). Other tools run live (HTTP, shell, files, other framework-native tools, undeclared functions); replay_contract.network_connections_live reports their connections. Fails closed (ADR-0300): an extra or unmatched call, an unsupported model surface (parse, legacy completions, with_raw_response), or an unconsumed recording makes the result failure with a divergence_reason; permissive=True only reports them. CI / regression
semantic Does not re-execute. Scores how similar the capsule's recorded model responses are to each other (mean pairwise text similarity, 0.0–1.0); no live model is called; returns similarity_score. Consistency check of recorded responses
exact Does not re-execute. Eligibility check for byte-exact replay (exact_eligible + reasons): deterministic env.lock, per-call seed, no tool-schema drift. Local / on-prem / compliance
intervention Re-executes with a spec-driven intervention overlay (experimental, ADR-0086). What-if / counterfactual analysis

NovaFabric explicitly does not claim byte-exact replay of remote LLM calls. exact mode reports eligibility, not a guarantee.

The allow_* flags form a safety ladder over the capsule's recorded tool calls (allow_readonly < allow_mutating < allow_external_side_effects < allow_unknown_mutation). They drive the dry_run report, and allow_mutating triggers an audited policy gate before the replay starts. Inside the replayed process they gate only the intercepted tool surfaces (ADR-0306, experimental): with permissive=True an unmatched MCP, record.tool, ADK or Agents SDK tool call runs live only if they permit its class (MCP calls count as unknown), and a replay.yaml allow: true override re-executes a tool only if they permit its class. A strict replay whose replay.yaml has an allow: false override on a tool it cannot intercept returns status: "aborted" with error["type"] == "ToolOverrideUnenforceable". Leaving them False does not sandbox tools that mocked mode does not intercept — run such replays in a sandbox or against test credentials.

ReplayEngine(capsule_dir, flags, base_dir=None, actor="cli")

Parameter Type
capsule_dir Path
flags ReplayFlags
actor str — the identity the replay_mutating policy gate evaluates (subject.user, consulted when a write-permitting flag is set) and the audit log records; default "cli"
base_dir Path | None — defaults to $NOVAFABRIC_HOME/replays/ (_paths.default_replays_dir(); ./.novafabric/replays/ under the current directory up to v0.106.0)

ReplayEngine.run() → ReplayResult

Replays the capsule according to the flags. Writes replay_result.yaml in the output directory.

from pathlib import Path
from novafabric.replay import ReplayEngine, ReplayFlags

flags = ReplayFlags(mode="forensic")
engine = ReplayEngine(
    capsule_dir=Path(".novafabric/runs/01HXAY7M5JZ8R7K4P9DPBYK2WX"),
    flags=flags,
)
result = engine.run()
print(result.status)              # "success" | "failure" | "aborted" | "dry_run"
print(result.model_calls_mocked)  # int
print(result.env_warnings)        # list of {field, original, current}

Example — a mocked replay gate in CI (fail-closed, ADR-0300)

result = ReplayEngine(capsule_dir=cap, flags=ReplayFlags(mode="mocked")).run()
if result.status != "success":
    # e.g. "model_queue_exhausted: no recorded openai response left for call #3 ..."
    raise SystemExit(result.divergence_reason or result.error)
print(result.model_calls_mocked, "of", result.model_calls_available, "model calls served")
print(result.tool_calls_mocked, "MCP tool results served;",
      result.replay_contract["tool_calls_not_interceptable"], "recorded tools not intercepted")

The replayed process raises a ReplayDivergenceError subclass (novafabric.replay._errors: ReplayQueueExhaustedError, ReplayProviderMismatchError, ReplayOrderMismatchError, ReplayUnsupportedSurfaceError, ReplayRecordMalformedError, ReplayToolUnmatchedError) at the call that diverged.

Example — semantic replay against a drifting remote LLM

flags = ReplayFlags(mode="semantic")
result = ReplayEngine(capsule_dir=cap, flags=flags).run()
if result.similarity_score is not None and result.similarity_score < 0.9:
    print(f"semantic drift detected: {result.similarity_score:.3f}")

Example — exact-replay eligibility gate for a compliance run

flags = ReplayFlags(mode="exact")
result = ReplayEngine(capsule_dir=cap, flags=flags).run()
if not result.exact_eligible:
    print("not byte-exact reproducible:", result.exact_reasons)

ReplayResult

@dataclass
class ReplayResult:
    replay_id: str
    replay_of_run_id: str
    mode: str
    status: str
    start_time: str
    end_time: str
    duration_ms: int
    policy_flags_used: list[str]
    env_warnings: list[dict[str, str]]
    model_calls_mocked: int = 0          # mocked: responses actually served
    tool_calls_mocked: int = 0           # mocked: MCP results actually served
    tool_calls_available: int | None = None  # recorded on a servable surface (MCP)
    tool_calls_recorded: int | None = None   # every recorded tool call
    # mocked mode, ADR-0300 (all optional)
    model_calls_available: int | None = None
    model_calls_unmatched: int | None = None
    tool_calls_live: int | None = None
    tool_calls_unmatched: int | None = None
    queues_fully_consumed: bool | None = None
    divergence_reason: str | None = None
    replay_contract: dict | None = None  # policy, surfaces, divergences, ...
    exit_code: int | None = None
    error: dict | None = None
    # semantic mode
    similarity_score: float | None = None   # 0.0–1.0; set when mode="semantic"
    matched_run_id: str | None = None
    # intervention mode (experimental, ADR-0086)
    intervention: dict | None = None
    # exact mode
    exact_eligible: bool | None = None      # set when mode="exact"
    exact_hash_count: int | None = None
    exact_reasons: list[str] | None = None
    # tool-call schema drift findings (ADR-0128; additive, optional)
    schema_drift: list[dict] | None = None

    def as_dict(self) -> dict: ...

Diff

from novafabric.diff import DiffEngine, DiffReport

DiffEngine structurally compares two capsules — aligning their model calls, tool calls, environment, and outputs — and reports what changed, was added, or was removed. This is the engine behind nova diff --assert-no-regressions.

DiffEngine.compare(capsule_a, capsule_b) → DiffReport

from pathlib import Path
from novafabric.diff import DiffEngine

report = DiffEngine().compare(
    capsule_a=Path(".novafabric/runs/01HX.../"),
    capsule_b=Path(".novafabric/runs/01HY.../"),
)

print(report.changed_count)  # int
print(report.added_count)    # int
print(report.removed_count)  # int
print(report.has_changes)    # bool: any changed, added or removed entry
print(report.is_complete)    # bool: False if any record line was skipped as malformed
print(report.as_dict())      # serialisable dict

Example — use the diff as a CI regression gate

import sys
report = DiffEngine().compare(baseline_capsule, candidate_capsule)
if not report.is_complete:
    print(report.skipped_malformed_lines)
    sys.exit(2)   # cannot compare: a record line was unreadable
if report.has_changes:
    print(report.as_dict())
    sys.exit(1)   # fail the build on any structural change

has_changes is the one definition of "any difference". The nova diff --assert-no-regressions CLI exits 1 on it, and the text and GitHub-annotation formatters read it too, and as_dict() carries it as a top-level has_changes key — reach for the CLI directly in shell-based pipelines. Gating on changed_count alone misses a call that was added or removed.

has_changes is computed over the records that parsed. A non-blank line of model-calls.jsonl or tool-calls.jsonl that is not UTF-8, not JSON, or not a JSON object is skipped and counted in skipped_malformed_lines (per side a/b, per file model_calls/tool_calls; also a top-level key of as_dict()), and is_complete turns False. The CLI gate then exits 2, "cannot compare", before it looks at has_changes (ADR-0303 Amendment 1).

DiffReport

@dataclass
class DiffReport:
    run_a_id: str
    run_b_id: str
    env_changes: list[dict]
    model_call_pairs: list[dict]
    tool_call_pairs: list[dict]
    output_changes: list[dict]
    skipped_malformed_lines: dict[str, dict[str, int]]  # {"a"|"b": {"model_calls", "tool_calls"}}

    @property
    def changed_count(self) -> int: ...
    @property
    def added_count(self) -> int: ...
    @property
    def removed_count(self) -> int: ...
    @property
    def has_changes(self) -> bool: ...
    @property
    def malformed_line_count(self) -> int: ...
    @property
    def is_complete(self) -> bool: ...
    def as_dict(self) -> dict: ...
    def write(self, output_path: Path) -> None: ...

Format helpers

from novafabric.diff._format import format_text, format_json, format_github_annotations

text = format_text(report)                    # human-readable diff
json_str = format_json(report)                # machine-readable
annotations = format_github_annotations(report)  # GitHub Actions annotations

Lineage

from novafabric.lineage import (
    LineageWriter,
    LineageStore,
    import_capsule_dir,
    index_capsule_lineage,
    ImportResult,
)

Lineage is a directed provenance graph in SQLite — a rebuildable cache derived from each capsule's lineage.jsonl. Use LineageWriter to infer edges, index_capsule_lineage (one capsule) or import_capsule_dir (a directory of capsules) to index them, and LineageStore to query the graph.

LineageWriter(capsule_dir, run_id, version=None)

Infers lineage edges from a capsule and writes lineage.jsonl. Each edge's emitter.version is version, defaulting to the installed novafabric.__version__. produced_by edges are read from the manifest's outputs — Artifact objects (keyed by their path, with name/content_hash/media_type carried into artifact_ref; an inline Artifact is keyed inline:<name>) or legacy path strings — and from produced.jsonl.

from pathlib import Path
from novafabric.lineage import LineageWriter

writer = LineageWriter(
    capsule_dir=Path(".novafabric/runs/01HX.../"),
    run_id="01HXAY7M5JZ8R7K4P9DPBYK2WX",
)
edges = writer.infer()      # list[LineageEdge]
path = writer.write(edges)  # writes lineage.jsonl, returns Path

LineageStore(db_path=None)

SQLite-backed graph store. Default: $NOVAFABRIC_HOME/registry.db (~/.novafabric/registry.db unless NOVAFABRIC_HOME is set; NOVAFABRIC_DB_PATH overrides the file).

from novafabric.lineage import LineageStore

store = LineageStore()

# What did this run depend on? (ancestors)
# A run node's ref is the bare run id; pass kind="run" to disambiguate.
ancestors = store.provenance(
    ref="01HXAY7M5JZ8R7K4P9DPBYK2WX",
    kind="run",
    depth=3,
)

# What runs consume this asset? (descendants / impact)
# An asset node's ref is "<registry>:<name>@<version>" (registry "local" by default).
dependents = store.blast_radius(
    ref="local:my-dataset@1.0.0",
    kind="asset",
    depth=5,
)

# Replay ancestry
chain = store.replay_chain(run_id="01HXREPLAY...")

Each result item is a dict: {"kind": "run|asset|artifact", "ref": "..."}. A time_travel query is also available for point-in-time graph reconstruction.

index_capsule_lineage(capsule_dir, db_path=None, *, run_id=None) → ImportResult

Index one capsule's lineage.jsonl into the graph store. The run is keyed by run_id when given, else by the directory name (a nova capture capsule directory is named after its run id). A capsule without lineage.jsonl returns skipped=True; a lineage.jsonl that is present but unreadable (a symlink, FIFO or directory) raises instead of reporting zero edges.

from pathlib import Path
from novafabric.lineage import index_capsule_lineage

result = index_capsule_lineage(Path("~/.novafabric/capsules/01HX...").expanduser())
print(result.edges_indexed)  # int
print(result.nodes_indexed)  # int

import_capsule_dir(runs_dir, db_path=None) → list[ImportResult]

Index every capsule directly under runs_dir (each subdirectory that holds a capsule.yaml), returning one ImportResult per capsule.

from pathlib import Path
from novafabric.lineage import import_capsule_dir

for result in import_capsule_dir(Path("~/.novafabric/capsules").expanduser()):
    print(result.capsule_run_id, result.edges_indexed, result.skipped)

ImportResult

@dataclass
class ImportResult:
    capsule_run_id: str
    edges_indexed: int
    nodes_indexed: int
    skipped: bool
    warning: str | None = None

The lineage store also emits OpenLineage 2.0.2 START/COMPLETE/FAIL events to a configured backend (Marquez, Atlan, OpenMetadata). See the CLI reference for nova lineage emission configuration.

Graph analytics (experimental — ADR-0212..0215)

from novafabric.lineage._store import LineageStore
from novafabric.lineage.analytics import (
    build_insights_report,            # ADR-0215 synthesized report
    compute_graph_metrics,            # ADR-0212 hubs / articulation points
    rank_root_causes,                 # ADR-0213 upstream suspect ranking
    to_cypher, to_gexf, to_graphml,   # ADR-0214 exports
)

store = LineageStore()
metrics = compute_graph_metrics(store, top_n=10)
report = build_insights_report(store)
graphml = to_graphml(store.all_nodes(), store.all_edges())

Read-only and deterministic; oversize graphs raise LineageGraphTooLargeError instead of truncating. See the CLI reference for the command equivalents (nova lineage metrics|root-cause|export-graph, nova insights).

Asset Registry

from novafabric.registry.service import (
    register_asset,
    list_assets,
    get_asset,
    promote_asset,
    promote_asset_direct,
)
from novafabric.spec.models import AssetStatus

The registry is a local SQLite store ($NOVAFABRIC_HOME/registry.db, overridable with NOVAFABRIC_DB_PATH) of versioned assets addressed as name@version. Promotion is governance metadata only — it updates one DB record and does not deploy, restart, or redeploy anything.

register_asset(spec, spec_path, db_path=None, *, actor="") → dict

from pathlib import Path
from novafabric.spec.validator import validate_spec
from novafabric.registry.service import register_asset

spec = validate_spec(Path("my-model.yaml"))
asset = register_asset(spec, spec_path=Path("my-model.yaml"))
print(asset["id"])   # UUID

actor is the identity recorded for the registration in the audit trail.

Raises DuplicateAssetError if name@version already exists.

list_assets(asset_type, status, db_path=None) → list[dict]

asset_type and status are required positional-or-keyword arguments with no default; pass None to leave a filter off.

from novafabric.registry.service import list_assets

agents = list_assets(asset_type="agent", status="production")
for a in agents:
    print(a["name"], a["version"], a["status"])

get_asset(name, version, db_path=None) → dict

from novafabric.registry.service import get_asset

asset = get_asset("my-model", "1.0.0")
# version is required; pass None (or "latest") for the latest registered version

Raises AssetNotFoundError if not found.

promote_asset(name, version, to_status, actor, force=False, db_path=None, significance_gate=False, ...) → dict

Advances an asset through the six-state lifecycle. The to_status argument is an AssetStatus enum member. The remaining keyword arguments (sig_p0, sig_p1, sig_alpha, sig_beta, sig_scores_path, sig_scores_metric) tune the opt-in significance gate (ADR-0080).

Use promote_asset_direct for a single-actor promotion. promote_asset is the ungated core: it runs the lifecycle, promote policy, eval and significance checks, but not the protected-stage gate. promote_asset_direct(name, version, to_status, actor, force=False, db_path=None, *, policy_subject=None, **gate_kwargs) is the one gate every single-actor route uses (nova promote direct, nova prompt promote, the nova serve promote endpoints, PUT /v0/assets/{id}/promote): a target in PROTECTED_STAGES (today production) needs an explicit allow from the promote_direct policy action before promote_asset runs — force does not bypass it and a no-OPA NoopEngine decision is not an allow (ADR-0312 D1). It raises DirectPromotionDeniedError (a PromotionBlockedError subclass) when refused.

from novafabric.registry.service import promote_asset
from novafabric.spec.models import AssetStatus

asset = promote_asset(
    name="my-model",
    version="1.0.0",
    to_status=AssetStatus.staging,
    actor="ci-bot",
)

Raises:

Lifecycle: development → validated → pending_approval → staging → production → archived. Promotion of agent assets is eval-gated; the OPA/Rego policy engine is consulted before the eval-score gate so decisions are always logged. See nova promote --help for the maker-checker CLI flow.

Spec validation

from novafabric.spec.validator import validate_spec, SpecValidationError
from novafabric.spec.models import AssetType, AssetStatus, BaseAssetSpec

validate_spec(yaml_path) → BaseAssetSpec

from pathlib import Path
from novafabric.spec.validator import validate_spec

spec = validate_spec(Path("my-agent.yaml"))
print(spec.name, spec.version, spec.asset_type)

Raises SpecValidationError with a Pydantic error list and a hint string on validation failure.

Asset type models

from novafabric.spec.models import (
    AssetType,    # Enum: model agent prompt tool dataset evaluation deployment
    AssetStatus,  # Enum: development validated pending_approval staging production archived
    BaseAssetSpec,
    ModelSpec,
    AgentSpec,
    PromptSpec,
    ToolSpec,
    DatasetSpec,
    EvaluationSpec,
    DeploymentSpec,
)

AssetType has seven members (one per asset type). AssetStatus has the six lifecycle states listed above.

Report generation

from novafabric.report.generator import generate_report

markdown = generate_report(format_="markdown")
json_str = generate_report(format_="json")

generate_report(format_="markdown", db_path=None) renders the current registry state. This is the engine behind nova report.

NovaSeal signing

from novafabric.trust.novaseal import NovaSeal, KeyConfig

NovaSeal is the cryptographic signing core for capsule manifests: a DSSE envelope (ECDSA P-256 / SHA-256), an optional RFC 3161 trusted timestamp, and an entry appended to a SQLite-backed append-only Merkle log. The local key profile is the default; cloud-KMS profiles (aws_kms, azure_kv, gcp_kms) are also supported by the config loader.

Maturity — experimental. NovaSeal is present and covered by tests, but its interfaces (like the rest of the v0.x surface) may change before the v1.0 schema freeze. Timestamping is best-effort: if the TSA is unreachable, the capsule is still sealed, without a timestamp.

Sign a capsule manifest

from novafabric.trust.novaseal import NovaSeal, KeyConfig

# Instantiate with a local ECDSA P-256 key
config = KeyConfig(
    profile="local",
    key_path="~/.novafabric/seal.key",
    cert_path="~/.novafabric/seal.crt",
)
seal = NovaSeal(
    config=config,
    tsa_url="",  # "" = no RFC 3161 timestamp, no network call; or your TSA's URL
    db_path="~/.novafabric/novaseal-merkle.db",
)

# Sign a capsule manifest (dict)
bundle = seal.seal(capsule_manifest)
# bundle.dsse_envelope  → bytes  (DSSE JSON)
# bundle.tsr            → bytes  (RFC 3161 DER; b"" if TSA skipped/unavailable)
# bundle.log_entry      → dict   (leaf_index, leaf_hash, root_hash, tree_size,
#                                  root_version, inclusion_proof, entry)
#                         root_version 2 = RFC 6962 root (new logs); absent or 1 =
#                         legacy root, meaningful only with tree_size (ADR-0313)
# bundle.capsule_id     → str    (SHA-256 hex of the signed payload)

A leading ~ in key_path, cert_path and a file db_path is expanded to the home directory (MerkleLog and open_merkle_log expand it too). A Postgres DSN db_path is used exactly as given. Up to v0.106.0 NovaSeal(...) and MerkleLog(...) did not expand ~, so the example above created a literal ./~ directory in the working directory.

seal() accepts an optional intent (a SigningIntent, default SigningIntent.AUTHORED) recorded in the DSSE envelope to support FDA 21 CFR §11.50 signing-intent semantics. Pass intent=None to omit it.

Verify an existing .seal/ directory

result = seal.verify(
    capsule_id=bundle.capsule_id,
    seal_dir=".novafabric/runs/01HX.../.seal",
)
# result.valid            → bool
# result.signature_ok     → bool
# result.timestamp_ok     → bool
# result.log_integrity_ok → bool  (True only when inclusion was actually checked)
# result.log_inclusion    → "local-log" | "carried-proof" | "not-checked" | "failed"
# result.pae_encoding     → "dsse-v1" | "legacy-le64" (sealed through v0.102.x)
# result.ca_chain_ok      → bool
# result.errors           → list[str]
assert result.valid, str(result)

# Independent verification — no key, no config, no Merkle log needed:
from novafabric.trust.novaseal import verify_seal_dir

result = verify_seal_dir(".novafabric/runs/01HX.../.seal")  # merkle_log=None

log_inclusion is not-checked (and valid stays True) when the capsule carries no inclusion proof and no log holding its entry was supplied — a capsule sealed by v0.102.x or earlier, verified away from the sealer's log. Any inconsistency (an entry that names another capsule, a proof that does not recompute its root) is failed.

seal.verify() passes the sealer's own log, and verify_seal_dir(..., merkle_log=log) any log you open. The log is checked at the index log-entry.json records, so a manifest sealed twice (the same leaf twice in the log) verifies at each index. When the supplied log holds other seals by the same signer (the entry's keyid, or the SHA-256 fingerprint of the certificate of any signature that verifies) under the record's root version, it is that sealer's log: an entry missing from it is failed, even if the carried proof verifies. A log with no seals by that signer, a fresh log, or a log of the other root version (after starting a new v2 log, ADR-0313) is not the sealer's, and absence there keeps carried-proof, with a note in log_notes.

timestamp_ok is False for a token that cannot be cryptographically verified. A token the strict DER parser rejects is checked on a lenient-encoding path (timestamp_strict False) that still requires a verifying TSA signature, a signed messageDigest equal to the hash of the TSTInfo, and that TSTInfo's imprint equal to the envelope hash.

verify() reads manifest.dsse, an optional manifest.dsse.tsr, and log-entry.json from the .seal/ directory. An absent or empty TSR is reported as absent — timestamp_ok=None (ADR-0301; it was True before) — and does not invalidate the seal; a present but invalid TSR fails verification (timestamp_ok=False). result.identity_trust is "self-asserted" when the signature verifies (no CA bundle is consulted here) and "none" otherwise; result.signer_subject is the embedded certificate's subject, an unverified claim. result.ca_chain_ok is False unless a CA bundle validated the chain (nova verify --ca-bundle).

Rotate the signing key

from novafabric.trust.novaseal import KeyConfig

new_config = KeyConfig(profile="local", key_path="new.key", cert_path="new.crt")
receipt = seal.rotate_key(new_config)
# receipt.old_key_fingerprint, receipt.new_key_fingerprint, receipt.rotation_log_entry

Rotation appends a key_rotation event to the Merkle log and switches subsequent seals to the new key.

Load a signing profile from config

Configuration is read from NOVAFABRIC_SEAL_CONFIG when set (a missing file is an error), else from $NOVAFABRIC_HOME/novaseal.yaml (~/.novafabric/novaseal.yaml by default):

from novafabric.trust.novaseal.config import load_signing_profile
from novafabric.trust.novaseal import NovaSeal, KeyConfig

profile = load_signing_profile()  # None if NovaSeal is not configured
if profile:
    seal = NovaSeal(
        config=KeyConfig(
            profile=profile.profile,
            key_path=str(profile.key_path),
            cert_path=str(profile.cert_path),
        ),
        tsa_url=profile.tsa_url,
        db_path=str(profile.merkle_db),
    )

The dedicated, hardened NovaSeal signing service (network service, qualified timestamps, Sigstore-keyless mode) described in ADR-0041 is a planned capability. The novafabric.trust.novaseal library documented here is the in-process signing core.

Score submission

Maturity: experimental (ADR-0119). The documented, supported way for an external evaluation pipeline — a CI job, a third-party LLM-as-judge, a human tool — to submit an already-computed score into a capsule's append-only scores.jsonl. Works offline, no server, no model call.

novafabric.scores.submit(capsule_dir, *, name, value, value_type, evaluator_id, subject, ...) → SubmitResult

from novafabric.scores import submit

result = submit(
    "./capsules/run-01HX...",
    name="answer_correct", value=True, value_type="boolean",
    source="judge", evaluator_id="ci://acme/repo#judge@v3",
    subject="sha256:...",            # must exist in the target capsule
    eval_card_digest="sha256:...",   # required — the eval card reproducibility key
    score_id=None,                   # supply a client-minted ULID for idempotency
    supersedes=None,                 # prior score_id being corrected (append-only)
)
result.score.score_id     # the appended (or replayed) evidence-grade Score
result.idempotent_replay  # True → score_id already existed; no second line appended
result.config_bound       # True → an ADR-0117 ScoreConfig validated the value

Remaining keyword-only parameters: source ("code" default; human / heuristic / judge), subject_kind ("span" default), run_id, significance (an ADR-0080 SignificanceBlock), and db_path (registry override for score-config lookup).

Validation is fail-closed — on any rejection nothing is written, and a named exception is raised (SubmissionInvalidError, CapsuleNotFoundError, SubjectNotFoundError, SupersedesNotFoundError, IdempotencyConflictError, ScoreConfigViolation — all importable from novafabric.scores). Corrections never edit a prior record: submit a new score whose supersedes names the prior score_id; both lines stay in the log. CLI equivalent: nova score submit.

REST client

Maturity: experimental (ADR-0202 P1, shipped 2026-07-24). A typed, sync httpx client for the multi-tenant nova server REST API (/v0, api/openapi.yaml). This is server-mode tooling: every local-first feature above works without it, and importing it performs no I/O — the client sends no request other than the ones you invoke (no telemetry, no update checks). It mirrors the TypeScript SDK's contract (ADR-0194) and adds one operation TS does not have yet: capsule upload.

novafabric.client.NovaFabricClient(base_url, *, api_key=None, token=None, timeout=None, retries=None, transport=None)

from pathlib import Path

from novafabric.capture import new_ulid
from novafabric.client import NovaFabricClient

with NovaFabricClient(
    "https://nova.example.com/v0",     # no default URL — include the /v0 prefix
    api_key="nvfk_...",                # or token="..." / token=lambda: fresh_jwt()
) as nc:
    print(nc.health().data.version)    # server version (skew diagnostics)

    # Capsules: upload a directory (packed to a deterministic ZIP), a .zip
    # path, or raw ZIP bytes; then read back.
    nc.upload_capsule(Path("./capsules/run-01HX..."))
    detail = nc.get_capsule("run-01HX...").data

    # Cursor pagination: one page, or a lazy iterator over all pages.
    page = nc.list_capsules(limit=100).data          # .items / .next_cursor / .total
    for capsule in nc.iter_capsules():               # one HTTP call per page
        print(capsule.run_id, capsule.status)

    # Assets mirror the same surface: get_asset / list_assets / iter_assets.

    # External scores (ADR-0119): 201 → stored record; a 200 idempotent
    # replay carries no body, so .data is None — inspect .meta.status.
    result = nc.submit_score("run-01HX...", {
        "name": "answer_correct", "value": True, "value_type": "boolean",
        "source": "judge", "evaluator_id": "ci://acme/repo#judge@v3",
        "subject": "sha256:...", "eval_card_digest": "sha256:...",
        "score_id": new_ulid(),        # client-minted ULID → safe to re-send
    })

Contract points (normative spec: design/spec/python-client-v0.md (private)):

An async twin (AsyncNovaFabricClient) and artifact streaming are planned (ADR-0202 P2); SSE and an OTLP endpoint helper are future design (P3).

Compliance-export provenance (experimental — ADR-0197)

Field-group compliance exporters tag each field-group with an evidence_source marker so a consumer can tell how NovaFabric established each value. The shared primitive lives in novafabric.compliance.export.provenance:

from novafabric.compliance.export.provenance import (
    EvidenceSource,      # operator_asserted | capsule_verified | unverifiable
    EvidenceSourceRef,   # re-performable ref: capsule_id + content_digest (+ seal path)
    build_capsule_ref,   # hash a capsule dir into an EvidenceSourceRef
    mark,                # validate a (source, ref) pairing
    validate_marked,     # assert a set of field-groups is honestly marked
)

# capsule_verified MUST carry a re-performable reference (ADR-0197 I-3):
ref = build_capsule_ref(capsule_dir, verified_at="2026-07-24T00:00:00Z")
source, ref = mark(EvidenceSource.capsule_verified, ref=ref)

# Enforce on an export path — raises UnmarkedFieldGroupError / MissingReperformableRefError:
from novafabric.compliance.export import AnnexIVExporter
doc = AnnexIVExporter().build_annex_iv_document("dep-1", capsule_dir)
validate_marked(doc.completeness_summary())

operator_asserted and unverifiable must not carry a ref (a reference would misrepresent them as a re-performed verification). The marker is additive and optional on the wire — a pre-ADR-0197 document deserializes with evidence_source is None. See design/spec/evidence-source-provenance-marker.md (private).

Utility

from novafabric.capture import new_ulid, new_span_id

run_id = new_ulid()      # str — ULID (26 chars, time-sortable)
span_id = new_span_id()  # str — 16-char hex

Summary and next steps

You want to… Use CLI equivalent
Capture one function in-process @agent decorator —
Capture an arbitrary command CaptureOrchestrator.run nova capture
Re-run or inspect a capsule ReplayEngine + ReplayFlags nova replay
Compare two runs / gate CI DiffEngine.compare nova diff --assert-no-regressions
Trace provenance / blast radius LineageStore nova lineage
Register / promote assets registry.service nova register / nova promote
Validate a spec validate_spec nova validate
Sign / verify a capsule NovaSeal nova seal / nova verify
Talk to a NovaFabric server from Python NovaFabricClient (experimental) —

Where to go next