jeevesagent.data

Data certification: provenance, freshness, and lineage policies.

Every value the agent reasons over can carry a CertifiedValue wrapper from jeevesagent.core.types (already there). This module adds the policies that decide whether a certified value is acceptable in context.

Submodules

• jeevesagent.data.lineage

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.

Package Contents

class jeevesagent.data.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.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.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.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.require_freshness(value: jeevesagent.core.types.CertifiedValue, policy: FreshnessPolicy, *, now: datetime.datetime | None = None) → None

Raise FreshnessError when check_freshness() fails.

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

Raise LineageError when check_lineage() fails.