Audit and capabilities#

Reference for the fg.audit namespace (explain, conflicts, diff_proof_frames) and the fg.meta namespace (capabilities). Both namespaces are read-only: assigning an attribute on either raises FrozenSnapshotError.

fg.audit#

The fg.audit namespace exposes post-hoc audit queries over the ledger and over recorded round events. The attribute fg.audit returns the audit namespace manager.

fg.audit  # read-only namespace; fg.audit.x = ... raises FrozenSnapshotError

fg.audit.explain(target)#

Explain chosen-policy state for the predicate/entity cell of one assertion.

Parameters:

• target (str | AssertionRecord): an assertion id string, or any object exposing a non-empty asrt_id attribute (for example an AssertionRecord from fg.assertions.by_id).

Returns: dict[str, Any] with the following keys:

• pred_id (str): the predicate id of the resolved claim.
• e_ref (str): the entity reference of the resolved claim.
• active_claims (list[dict[str, Any]]): one entry per active claim on (pred_id, e_ref) whose argument tail matches the target’s value atoms. Each entry has asrt_id (str), args (list), and meta (dict). The meta dict is the subset of assertion metadata restricted to the keys ingested_at, source, run_id, candidate_id, candidate_key, cand_key_digest.
• chosen_asrt_id (str | None): the chosen assertion id for the predicate cell, or None when no active claim is chosen.
• asrt_id (str): the resolved assertion id of target.
• chosen (bool): True when chosen_asrt_id equals the target’s asrt_id.

Raises:

• SDKStoreError when target is neither a non-empty string nor an object with a non-empty asrt_id attribute (message: fg.audit.explain(...) expects asrt_id string or AssertionRecord).
• SDKStoreError when the resolved assertion id is not found in the ledger (message: fg.audit.explain(...) assertion id not found: <asrt_id>).

Example:

record = fg.assertions.by_id(asrt_id)
explanation = fg.audit.explain(record)
explanation["asrt_id"]   # the target assertion id
explanation["pred_id"]   # the predicate id
explanation["e_ref"]     # the entity reference
explanation["chosen"]    # True if this assertion is the chosen one

fg.audit.conflicts(target)#

Return conflict diagnostics for an assertion record or for an entity field cell.

Parameters:

• target: one of:
  • an assertion id str or an object exposing a non-empty asrt_id attribute, resolved to the claim’s (pred_id, e_ref) cell; or
  • a 2-tuple (entity, field) where field is an sdk.Field descriptor or a field-name str, and entity is an entity reference str or an object exposing a ref attribute (for example an EntitySnapshot).

Returns: dict[str, Any] with the following keys:

• pred_id (str): the predicate id of the resolved cell.
• e_ref (str): the entity reference of the resolved cell.
• active_asrt_ids (list[str]): the ids of all active claims on (pred_id, e_ref).
• chosen_asrt_id (str | None): the chosen assertion id for the cell, or None when no active claim is chosen.

Raises:

• SDKStoreError when a (entity, field) tuple is supplied with a field-name string but entity is not an entity reference string or EntitySnapshot (message: fg.audit.conflicts((entity, field_name)) expects entity ref string or EntitySnapshot).
• SDKStoreError when the second tuple element is neither an sdk.Field descriptor nor a field-name string (message: fg.audit.conflicts((entity, field)) expects sdk.Field descriptor or field name).
• SDKStoreError when no schema predicate is found for the resolved (entity_type, field) (message: schema predicate not found for <entity_type>.<field>).
• SDKStoreError when the resolved entity argument is not an entity reference string or EntitySnapshot (message: fg.audit.conflicts((entity, field)) expects entity ref string or EntitySnapshot).
• SDKStoreError when a non-tuple target resolves to an assertion id that is not a non-empty string or that is not found in the ledger (see the resolution and lookup messages under fg.audit.explain, with method name conflicts).

Example:

# By assertion record:
conflicts = fg.audit.conflicts(record)
conflicts["pred_id"]
conflicts["e_ref"]
conflicts["active_asrt_ids"]   # list of active claim ids on this cell

# By (entity ref, field name):
ref = fg.entities.ref(User, user_id="u-1")
conflicts = fg.audit.conflicts((ref, "name"))
conflicts["e_ref"]             # == ref
conflicts["active_asrt_ids"]

fg.audit.diff_proof_frames(round_a_id, round_b_id, round_a_events, round_b_events, *, warnings=(), include_unchanged=False)#

Compare two recorded rounds’ proof-frame outcomes. The method performs no I/O: callers supply the round events directly, for example loaded via factgraph.audit.load_audit_package or held from a recorder.

Parameters:

• round_a_id (str): non-empty identifier for the A-side round.
• round_b_id (str): non-empty identifier for the B-side round.
• round_a_events (tuple[RoundEvent, ...]): the raw RoundEvent tuple for the A-side round. RoundEvent is defined in factgraph.audit.round_events.
• round_b_events (tuple[RoundEvent, ...]): the raw RoundEvent tuple for the B-side round.
• warnings (tuple[WarningDTO, ...], default ()): warnings carried onto the result. WarningDTO is defined in factgraph.application.protocol.common.
• include_unchanged (bool, default False): when True, emit a delta for frames present on both sides whose status and atoms are unchanged.

Returns: ProofFrameDiff (from factgraph.audit.proof_frame_diff), a frozen dataclass with fields:

• round_a_id (str)
• round_b_id (str)
• frame_deltas (tuple[FrameDelta, ...])
• warnings (tuple[WarningDTO, ...], default empty tuple)

Each FrameDelta carries frame_identity (FrameIdentity), source_a (EventReference | None), source_b (EventReference | None), frame_status_change (FrameStatusChange | None), atom_deltas (tuple[AtomDelta, ...], default ()), and markers (tuple[FrameMarker, ...], default ()). FrameStatusChange has before and after, each a ProofFrameStatus | None. ProofFrameStatus is the literal "still_valid" | "invalidated" | "unknown". AtomDelta has condition_key (str), kind (one of "atom_added", "atom_removed", "atom_verdict_changed"), before_verdict (ProofFrameStatus | None), and after_verdict (ProofFrameStatus | None). The only FrameMarker value is "rule_refs_unsupported".

Raises: SDKStoreError for any input or runtime failure crossing the SDK boundary. The error’s path field locates the failure:

• $.diff_proof_frames.round_a_id: round_a_id is not a non-empty string.
• $.diff_proof_frames.round_b_id: round_b_id is not a non-empty string.
• $.diff_proof_frames.round_a_events: round_a_events is not a tuple, or contains a non-RoundEvent element.
• $.diff_proof_frames.round_b_events: round_b_events is not a tuple, or contains a non-RoundEvent element.
• $.diff_proof_frames.warnings: warnings is not a tuple, or contains a non-WarningDTO element.
• $.diff_proof_frames.include_unchanged: include_unchanged is not a bool.
• $.diff_proof_frames.request: a ProofFrameDiffError from internal validation (malformed event payloads, duplicate frame identity).
• $.diff_proof_frames: any other unexpected runtime exception.

The original exception is preserved as __cause__.

Example:

diff = fg.audit.diff_proof_frames(
    "round-a",
    "round-b",
    round_a_events,   # tuple[RoundEvent, ...]
    round_b_events,   # tuple[RoundEvent, ...]
)
diff.round_a_id
diff.round_b_id
for delta in diff.frame_deltas:
    delta.frame_identity
    delta.frame_status_change   # FrameStatusChange | None
    delta.atom_deltas           # tuple[AtomDelta, ...]

fg.meta#

The fg.meta namespace exposes read-only runtime introspection. The attribute fg.meta returns the meta namespace manager.

fg.meta  # read-only namespace; fg.meta.x = ... raises FrozenSnapshotError

fg.meta.capabilities()#

Return the enumerations the shipped runtime accepts.

Parameters: none.

Returns: Mapping[str, frozenset[str]]. The result is a frozen MappingProxyType whose values are frozenset[str]; both levels are immutable. The mapping has exactly these keys:

• value_kinds: frozenset({"scalar", "entity_ref"}).
• scalar_tags: the canonical primitive tags minus entity_ref, namely frozenset({"string", "int", "float64", "bool", "bytes", "time", "uuid"}).
• cardinalities: frozenset({"single", "multi"}).

Raises: none.

Example:

caps = fg.meta.capabilities()
set(caps.keys())          # {"value_kinds", "scalar_tags", "cardinalities"}
caps["value_kinds"]       # frozenset({"scalar", "entity_ref"})
caps["cardinalities"]     # frozenset({"single", "multi"})

See also#

• Three-layer API for fg.assertions and fg.entities, which produce the records and references passed to fg.audit.
• Evaluation for the proof-frame outcomes that diff_proof_frames compares.
• Errors for SDKStoreError and FrozenSnapshotError.