jeevesagent.data.lineage

Freshness and lineage policies for CertifiedValue.

A CertifiedValue (defined in jeevesagent.core.types) carries provenance metadata: source, fetched_at, optional valid_until, schema_version, and a tuple of upstream value IDs in lineage.

This module supplies the policy types and helpers. Two flavours:

• FreshnessPolicy — declare a maximum age per source prefix; valid_until always wins when set on the value itself.

• LineagePolicy — declare an allow-list of source prefixes every value in a lineage chain must originate from.

Two helper styles for each:

• check_* returns True/False so callers can branch.

• require_* raises the appropriate FreshnessError / LineageError so callers can rely on exception propagation.

Classes

FreshnessPolicy	Maximum age for certified values from each source.
LineagePolicy	Allow-list of source prefixes for the entire lineage chain.

Functions

check_freshness(→ bool)	Return True if value satisfies policy at now.
check_lineage(→ bool)	Return True if every lineage source is allowed.
require_freshness(→ None)	Raise FreshnessError when check_freshness() fails.
require_lineage(→ None)	Raise LineageError when check_lineage() fails.

Module Contents

class jeevesagent.data.lineage.FreshnessPolicy

Maximum age for certified values from each source.

per_source maps a source-prefix (matched with startswith) to a timedelta. The first prefix that matches wins. default is used when no prefix matches; if also None, the policy treats all values as fresh.

classmethod from_dict(per_source: dict[str, datetime.timedelta] | None = None, *, default: datetime.timedelta | None = None) → FreshnessPolicy

max_age_for(source: str) → datetime.timedelta | None

default: datetime.timedelta | None = None

per_source: tuple[tuple[str, datetime.timedelta], Ellipsis] = ()

class jeevesagent.data.lineage.LineagePolicy

Allow-list of source prefixes for the entire lineage chain.

A CertifiedValue is acceptable if every entry in value.lineage (interpreted as a source prefix) starts with one of the allowed prefixes.

classmethod from_iter(sources: list[str] | tuple[str, Ellipsis]) → LineagePolicy

allowed_sources: frozenset[str]

jeevesagent.data.lineage.check_freshness(value: jeevesagent.core.types.CertifiedValue, policy: FreshnessPolicy, *, now: datetime.datetime | None = None) → bool

Return True if value satisfies policy at now.

Logic:

1. If valid_until is set on the value, fail if now > valid_until.

2. Look up policy.max_age_for(source). If None (no rule), the value is fresh by default.

3. Otherwise fail if now - fetched_at > max_age.

jeevesagent.data.lineage.check_lineage(value: jeevesagent.core.types.CertifiedValue, policy: LineagePolicy) → bool

Return True if every lineage source is allowed.

The value’s own source is also required to be in the allow-list — there’s no point trusting a chain whose tip you don’t.

jeevesagent.data.lineage.require_freshness(value: jeevesagent.core.types.CertifiedValue, policy: FreshnessPolicy, *, now: datetime.datetime | None = None) → None

Raise FreshnessError when check_freshness() fails.

jeevesagent.data.lineage.require_lineage(value: jeevesagent.core.types.CertifiedValue, policy: LineagePolicy) → None

Raise LineageError when check_lineage() fails.