Python API Reference

Top-level exports

import foghorn

foghorn

foghorn - Decision staleness alerts for AI agents.

ActivityFrame(frame_id, application, site, t_start, t_end, input_volume, evidence_ptrs, frame_type='activity', row_count=0) dataclass

Compiled typed activity frame (deterministic, zero-model).

Bounded episode carrying application, site, timing, input volume, and evidence pointers back to raw rows - byte-identical and cacheable.

RawCaptureRow(row_id, timestamp, application, site='', input_kind='', meta=dict()) dataclass

One raw capture stream row (pre-compile).

Attributes:

Name	Type	Description
row_id	str	Stable id of the raw row (evidence pointer target).
timestamp	float	Unix seconds (or any monotonic clock).
application	str	Foreground application name.
site	str	Optional site/URL host (browser) or window title token.
input_kind	str	Optional input class (key, click, scroll, …).
meta	dict[str, Any]	Optional extra fields (ignored by compile fingerprint except when callers pass them through explicitly).

ClosedLoopError

Bases: ValueError

Raised when the gate refuses empty, unusable, or misused worlds.

GateOutcome(ok, verdict, reason, exit_code, alerts=(), max_impact=0.0, stale_source_ids=(), oldest_age_seconds=None, source_count=0, human_required=False) dataclass

Result of a closed-loop read of a foghorn world or source-age gate.

Attributes:

Name	Type	Description
ok	bool	True only when a pipeline may continue (no high-impact stale).
verdict	str	PASS, FAIL, or FAIL_LOUD.
reason	str	Human-readable explanation (always non-empty).
exit_code	int	0 PASS, 1 FAIL (stale), 2 FAIL_LOUD (empty/misuse).
alerts	tuple[StalenessAlert, ...]	Staleness alerts when scoring ran.
max_impact	float	Highest impact_score among alerts (0.0 if none).
stale_source_ids	tuple[str, ...]	Fact ids that exceeded max source age.
oldest_age_seconds	float | None	Age of the oldest examined source fact.
source_count	int	Number of source facts examined.
human_required	bool	True when refresh/re-retrieval needs a human or re-fetch.

to_dict()

Serialise for JSON reports (eagle-eyes dogfood, CI artifacts).

Source code in src/foghorn/closed_loop.py

def to_dict(self) -> dict[str, Any]:
    """Serialise for JSON reports (eagle-eyes dogfood, CI artifacts)."""
    return {
        "ok": self.ok,
        "verdict": self.verdict,
        "reason": self.reason,
        "exit_code": self.exit_code,
        "max_impact": self.max_impact,
        "alert_count": len(self.alerts),
        "alerts": [a.to_dict() for a in self.alerts],
        "stale_source_ids": list(self.stale_source_ids),
        "oldest_age_seconds": self.oldest_age_seconds,
        "source_count": self.source_count,
        "human_required": self.human_required,
    }

EvidenceLink(feature_id, source_id, source_kind='evidence_link', detail='') dataclass

Provenance edge from a feature to a source artifact.

Attributes:

Name	Type	Description
feature_id	str	Feature this evidence supports.
source_id	str	Id of source table row, fact, guideline, or raw capture.
source_kind	str	Class of source (source_table, guideline_cite, …).
detail	str	Optional free-text cite / path.

FeatureRecord(feature_id, name='', kind='derived_feature', rubric_ok=True, value=None, meta=dict()) dataclass

A derived/engineered feature that must carry evidence provenance.

Attributes:

Name	Type	Description
feature_id	str	Stable id of the feature.
name	str	Human-readable feature name.
kind	str	Feature class (derived_feature, aggregated_feature, …).
rubric_ok	bool	Whether rubric/structure checks passed (default True).
value	Any	Optional payload (not gated on content).
meta	dict[str, Any]	Optional extra fields.

Decision(label, content, fact_ids=list(), recorded_at=time.time()) dataclass

A named agent conclusion recorded alongside the facts it depended on.

Decisions are the nodes that foghorn watches for staleness. When any Fact listed in fact_ids changes, this Decision is marked stale.

Attributes:

Name	Type	Description
id	str	Content-addressed identifier - SHA-256[:16] of "{label}|{content}".
label	str	Short slug describing the decision (e.g. "chose-redis-for-rate-limiting").
content	str	Full reasoning text or justification.
fact_ids	list[str]	IDs of the Facts this decision directly depended on.
recorded_at	float	Unix timestamp when this decision was recorded.

to_dict()

Serialize to a JSON-compatible dict.

Source code in src/foghorn/fact.py

def to_dict(self) -> dict[str, Any]:
    """Serialize to a JSON-compatible dict."""
    return {
        "id": self.id,
        "label": self.label,
        "content": self.content,
        "fact_ids": self.fact_ids,
        "recorded_at": self.recorded_at,
    }

from_dict(d) classmethod

Deserialize from a dict produced by to_dict().

Source code in src/foghorn/fact.py

@classmethod
def from_dict(cls, d: dict[str, Any]) -> Decision:
    """Deserialize from a dict produced by to_dict()."""
    dec = cls(
        label=d["label"],
        content=d["content"],
        fact_ids=d.get("fact_ids", []),
        recorded_at=d.get("recorded_at", 0.0),
    )
    return dec

Fact(subject, predicate, object, confidence=1.0, recorded_at=time.time()) dataclass

An immutable, content-addressed triple that agents assert about the world.

Facts are the atoms of foghorn. Two Facts with the same subject, predicate, and object always have the same ID, regardless of when they were recorded.

Attributes:

Name	Type	Description
id	str	Content-addressed identifier - SHA-256[:16] of "{subject}|{predicate}|{object}".
subject	str	The entity this fact is about (e.g. "Redis").
predicate	str	The relationship being asserted (e.g. "is-appropriate-for").
object	str	The value of the assertion (e.g. "rate-limiting").
confidence	float	Belief weight in [0.0, 1.0]. Default 1.0 (certain).
recorded_at	float	Unix timestamp when this fact was committed.

to_dict()

Serialize to a JSON-compatible dict.

Source code in src/foghorn/fact.py

def to_dict(self) -> dict[str, Any]:
    """Serialize to a JSON-compatible dict."""
    return {
        "id": self.id,
        "subject": self.subject,
        "predicate": self.predicate,
        "object": self.object,
        "confidence": self.confidence,
        "recorded_at": self.recorded_at,
    }

from_dict(d) classmethod

Deserialize from a dict produced by to_dict().

Source code in src/foghorn/fact.py

@classmethod
def from_dict(cls, d: dict[str, Any]) -> Fact:
    """Deserialize from a dict produced by to_dict()."""
    f = cls(
        subject=d["subject"],
        predicate=d["predicate"],
        object=d["object"],
        confidence=d.get("confidence", 1.0),
        recorded_at=d.get("recorded_at", 0.0),
    )
    return f

StalenessAlert(decision_id, decision_label, stale_fact_ids, impact_score) dataclass

Emitted when a Decision's upstream facts have changed.

Attributes:

Name	Type	Description
decision_id	str	ID of the stale Decision.
decision_label	str	Human-readable label for display.
stale_fact_ids	list[str]	Which specific facts changed and triggered this alert.
impact_score	float	Confidence-weighted importance in [0.0, 1.0]. Higher = more confidence was placed in the now-changed facts.

to_dict()

Serialize to a JSON-compatible dict.

Source code in src/foghorn/fact.py

def to_dict(self) -> dict[str, Any]:
    """Serialize to a JSON-compatible dict."""
    return {
        "decision_id": self.decision_id,
        "decision_label": self.decision_label,
        "stale_fact_ids": self.stale_fact_ids,
        "impact_score": round(self.impact_score, 4),
    }

PropagationResult(changed_fact_ids, directly_stale=list(), transitively_stale=list(), propagation_depth=0, impact_summary='') dataclass

Result of propagating staleness from a set of changed facts.

Attributes:

Name	Type	Description
changed_fact_ids	list[str]	The fact IDs that triggered the propagation.
directly_stale	list[str]	Labels of decisions that directly depend on changed facts.
transitively_stale	list[str]	Labels of decisions that depend on directly stale decisions (requires decisions to reference other decisions via their fact_ids - if the architecture only records fact→decision edges, this will be empty).
propagation_depth	int	Maximum depth reached in the propagation graph.
impact_summary	str	Human-readable summary of the propagation result.

Recommendation(decision_label, reason, action, priority, stale_facts=list()) dataclass

An actionable recommendation for a stale decision.

Attributes:

Name	Type	Description
decision_label	str	The label of the stale decision.
reason	str	Why the decision is considered stale.
action	str	Recommended action: "re-evaluate", "archive", or "update-fact".
priority	str	"critical", "high", "medium", or "low".
stale_facts	list[str]	Subjects of the facts that triggered this recommendation.

to_dict()

Serialize to a plain dict.

Source code in src/foghorn/recommend.py

def to_dict(self) -> dict[str, object]:
    """Serialize to a plain dict."""
    return {
        "decision_label": self.decision_label,
        "reason": self.reason,
        "action": self.action,
        "priority": self.priority,
        "stale_facts": self.stale_facts,
    }

WorldRepo(store)

A foghorn repository: a versioned store of agent facts and decisions.

WorldRepo is the main user-facing API. It wraps WorldStore with the higher-level operations of a version-controlled knowledge base.

Typical workflow::

repo = WorldRepo.init(".foghorn")
repo.add_fact("Redis", "is-appropriate-for", "rate-limiting")
repo.decide("chose-redis", "Redis fits our rate-limiter needs",
            depends_on=[...fact_ids...])
commit = repo.commit("Initial architecture decisions")

Attributes:

Name	Type	Description
store		The underlying WorldStore.
path		Path to the repository database.

Source code in src/foghorn/repo.py

def __init__(self, store: WorldStore) -> None:
    self.store = store
    self.path = store.path

init(path='.foghorn/world.db') classmethod

Create or open a WorldRepo at the given path.

Parameters:

Name	Type	Description	Default
path	str | Path	Path to the SQLite database file. Parent directories are created automatically.	'.foghorn/world.db'

Returns:

Type	Description
WorldRepo	A WorldRepo ready for use.

Source code in src/foghorn/repo.py

@classmethod
def init(cls, path: str | Path = ".foghorn/world.db") -> WorldRepo:
    """Create or open a WorldRepo at the given path.

    Args:
        path: Path to the SQLite database file. Parent directories
            are created automatically.

    Returns:
        A WorldRepo ready for use.
    """
    return cls(WorldStore(path))

add_fact(subject, predicate, obj, confidence=1.0)

Stage a new Fact for the next commit.

Parameters:

Name	Type	Description	Default
subject	str	The entity this fact is about.	required
predicate	str	The relationship being asserted.	required
obj	str	The value of the assertion.	required
confidence	float	Belief weight in [0.0, 1.0].	1.0

Returns:

Type	Description
Fact	The created (and staged) Fact.

Source code in src/foghorn/repo.py

def add_fact(
    self,
    subject: str,
    predicate: str,
    obj: str,
    confidence: float = 1.0,
) -> Fact:
    """Stage a new Fact for the next commit.

    Args:
        subject: The entity this fact is about.
        predicate: The relationship being asserted.
        obj: The value of the assertion.
        confidence: Belief weight in [0.0, 1.0].

    Returns:
        The created (and staged) Fact.
    """
    fact = Fact(subject=subject, predicate=predicate, object=obj, confidence=confidence)
    self.store.add_fact(fact)
    return fact

decide(label, content, depends_on=None)

Stage a new Decision for the next commit.

Parameters:

Name	Type	Description	Default
label	str	Short slug for this decision (e.g. "chose-redis-for-rate-limiting").	required
content	str	Full reasoning text.	required
depends_on	list[str] | None	List of Fact IDs this decision relied on.	None

Returns:

Type	Description
Decision	The created (and staged) Decision.

Source code in src/foghorn/repo.py

def decide(
    self,
    label: str,
    content: str,
    depends_on: list[str] | None = None,
) -> Decision:
    """Stage a new Decision for the next commit.

    Args:
        label: Short slug for this decision (e.g. "chose-redis-for-rate-limiting").
        content: Full reasoning text.
        depends_on: List of Fact IDs this decision relied on.

    Returns:
        The created (and staged) Decision.
    """
    decision = Decision(
        label=label,
        content=content,
        fact_ids=depends_on or [],
    )
    self.store.add_decision(decision)
    return decision

commit(message)

Commit all staged facts and decisions.

Parameters:

Name	Type	Description	Default
message	str	Human-readable commit message.	required

Returns:

Type	Description
WorldCommit	The new WorldCommit.

Raises:

Type	Description
ValueError	If there is nothing staged to commit.

Source code in src/foghorn/repo.py

def commit(self, message: str) -> WorldCommit:
    """Commit all staged facts and decisions.

    Args:
        message: Human-readable commit message.

    Returns:
        The new WorldCommit.

    Raises:
        ValueError: If there is nothing staged to commit.
    """
    if self.store.staged_count() == 0:
        raise ValueError("Nothing to commit - stage facts or decisions first.")
    return self.store.commit(message)

retract_fact(fact_id)

Stage a fact retraction so it is excluded from the next commit snapshot.

Parameters:

Name	Type	Description	Default
fact_id	str	ID of the Fact to remove from the next snapshot.	required

Source code in src/foghorn/repo.py

def retract_fact(self, fact_id: str) -> None:
    """Stage a fact retraction so it is excluded from the next commit snapshot.

    Args:
        fact_id: ID of the Fact to remove from the next snapshot.
    """
    self.store.retract_fact(fact_id)

stale(since=None)

Return staleness alerts for decisions affected by recent fact changes.

Compares HEAD to since (or HEAD's parent if None) and finds all Decisions whose upstream facts changed.

Parameters:

Name	Type	Description	Default
since	WorldCommit | None	The base commit to diff against. Defaults to HEAD's parent.	None

Returns:

Type	Description
list[StalenessAlert]	List of StalenessAlert sorted by impact_score descending.
list[StalenessAlert]	Empty list if nothing has changed or there are no decisions.

Source code in src/foghorn/repo.py

def stale(self, since: WorldCommit | None = None) -> list[StalenessAlert]:
    """Return staleness alerts for decisions affected by recent fact changes.

    Compares HEAD to ``since`` (or HEAD's parent if None) and finds all
    Decisions whose upstream facts changed.

    Args:
        since: The base commit to diff against. Defaults to HEAD's parent.

    Returns:
        List of StalenessAlert sorted by impact_score descending.
        Empty list if nothing has changed or there are no decisions.
    """
    head = self.store.head()
    if head is None:
        return []

    base: WorldCommit | None
    if since is not None:
        base = since
    else:
        if head.parent_id is None:
            # No prior state to compare against; nothing can be stale yet.
            return []
        base = self.store.get_commit(head.parent_id)

    diff = diff_commits(self.store, base, head)
    return compute_staleness(self.store, diff.changed_fact_ids)

diff(commit_a=None, commit_b=None)

Diff two commits (defaults to HEAD~1 vs HEAD).

Parameters:

Name	Type	Description	Default
commit_a	WorldCommit | None	Base commit (None = empty state).	None
commit_b	WorldCommit | None	Head commit (None = current HEAD).	None

Returns:

Type	Description
DiffResult	DiffResult with added and removed facts.

Raises:

Type	Description
ValueError	If HEAD is empty and no commits are provided.

Source code in src/foghorn/repo.py

def diff(
    self,
    commit_a: WorldCommit | None = None,
    commit_b: WorldCommit | None = None,
) -> DiffResult:
    """Diff two commits (defaults to HEAD~1 vs HEAD).

    Args:
        commit_a: Base commit (None = empty state).
        commit_b: Head commit (None = current HEAD).

    Returns:
        DiffResult with added and removed facts.

    Raises:
        ValueError: If HEAD is empty and no commits are provided.
    """
    if commit_b is None:
        commit_b = self.store.head()
        if commit_b is None:
            raise ValueError("Repository has no commits yet.")

    if commit_a is None and commit_b.parent_id:
        commit_a = self.store.get_commit(commit_b.parent_id)

    return diff_commits(self.store, commit_a, commit_b)

log()

Return all commits from HEAD to root, newest first.

Source code in src/foghorn/repo.py

def log(self) -> list[WorldCommit]:
    """Return all commits from HEAD to root, newest first."""
    return self.store.log()

export_json()

Export the entire repository state as a JSON string.

Delegates to :func:foghorn.export.export_json.

Returns:

Type	Description
str	A JSON string with all facts, decisions, and commits.

Source code in src/foghorn/repo.py

def export_json(self) -> str:
    """Export the entire repository state as a JSON string.

    Delegates to :func:`foghorn.export.export_json`.

    Returns:
        A JSON string with all facts, decisions, and commits.
    """
    from foghorn.export import export_json

    return export_json(self)

recommend()

Generate actionable staleness recommendations.

Delegates to :func:foghorn.recommend.recommend.

Returns:

Type	Description
list[Recommendation]	Sorted list of :class:~foghorn.recommend.Recommendation objects.

Source code in src/foghorn/repo.py

def recommend(self) -> list[Recommendation]:
    """Generate actionable staleness recommendations.

    Delegates to :func:`foghorn.recommend.recommend`.

    Returns:
        Sorted list of :class:`~foghorn.recommend.Recommendation` objects.
    """
    from foghorn.recommend import recommend

    return recommend(self)

propagate(fact_ids)

Propagate staleness from a set of changed facts.

Delegates to :func:foghorn.propagate.propagate_staleness.

Parameters:

Name	Type	Description	Default
fact_ids	list[str]	IDs of facts that have changed.	required

Returns:

Name	Type	Description
A	PropagationResult	class:~foghorn.propagate.PropagationResult with stale decisions.

Source code in src/foghorn/repo.py

def propagate(self, fact_ids: list[str]) -> PropagationResult:
    """Propagate staleness from a set of changed facts.

    Delegates to :func:`foghorn.propagate.propagate_staleness`.

    Args:
        fact_ids: IDs of facts that have changed.

    Returns:
        A :class:`~foghorn.propagate.PropagationResult` with stale decisions.
    """
    from foghorn.propagate import propagate_staleness

    return propagate_staleness(self, fact_ids)

close()

Close the underlying database connection.

Source code in src/foghorn/repo.py

def close(self) -> None:
    """Close the underlying database connection."""
    self.store.close()

DiffResult(added_facts, removed_facts, changed_fact_ids, commit_a_id, commit_b_id) dataclass

Summary of changes between two world commits.

Attributes:

Name	Type	Description
added_facts	list[Fact]	Facts present in b but not a.
removed_facts	list[Fact]	Facts present in a but not b.
changed_fact_ids	set[str]	Union of added and removed fact IDs (convenience set).
commit_a_id	str | None	ID of the base commit (or None for the empty state).
commit_b_id	str	ID of the head commit.

WorldCommit(message, fact_ids=set(), decision_ids=set(), parent_id=None, timestamp=time.time()) dataclass

A snapshot of world state at a point in time.

Attributes:

Name	Type	Description
id	str	Content-addressed identifier.
message	str	Human-readable commit message.
fact_ids	set[str]	Set of Fact IDs in this snapshot.
decision_ids	set[str]	Set of Decision IDs in this snapshot.
parent_id	str | None	ID of the parent commit, or None for the initial commit.
timestamp	float	Unix timestamp of this commit.

to_dict()

Serialize to a JSON-compatible dict.

Source code in src/foghorn/store.py

def to_dict(self) -> dict[str, Any]:
    """Serialize to a JSON-compatible dict."""
    return {
        "id": self.id,
        "message": self.message,
        "fact_ids": sorted(self.fact_ids),
        "decision_ids": sorted(self.decision_ids),
        "parent_id": self.parent_id,
        "timestamp": self.timestamp,
    }

from_dict(d) classmethod

Deserialize from a dict produced by to_dict().

Source code in src/foghorn/store.py

@classmethod
def from_dict(cls, d: dict[str, Any]) -> WorldCommit:
    """Deserialize from a dict produced by to_dict()."""
    c = cls(
        message=d["message"],
        fact_ids=set(d.get("fact_ids", [])),
        decision_ids=set(d.get("decision_ids", [])),
        parent_id=d.get("parent_id"),
        timestamp=d.get("timestamp", 0.0),
    )
    if "id" in d:
        c.id = d["id"]
    return c

WorldStore(path)

SQLite-backed persistence layer for foghorn.

All Facts, Decisions, and Commits are stored in a single SQLite database. Content-addressed IDs guarantee deduplication: storing the same fact twice is a no-op.

Attributes:

Name	Type	Description
path		Path to the SQLite database file.

Source code in src/foghorn/store.py

def __init__(self, path: str | Path) -> None:
    self.path = Path(path)
    self.path.parent.mkdir(parents=True, exist_ok=True)
    self._conn = sqlite3.connect(str(self.path))
    self._conn.row_factory = sqlite3.Row
    self._conn.executescript(self._SCHEMA)
    self._conn.commit()

close()

Close the database connection.

Source code in src/foghorn/store.py

def close(self) -> None:
    """Close the database connection."""
    self._conn.close()

add_fact(fact)

Store a Fact (no-op if already stored).

Source code in src/foghorn/store.py

def add_fact(self, fact: Fact) -> None:
    """Store a Fact (no-op if already stored)."""
    self._conn.execute(
        "INSERT OR IGNORE INTO facts VALUES (?,?,?,?,?,?)",
        (fact.id, fact.subject, fact.predicate, fact.object, fact.confidence, fact.recorded_at),
    )
    self._conn.execute(
        "INSERT OR IGNORE INTO staging VALUES (?,?)",
        ("fact", fact.id),
    )
    self._conn.commit()

get_fact(fact_id)

Retrieve a Fact by ID, or None if not found.

Source code in src/foghorn/store.py

def get_fact(self, fact_id: str) -> Fact | None:
    """Retrieve a Fact by ID, or None if not found."""
    row = self._conn.execute("SELECT * FROM facts WHERE id=?", (fact_id,)).fetchone()
    if row is None:
        return None
    return Fact.from_dict(dict(row))

list_facts()

Return all stored Facts ordered by recorded_at (oldest first).

D-FOGHORN warning: this is an append-oriented history view. Do not use next(iter(list_facts())) or list_facts()[0] as the "current" value of a subject/predicate - that is the oldest fact and caused full pipeline recaptures in production (Pioneer Content Foundry, 2026-07-22). Use :meth:latest_fact or :meth:list_facts_for instead.

Source code in src/foghorn/store.py

def list_facts(self) -> list[Fact]:
    """Return all stored Facts ordered by recorded_at (oldest first).

    **D-FOGHORN warning:** this is an append-oriented history view.
    Do **not** use ``next(iter(list_facts()))`` or ``list_facts()[0]`` as
    the "current" value of a subject/predicate - that is the *oldest*
    fact and caused full pipeline recaptures in production (Pioneer
    Content Foundry, 2026-07-22). Use :meth:`latest_fact` or
    :meth:`list_facts_for` instead.
    """
    rows = self._conn.execute("SELECT * FROM facts ORDER BY recorded_at").fetchall()
    return [Fact.from_dict(dict(r)) for r in rows]

list_facts_for(subject, predicate=None)

Return facts for a subject (and optional predicate), oldest first.

Use this when you need the history of a key. For the single current value of (subject, predicate), prefer :meth:latest_fact.

Source code in src/foghorn/store.py

def list_facts_for(self, subject: str, predicate: str | None = None) -> list[Fact]:
    """Return facts for a subject (and optional predicate), oldest first.

    Use this when you need the *history* of a key. For the single current
    value of ``(subject, predicate)``, prefer :meth:`latest_fact`.
    """
    if predicate is None:
        rows = self._conn.execute(
            "SELECT * FROM facts WHERE subject=? ORDER BY recorded_at",
            (subject,),
        ).fetchall()
    else:
        rows = self._conn.execute(
            "SELECT * FROM facts WHERE subject=? AND predicate=? ORDER BY recorded_at",
            (subject, predicate),
        ).fetchall()
    return [Fact.from_dict(dict(r)) for r in rows]

latest_fact(subject, predicate)

Return the most recently recorded fact for (subject, predicate).

Real-world case (Qdrant / farm_memory D-FOGHORN): Foundry modules treated list_facts() as a LWW current-state map and took next(...) → oldest object → false script_changed → wiped all frame dirs (~95 min recapture). This method is the correct reader for "what is the current value of this key in the append-only log?"

Note: Fact IDs are content-addressed on subject|predicate|object, so changing the object creates a new row; "current" = max(recorded_at).

Source code in src/foghorn/store.py

def latest_fact(self, subject: str, predicate: str) -> Fact | None:
    """Return the most recently *recorded* fact for ``(subject, predicate)``.

    Real-world case (Qdrant / farm_memory D-FOGHORN): Foundry modules treated
    ``list_facts()`` as a LWW current-state map and took ``next(...)`` →
    oldest object → false ``script_changed`` → wiped all frame dirs (~95 min
    recapture). This method is the correct reader for "what is the current
    value of this key in the append-only log?"

    Note: Fact IDs are content-addressed on ``subject|predicate|object``, so
    changing the object creates a new row; "current" = max(recorded_at).
    """
    row = self._conn.execute(
        "SELECT * FROM facts WHERE subject=? AND predicate=? ORDER BY recorded_at DESC LIMIT 1",
        (subject, predicate),
    ).fetchone()
    if row is None:
        return None
    return Fact.from_dict(dict(row))

current_fact_map()

Map each (subject, predicate) to its latest recorded Fact.

Safe replacement for the anti-pattern of scanning list_facts() in insertion order and treating the first hit as current.

Source code in src/foghorn/store.py

def current_fact_map(self) -> dict[tuple[str, str], Fact]:
    """Map each ``(subject, predicate)`` to its latest recorded Fact.

    Safe replacement for the anti-pattern of scanning ``list_facts()`` in
    insertion order and treating the first hit as current.
    """
    # One pass: list ordered by recorded_at ascending, last write wins per key
    current: dict[tuple[str, str], Fact] = {}
    for fact in self.list_facts():
        current[(fact.subject, fact.predicate)] = fact
    return current

add_decision(decision)

Store a Decision and its dependency edges (no-op if already stored).

Source code in src/foghorn/store.py

def add_decision(self, decision: Decision) -> None:
    """Store a Decision and its dependency edges (no-op if already stored)."""
    self._conn.execute(
        "INSERT OR IGNORE INTO decisions VALUES (?,?,?,?)",
        (decision.id, decision.label, decision.content, decision.recorded_at),
    )
    for fid in decision.fact_ids:
        self._conn.execute(
            "INSERT OR IGNORE INTO decision_facts VALUES (?,?)",
            (decision.id, fid),
        )
    self._conn.execute(
        "INSERT OR IGNORE INTO staging VALUES (?,?)",
        ("decision", decision.id),
    )
    self._conn.commit()

get_decision(decision_id)

Retrieve a Decision by ID, or None if not found.

Source code in src/foghorn/store.py

def get_decision(self, decision_id: str) -> Decision | None:
    """Retrieve a Decision by ID, or None if not found."""
    row = self._conn.execute("SELECT * FROM decisions WHERE id=?", (decision_id,)).fetchone()
    if row is None:
        return None
    fact_ids = [
        r[0]
        for r in self._conn.execute(
            "SELECT fact_id FROM decision_facts WHERE decision_id=?", (decision_id,)
        ).fetchall()
    ]
    d = Decision.from_dict(dict(row))
    d.fact_ids = fact_ids
    return d

list_decisions()

Return all stored Decisions ordered by recorded_at.

Source code in src/foghorn/store.py

def list_decisions(self) -> list[Decision]:
    """Return all stored Decisions ordered by recorded_at."""
    rows = self._conn.execute("SELECT * FROM decisions ORDER BY recorded_at").fetchall()

    if not rows:
        return []

    # Batch fetch all fact IDs for these decisions in one query
    decision_ids = [r["id"] for r in rows]
    placeholders = ",".join("?" * len(decision_ids))
    # Placeholders are built from "?" * N - no injection risk
    df_sql = (
        "SELECT decision_id, fact_id FROM decision_facts "
        f"WHERE decision_id IN ({placeholders})"  # nosec B608 - placeholders are "?" only
    )
    fact_rows = self._conn.execute(df_sql, decision_ids).fetchall()

    # Group fact IDs by decision
    fact_map: dict[str, list[str]] = {r["id"]: [] for r in rows}
    for fr in fact_rows:
        fact_map[fr["decision_id"]].append(fr["fact_id"])

    result = []
    for row in rows:
        d = Decision.from_dict(dict(row))
        d.fact_ids = fact_map[d.id]
        result.append(d)
    return result

get_decisions_for_fact(fact_id)

Return all Decisions that depend on a given Fact.

Source code in src/foghorn/store.py

def get_decisions_for_fact(self, fact_id: str) -> list[Decision]:
    """Return all Decisions that depend on a given Fact."""
    decision_id_rows = self._conn.execute(
        "SELECT decision_id FROM decision_facts WHERE fact_id=?", (fact_id,)
    ).fetchall()

    if not decision_id_rows:
        return []

    decision_ids = [r[0] for r in decision_id_rows]
    placeholders = ",".join("?" * len(decision_ids))

    # Batch fetch all decision rows in one query
    sql_dec = f"SELECT * FROM decisions WHERE id IN ({placeholders})"  # nosec B608
    rows = self._conn.execute(sql_dec, decision_ids).fetchall()

    if not rows:
        return []

    # Batch fetch all fact IDs for these decisions in one query
    # Placeholders are built from "?" * N - no injection risk
    df2_sql = (
        "SELECT decision_id, fact_id FROM decision_facts "
        f"WHERE decision_id IN ({placeholders})"  # nosec B608
    )
    fact_rows = self._conn.execute(df2_sql, decision_ids).fetchall()

    # Group fact IDs by decision
    fact_map: dict[str, list[str]] = {did: [] for did in decision_ids}
    for fr in fact_rows:
        fact_map[fr["decision_id"]].append(fr["fact_id"])

    decisions = []
    for row in rows:
        d = Decision.from_dict(dict(row))
        d.fact_ids = fact_map.get(d.id, [])
        decisions.append(d)
    return decisions

retract_fact(fact_id)

Stage a fact retraction so it is excluded from the next commit snapshot.

Source code in src/foghorn/store.py

def retract_fact(self, fact_id: str) -> None:
    """Stage a fact retraction so it is excluded from the next commit snapshot."""
    self._conn.execute(
        "INSERT OR IGNORE INTO staging VALUES (?,?)",
        ("retraction", fact_id),
    )
    self._conn.commit()

commit(message)

Create a new commit from staged facts and decisions.

Source code in src/foghorn/store.py

def commit(self, message: str) -> WorldCommit:
    """Create a new commit from staged facts and decisions."""
    staged_facts = {
        r[0] for r in self._conn.execute("SELECT id FROM staging WHERE type='fact'").fetchall()
    }
    staged_decisions = {
        r[0]
        for r in self._conn.execute("SELECT id FROM staging WHERE type='decision'").fetchall()
    }
    retracted_facts = {
        r[0]
        for r in self._conn.execute("SELECT id FROM staging WHERE type='retraction'").fetchall()
    }

    head_id = self._get_ref("HEAD")
    parent = self.get_commit(head_id) if head_id else None

    parent_fact_ids = parent.fact_ids if parent else set()
    parent_decision_ids = parent.decision_ids if parent else set()

    wc = WorldCommit(
        message=message,
        fact_ids=(parent_fact_ids | staged_facts) - retracted_facts,
        decision_ids=parent_decision_ids | staged_decisions,
        parent_id=head_id,
    )

    self._conn.execute(
        "INSERT OR IGNORE INTO commits VALUES (?,?,?,?)",
        (wc.id, wc.message, wc.parent_id, wc.timestamp),
    )
    for fid in wc.fact_ids:
        self._conn.execute("INSERT OR IGNORE INTO commit_facts VALUES (?,?)", (wc.id, fid))
    for did in wc.decision_ids:
        self._conn.execute("INSERT OR IGNORE INTO commit_decisions VALUES (?,?)", (wc.id, did))
    self._conn.execute("INSERT OR REPLACE INTO refs VALUES (?,?)", ("HEAD", wc.id))
    self._conn.execute("DELETE FROM staging")
    self._conn.commit()
    return wc

get_commit(commit_id)

Retrieve a WorldCommit by ID, or None if not found.

Source code in src/foghorn/store.py

def get_commit(self, commit_id: str) -> WorldCommit | None:
    """Retrieve a WorldCommit by ID, or None if not found."""
    row = self._conn.execute("SELECT * FROM commits WHERE id=?", (commit_id,)).fetchone()
    if row is None:
        return None
    fact_ids = {
        r[0]
        for r in self._conn.execute(
            "SELECT fact_id FROM commit_facts WHERE commit_id=?", (commit_id,)
        ).fetchall()
    }
    decision_ids = {
        r[0]
        for r in self._conn.execute(
            "SELECT decision_id FROM commit_decisions WHERE commit_id=?", (commit_id,)
        ).fetchall()
    }
    wc = WorldCommit.from_dict(dict(row))
    wc.fact_ids = fact_ids
    wc.decision_ids = decision_ids
    return wc

log()

Return all commits from HEAD to root, newest first.

Source code in src/foghorn/store.py

def log(self) -> list[WorldCommit]:
    """Return all commits from HEAD to root, newest first."""
    head_id = self._get_ref("HEAD")
    if not head_id:
        return []
    commits: list[WorldCommit] = []
    current_id: str | None = head_id
    seen: set[str] = set()
    while current_id and current_id not in seen:
        seen.add(current_id)
        wc = self.get_commit(current_id)
        if wc is None:
            break
        commits.append(wc)
        current_id = wc.parent_id
    return commits

head()

Return the HEAD commit, or None if the repo is empty.

Source code in src/foghorn/store.py

def head(self) -> WorldCommit | None:
    """Return the HEAD commit, or None if the repo is empty."""
    head_id = self._get_ref("HEAD")
    return self.get_commit(head_id) if head_id else None

staged_count()

Return number of staged (uncommitted) items.

Source code in src/foghorn/store.py

def staged_count(self) -> int:
    """Return number of staged (uncommitted) items."""
    row = self._conn.execute("SELECT COUNT(*) FROM staging").fetchone()
    return row[0] if row else 0

activity_frame_fingerprint(*, application, site, t_start, t_end, evidence_ptrs, input_volume)

Stable SHA-256 hex of the frame content (byte-identical across runs).

Source code in src/foghorn/activity.py

def activity_frame_fingerprint(
    *,
    application: str,
    site: str,
    t_start: float,
    t_end: float,
    evidence_ptrs: Sequence[str],
    input_volume: int,
) -> str:
    """Stable SHA-256 hex of the frame content (byte-identical across runs)."""
    payload = {
        "application": _canon_app(application),
        "site": _canon_site(site),
        "t_start": float(t_start),
        "t_end": float(t_end),
        "evidence_ptrs": list(evidence_ptrs),
        "input_volume": int(input_volume),
    }
    blob = json.dumps(payload, sort_keys=True, separators=(",", ":")).encode("utf-8")
    return hashlib.sha256(blob).hexdigest()

assert_activity_memory_ok(frames=None, **kwargs)

Raise :class:ClosedLoopError unless :func:gate_activity_memory is ok.

Source code in src/foghorn/activity.py

def assert_activity_memory_ok(
    frames: Sequence[ActivityFrame | dict[str, Any]] | None = None,
    **kwargs: Any,
) -> GateOutcome:
    """Raise :class:`ClosedLoopError` unless :func:`gate_activity_memory` is ok."""
    outcome = gate_activity_memory(frames, **kwargs)
    if not outcome.ok:
        raise ClosedLoopError(f"{outcome.verdict}: {outcome.reason}")
    return outcome

compile_activity_frames(rows, *, gap_split_seconds=DEFAULT_GAP_SPLIT_SECONDS)

Segment raw capture into typed activity frames (deterministic, zero-model).

Split rules (all mechanical - no LLM):

1. Sort by timestamp ascending (stable on equal timestamps by row_id).
2. Start a new frame when application or site changes vs previous row.
3. Start a new frame when the gap from previous row exceeds gap_split_seconds (default 300s).
4. input_volume = count of rows with non-empty input_kind.
5. evidence_ptrs = ordered row_id list for rows in the frame.
6. frame_id = :func:activity_frame_fingerprint of the content.

Empty input → empty list (caller may FAIL_LOUD via the gate).

Source code in src/foghorn/activity.py

def compile_activity_frames(
    rows: Sequence[RawCaptureRow | dict[str, Any]],
    *,
    gap_split_seconds: float = DEFAULT_GAP_SPLIT_SECONDS,
) -> list[ActivityFrame]:
    """Segment raw capture into typed activity frames (deterministic, zero-model).

    Split rules (all mechanical - no LLM):

    1. Sort by ``timestamp`` ascending (stable on equal timestamps by ``row_id``).
    2. Start a new frame when **application** or **site** changes vs previous row.
    3. Start a new frame when the gap from previous row exceeds
       ``gap_split_seconds`` (default 300s).
    4. ``input_volume`` = count of rows with non-empty ``input_kind``.
    5. ``evidence_ptrs`` = ordered ``row_id`` list for rows in the frame.
    6. ``frame_id`` = :func:`activity_frame_fingerprint` of the content.

    Empty input → empty list (caller may FAIL_LOUD via the gate).
    """
    if gap_split_seconds < 0:
        raise ValueError("gap_split_seconds must be >= 0")

    parsed = [_row_from_mapping(r) for r in rows]
    if not parsed:
        return []

    ordered = sorted(parsed, key=lambda r: (r.timestamp, r.row_id))

    frames: list[ActivityFrame] = []
    bucket: list[RawCaptureRow] = [ordered[0]]

    def _flush(group: list[RawCaptureRow]) -> None:
        if not group:
            return
        app = _canon_app(group[0].application)
        site = _canon_site(group[0].site)
        t_start = float(group[0].timestamp)
        t_end = float(group[-1].timestamp)
        ptrs = tuple(r.row_id for r in group)
        volume = sum(1 for r in group if (r.input_kind or "").strip())
        fid = activity_frame_fingerprint(
            application=app,
            site=site,
            t_start=t_start,
            t_end=t_end,
            evidence_ptrs=ptrs,
            input_volume=volume,
        )
        frames.append(
            ActivityFrame(
                frame_id=fid,
                application=app,
                site=site,
                t_start=t_start,
                t_end=t_end,
                input_volume=volume,
                evidence_ptrs=ptrs,
                frame_type="activity",
                row_count=len(group),
            )
        )

    for prev, cur in itertools.pairwise(ordered):
        app_change = _canon_app(cur.application) != _canon_app(prev.application)
        site_change = _canon_site(cur.site) != _canon_site(prev.site)
        gap = float(cur.timestamp) - float(prev.timestamp)
        gap_split = gap > gap_split_seconds
        if app_change or site_change or gap_split:
            _flush(bucket)
            bucket = [cur]
        else:
            bucket.append(cur)
    _flush(bucket)
    return frames

frame_is_valid(frame)

True when a compiled frame is load-bearing (evidence + timing).

Source code in src/foghorn/activity.py

def frame_is_valid(frame: ActivityFrame) -> bool:
    """True when a compiled frame is load-bearing (evidence + timing)."""
    if not frame.frame_id or not frame.application:
        return False
    if frame.t_end < frame.t_start:
        return False
    if not frame.evidence_ptrs:
        return False
    return not any(not str(p).strip() for p in frame.evidence_ptrs)

gate_activity_memory(frames=None, *, memory_mode='compiled', raw_rows=None, require_frames=True, require_evidence=True, gap_split_seconds=DEFAULT_GAP_SPLIT_SECONDS, claimed_frame_ids=None)

Refuse non-deterministic or evidence-free activity memory.

Activity Frames class (arXiv 2608.05784):

• memory_mode="llm_summary" → FAIL - LLM day-summary is not load-bearing memory (paper accuracy gap vs compiled frames).
• memory_mode="raw_uncompiled" → FAIL - must compile first.
• memory_mode="compiled" with no frames when required → FAIL_LOUD.
• Frame missing evidence pointers → FAIL_LOUD.
• Invalid timing (t_end < t_start) → FAIL.
• claimed_frame_ids not subset of compiled inventory → FAIL.
• Valid compiled frames with evidence → PASS.

If raw_rows is provided and frames is None/empty under compiled mode, frames are compiled in-process (deterministic) before gating.

Source code in src/foghorn/activity.py

def gate_activity_memory(
    frames: Sequence[ActivityFrame | dict[str, Any]] | None = None,
    *,
    memory_mode: MemoryMode = "compiled",
    raw_rows: Sequence[RawCaptureRow | dict[str, Any]] | None = None,
    require_frames: bool = True,
    require_evidence: bool = True,
    gap_split_seconds: float = DEFAULT_GAP_SPLIT_SECONDS,
    claimed_frame_ids: Iterable[str] | None = None,
) -> GateOutcome:
    """Refuse non-deterministic or evidence-free activity memory.

    Activity Frames class (arXiv 2608.05784):

    * ``memory_mode="llm_summary"`` → **FAIL** - LLM day-summary is not
      load-bearing memory (paper accuracy gap vs compiled frames).
    * ``memory_mode="raw_uncompiled"`` → **FAIL** - must compile first.
    * ``memory_mode="compiled"`` with no frames when required → **FAIL_LOUD**.
    * Frame missing evidence pointers → **FAIL_LOUD**.
    * Invalid timing (``t_end < t_start``) → **FAIL**.
    * ``claimed_frame_ids`` not subset of compiled inventory → **FAIL**.
    * Valid compiled frames with evidence → **PASS**.

    If ``raw_rows`` is provided and ``frames`` is None/empty under compiled mode,
    frames are compiled in-process (deterministic) before gating.
    """
    mode = (memory_mode or "compiled").strip().lower()
    if mode not in {"compiled", "llm_summary", "raw_uncompiled"}:
        return _fail_loud(
            f"ACTIVITY-FRAMES: unknown memory_mode={memory_mode!r} "
            "(use compiled|llm_summary|raw_uncompiled)"
        )

    if mode == "llm_summary":
        return _fail(
            "ACTIVITY-FRAMES: memory_mode=llm_summary refused - "
            "LLM summaries of screen capture are not load-bearing activity "
            "memory (arXiv 2608.05784: compiled frames beat summaries). "
            "Compile raw rows with compile_activity_frames and use "
            "memory_mode='compiled'."
        )

    if mode == "raw_uncompiled":
        return _fail(
            "ACTIVITY-FRAMES: memory_mode=raw_uncompiled refused - "
            "raw capture must be compiled into typed activity frames "
            "(app/site/timing/input_volume/evidence_ptrs) before use as "
            "agent memory. Call compile_activity_frames(...)."
        )

    # compiled mode
    compiled: list[ActivityFrame] = []
    if frames:
        try:
            compiled = [_frame_from_mapping(f) for f in frames]
        except (TypeError, ValueError) as exc:
            return _fail_loud(
                f"ACTIVITY-FRAMES: invalid frame payload: {exc}",
            )

    if not compiled and raw_rows is not None:
        try:
            compiled = compile_activity_frames(raw_rows, gap_split_seconds=gap_split_seconds)
        except (TypeError, ValueError) as exc:
            return _fail_loud(
                f"ACTIVITY-FRAMES: compile failed: {exc}",
            )

    if require_frames and len(compiled) == 0:
        return _fail_loud(
            "ACTIVITY-FRAMES: no compiled activity frames - cannot ground "
            "session/day answers on empty activity inventory (record raw "
            "capture then compile_activity_frames)",
            source_count=0,
        )

    if not compiled:
        return GateOutcome(
            ok=True,
            verdict="PASS",
            reason="ACTIVITY-FRAMES: no frames required; nothing to gate",
            exit_code=0,
            source_count=0,
            human_required=False,
        )

    invalid_timing: list[str] = []
    missing_evidence: list[str] = []
    for fr in compiled:
        if fr.t_end < fr.t_start:
            invalid_timing.append(fr.frame_id[:16])
        if (
            require_evidence
            and not frame_is_valid(fr)
            and (not fr.evidence_ptrs or any(not str(p).strip() for p in fr.evidence_ptrs))
        ):
            # distinguish timing already caught
            missing_evidence.append(fr.frame_id[:16] or "(empty-id)")

    if missing_evidence:
        return _fail_loud(
            f"ACTIVITY-FRAMES: {len(missing_evidence)} frame(s) lack evidence "
            f"pointers back to raw rows ids={missing_evidence[:8]} - "
            "compiled memory must be mechanically auditable",
            source_count=len(compiled),
        )

    if invalid_timing:
        return _fail(
            f"ACTIVITY-FRAMES: {len(invalid_timing)} frame(s) have t_end < t_start "
            f"ids={invalid_timing[:8]}",
            source_count=len(compiled),
        )

    if claimed_frame_ids is not None:
        inventory = {f.frame_id for f in compiled}
        claimed = [str(c).strip() for c in claimed_frame_ids if str(c).strip()]
        missing = [c for c in claimed if c not in inventory]
        if missing:
            return _fail(
                f"ACTIVITY-FRAMES: claimed_frame_ids not in compiled inventory "
                f"missing={missing[:8]} inventory_size={len(inventory)} - "
                "refuse answers citing uncompiled/unknown frames",
                source_count=len(compiled),
            )

    # Re-fingerprint check: frame_id must match content (tamper / non-deterministic).
    mismatched: list[str] = []
    for fr in compiled:
        expected = activity_frame_fingerprint(
            application=fr.application,
            site=fr.site,
            t_start=fr.t_start,
            t_end=fr.t_end,
            evidence_ptrs=fr.evidence_ptrs,
            input_volume=fr.input_volume,
        )
        if fr.frame_id != expected:
            mismatched.append(fr.frame_id[:16])
    if mismatched:
        return _fail(
            f"ACTIVITY-FRAMES: {len(mismatched)} frame_id(s) do not match "
            f"deterministic fingerprint ids={mismatched[:8]} - "
            "frames must be byte-identical / cacheable (no model rewrite)",
            source_count=len(compiled),
        )

    return GateOutcome(
        ok=True,
        verdict="PASS",
        reason=(f"ACTIVITY-FRAMES ok: frames={len(compiled)} evidence_ok mode=compiled"),
        exit_code=0,
        source_count=len(compiled),
        human_required=False,
    )

assert_fresh(source, **kwargs)

Gate staleness and raise :class:ClosedLoopError unless outcome is ok.

Source code in src/foghorn/closed_loop.py

def assert_fresh(
    source: WorldRepo | str | Path,
    **kwargs: Any,
) -> GateOutcome:
    """Gate staleness and raise :class:`ClosedLoopError` unless outcome is ok."""
    outcome = gate_staleness(source, **kwargs)
    if not outcome.ok:
        raise ClosedLoopError(f"{outcome.verdict}: {outcome.reason}")
    return outcome

assert_not_current_state_store()

Explicit D-FOGHORN guard for integrators (always raises if called wrong).

Prefer calling :func:gate_staleness with mode='staleness' only.

Source code in src/foghorn/closed_loop.py

def assert_not_current_state_store() -> None:
    """Explicit D-FOGHORN guard for integrators (always raises if called wrong).

    Prefer calling :func:`gate_staleness` with ``mode='staleness'`` only.
    """
    raise ClosedLoopError(
        "D-FOGHORN: do not use foghorn as current-state/LWW store; "
        "list_facts() is chronological (oldest first), not latest-wins"
    )

assert_sources_fresh(source, **kwargs)

Raise :class:ClosedLoopError unless :func:gate_source_freshness is ok.

Source code in src/foghorn/closed_loop.py

def assert_sources_fresh(
    source: WorldRepo | Sequence[Fact],
    **kwargs: Any,
) -> GateOutcome:
    """Raise :class:`ClosedLoopError` unless :func:`gate_source_freshness` is ok."""
    outcome = gate_source_freshness(source, **kwargs)
    if not outcome.ok:
        raise ClosedLoopError(f"{outcome.verdict}: {outcome.reason}")
    return outcome

gate_source_freshness(source, *, max_age_seconds=DEFAULT_MAX_SOURCE_AGE_SECONDS, now=None, predicates=None, subjects=None, require_source_facts=True, use_latest_only=True)

Refuse decisions grounded on expired wiki/docs (Amazon Q stale-wiki class).

Public incident: Amazon Q / stale internal wiki - agents answer from retrieved documentation that is no longer current. gate_staleness only fires when fact ids change under a decision; it does not fail on wall-clock age of an unchanging wiki page fact.

Rules:

• No source facts when require_source_facts → FAIL_LOUD
• Any source fact with age > max_age_seconds → FAIL (human_required - re-retrieve or human review)
• Fresh sources only → PASS
• use_latest_only (default): apply D-FOGHORN - age the newest fact per (subject, predicate), not the oldest log row.

Parameters:

Name	Type	Description	Default
source	WorldRepo | Sequence[Fact]	WorldRepo or sequence of Facts.	required
max_age_seconds	float	Maximum allowed age (default 7 days).	DEFAULT_MAX_SOURCE_AGE_SECONDS
now	float | None	Reference time (default time.time()).	None
predicates	Iterable[str] | None	Override source predicate set (default wiki/doc set).	None
subjects	Iterable[str] | None	If set, only examine these subjects.	None
require_source_facts	bool	Empty source inventory → FAIL_LOUD.	True
use_latest_only	bool	Deduplicate to latest per key before aging.	True

Source code in src/foghorn/closed_loop.py

def gate_source_freshness(
    source: WorldRepo | Sequence[Fact],
    *,
    max_age_seconds: float = DEFAULT_MAX_SOURCE_AGE_SECONDS,
    now: float | None = None,
    predicates: Iterable[str] | None = None,
    subjects: Iterable[str] | None = None,
    require_source_facts: bool = True,
    use_latest_only: bool = True,
) -> GateOutcome:
    """Refuse decisions grounded on expired wiki/docs (Amazon Q stale-wiki class).

    Public incident: Amazon Q / stale internal wiki - agents answer from
    retrieved documentation that is no longer current. ``gate_staleness`` only
    fires when fact *ids* change under a decision; it does **not** fail on
    wall-clock age of an unchanging wiki page fact.

    Rules:

    * No source facts when ``require_source_facts`` → **FAIL_LOUD**
    * Any source fact with age > ``max_age_seconds`` → **FAIL**
      (``human_required`` - re-retrieve or human review)
    * Fresh sources only → **PASS**
    * ``use_latest_only`` (default): apply D-FOGHORN - age the newest fact per
      (subject, predicate), not the oldest log row.

    Args:
        source: WorldRepo or sequence of Facts.
        max_age_seconds: Maximum allowed age (default 7 days).
        now: Reference time (default ``time.time()``).
        predicates: Override source predicate set (default wiki/doc set).
        subjects: If set, only examine these subjects.
        require_source_facts: Empty source inventory → FAIL_LOUD.
        use_latest_only: Deduplicate to latest per key before aging.
    """
    if max_age_seconds < 0:
        return _fail_loud(
            "STALE-WIKI: max_age_seconds must be >= 0",
            human_required=True,
        )

    t_now = float(now if now is not None else time.time())
    all_facts = _facts_from_source(source)
    extra_preds = predicates
    subject_filter = {str(s).strip() for s in subjects if str(s).strip()} if subjects else None

    source_facts = [
        f
        for f in all_facts
        if is_source_predicate(f.predicate, extra=extra_preds)
        and (subject_filter is None or f.subject in subject_filter)
    ]

    if use_latest_only and source_facts:
        source_facts = _latest_by_subject_predicate(source_facts)

    if require_source_facts and len(source_facts) == 0:
        return _fail_loud(
            "STALE-WIKI/Amazon-Q: no wiki/doc source facts to gate "
            f"(predicates={sorted(DEFAULT_SOURCE_PREDICATES)[:6]}…) - "
            "cannot ground answers without a retrievable source inventory",
            human_required=True,
            source_count=0,
        )

    if not source_facts:
        return GateOutcome(
            ok=True,
            verdict="PASS",
            reason="STALE-WIKI: no source facts required; nothing to age-check",
            exit_code=0,
            source_count=0,
            human_required=False,
        )

    stale_ids: list[str] = []
    oldest_age = 0.0
    for f in source_facts:
        age = t_now - float(f.recorded_at)
        if age > oldest_age:
            oldest_age = age
        if age > max_age_seconds:
            stale_ids.append(f.id)

    if stale_ids:
        return _fail(
            f"STALE-WIKI/Amazon-Q: {len(stale_ids)} source fact(s) older than "
            f"max_age={max_age_seconds:.0f}s (oldest_age={oldest_age:.0f}s) "
            f"ids={stale_ids[:8]} - refuse answer grounded on expired wiki/docs; "
            f"re-retrieve before decision",
            human_required=True,
            source_count=len(source_facts),
            oldest_age_seconds=oldest_age,
            stale_source_ids=tuple(stale_ids[:20]),
        )

    return GateOutcome(
        ok=True,
        verdict="PASS",
        reason=(
            f"STALE-WIKI ok: sources={len(source_facts)} "
            f"oldest_age={oldest_age:.0f}s max_age={max_age_seconds:.0f}s"
        ),
        exit_code=0,
        source_count=len(source_facts),
        oldest_age_seconds=oldest_age,
        stale_source_ids=(),
        human_required=False,
    )

gate_staleness(source, *, mode='staleness', impact_threshold=0.5, require_decisions=True)

Read a world, surface staleness, fail loudly on empty or D-FOGHORN misuse.

Parameters:

Name	Type	Description	Default
source	WorldRepo | str | Path	Open :class:WorldRepo or path to a world SQLite db.	required
mode	Mode	Must be "staleness". "current_state" is always FAIL_LOUD (D-FOGHORN - foghorn is not a LWW episode store).	'staleness'
impact_threshold	float	Max alert impact allowed for PASS (any alert at/above this impact → FAIL exit 1).	0.5
require_decisions	bool	If True, a world with zero decisions is FAIL_LOUD (nothing load-bearing to gate).	True

Returns:

Type	Description
GateOutcome	class:GateOutcome - callers should sys.exit(outcome.exit_code).

Source code in src/foghorn/closed_loop.py

def gate_staleness(
    source: WorldRepo | str | Path,
    *,
    mode: Mode = "staleness",
    impact_threshold: float = 0.5,
    require_decisions: bool = True,
) -> GateOutcome:
    """Read a world, surface staleness, fail loudly on empty or D-FOGHORN misuse.

    Args:
        source: Open :class:`WorldRepo` or path to a world SQLite db.
        mode: Must be ``\"staleness\"``. ``\"current_state\"`` is always FAIL_LOUD
            (D-FOGHORN - foghorn is not a LWW episode store).
        impact_threshold: Max alert impact allowed for PASS (any alert at/above
            this impact → FAIL exit 1).
        require_decisions: If True, a world with zero decisions is FAIL_LOUD
            (nothing load-bearing to gate).

    Returns:
        :class:`GateOutcome` - callers should ``sys.exit(outcome.exit_code)``.
    """
    if mode == "current_state":
        return _fail_loud(
            "D-FOGHORN: mode=current_state forbidden - foghorn is fact→decision "
            "staleness only, never LWW episode/current-state store"
        )
    if mode != "staleness":
        return _fail_loud(f"unknown mode={mode!r} - only mode='staleness' is valid")

    owns = False
    repo: WorldRepo | None = None
    try:
        try:
            repo, owns = _open_repo(source)
        except ClosedLoopError as exc:
            return _fail_loud(str(exc))
        except Exception as exc:
            return _fail_loud(f"open world failed: {exc.__class__.__name__}: {exc}")

        facts = list(repo.store.list_facts())
        decisions = (
            list(repo.store.list_decisions()) if hasattr(repo.store, "list_decisions") else []
        )
        # Fallback: decisions may only live in commits - try common APIs
        if not decisions and hasattr(repo.store, "all_decisions"):
            decisions = list(repo.store.all_decisions())

        if require_decisions and len(decisions) == 0:
            return _fail_loud(
                "empty decisions - no load-bearing fact→decision edges to gate "
                "(write-only fact log is ornament)"
            )

        if len(facts) == 0 and len(decisions) == 0:
            return _fail_loud("empty world - nothing to gate")

        try:
            alerts = tuple(repo.stale())
        except Exception as exc:
            return _fail_loud(f"stale() failed: {exc.__class__.__name__}: {exc}")

        max_impact = max((a.impact_score for a in alerts), default=0.0)
        hot = [a for a in alerts if a.impact_score >= impact_threshold]

        if hot:
            return GateOutcome(
                ok=False,
                verdict="FAIL",
                reason=(
                    f"stale decisions impact>={impact_threshold}: "
                    f"count={len(hot)} max_impact={max_impact:.3f}"
                ),
                exit_code=1,
                alerts=alerts,
                max_impact=max_impact,
            )

        return GateOutcome(
            ok=True,
            verdict="PASS",
            reason=(
                f"no high-impact staleness (alerts={len(alerts)} "
                f"max_impact={max_impact:.3f} threshold={impact_threshold})"
            ),
            exit_code=0,
            alerts=alerts,
            max_impact=max_impact,
        )
    finally:
        if owns and repo is not None:
            with contextlib.suppress(Exception):
                repo.close()

is_source_predicate(predicate, *, extra=None)

True if predicate marks wiki/doc/retrieval grounding.

Source code in src/foghorn/closed_loop.py

def is_source_predicate(
    predicate: str,
    *,
    extra: Iterable[str] | None = None,
) -> bool:
    """True if *predicate* marks wiki/doc/retrieval grounding."""
    p = (predicate or "").strip().lower().replace("-", "_").replace(" ", "_")
    if not p:
        return False
    banned = set(DEFAULT_SOURCE_PREDICATES)
    if extra:
        banned |= {str(x).strip().lower().replace("-", "_") for x in extra}
    if p in banned:
        return True
    return any(p.startswith(b + "_") or p.endswith("_" + b) for b in banned)

analyze_evidence_links(features=None, evidence_links=None, *, known_source_ids=None, min_links_per_feature=1)

Summarize feature provenance coverage (does not gate).

Returns covered/unlinked feature ids, broken source pointers, and rubric failures. Use :func:gate_evidence_links to refuse decision-grade use.

Source code in src/foghorn/evidence.py

def analyze_evidence_links(
    features: Sequence[Any] | None = None,
    evidence_links: Sequence[Any] | None = None,
    *,
    known_source_ids: Sequence[str] | None = None,
    min_links_per_feature: int = 1,
) -> dict[str, Any]:
    """Summarize feature provenance coverage (does not gate).

    Returns covered/unlinked feature ids, broken source pointers, and rubric
    failures. Use :func:`gate_evidence_links` to refuse decision-grade use.
    """
    feats = [_as_feature(f) for f in (features or [])]
    links: list[EvidenceLink] = []
    for raw in evidence_links or []:
        links.append(_as_link(raw))

    known = {str(s).strip() for s in (known_source_ids or []) if str(s).strip()}
    by_feature: dict[str, list[EvidenceLink]] = {f.feature_id: [] for f in feats}
    for link in links:
        by_feature.setdefault(link.feature_id, []).append(link)

    unlinked: list[str] = []
    underlinked: list[str] = []
    for f in feats:
        n = len(by_feature.get(f.feature_id) or [])
        if n == 0:
            unlinked.append(f.feature_id)
        elif n < min_links_per_feature:
            underlinked.append(f.feature_id)

    broken: list[dict[str, str]] = []
    empty_source: list[dict[str, str]] = []
    for link in links:
        if not link.source_id:
            empty_source.append(link.to_dict())
            continue
        if known and link.source_id not in known:
            broken.append(
                {
                    "feature_id": link.feature_id,
                    "source_id": link.source_id,
                    "kind": "missing_source",
                }
            )

    rubric_fail = [f.feature_id for f in feats if not f.rubric_ok]

    return {
        "feature_count": len(feats),
        "evidence_link_count": len(links),
        "feature_ids": [f.feature_id for f in feats],
        "unlinked_feature_ids": unlinked,
        "underlinked_feature_ids": underlinked,
        "broken_source_ptrs": broken,
        "empty_source_ptrs": empty_source,
        "rubric_fail_ids": rubric_fail,
        "links_by_feature": {
            fid: [lk.to_dict() for lk in lks] for fid, lks in by_feature.items()
        },
        "fully_linked": (
            len(feats) > 0
            and len(unlinked) == 0
            and len(underlinked) == 0
            and len(broken) == 0
            and len(empty_source) == 0
            and len(rubric_fail) == 0
        ),
    }

assert_evidence_linked(features=None, evidence_links=None, **kwargs)

Raise :class:ClosedLoopError unless :func:gate_evidence_links is ok.

Source code in src/foghorn/evidence.py

def assert_evidence_linked(
    features: Sequence[Any] | None = None,
    evidence_links: Sequence[Any] | None = None,
    **kwargs: Any,
) -> GateOutcome:
    """Raise :class:`ClosedLoopError` unless :func:`gate_evidence_links` is ok."""
    outcome = gate_evidence_links(features, evidence_links, **kwargs)
    if not outcome.ok:
        raise ClosedLoopError(f"{outcome.verdict}: {outcome.reason}")
    return outcome

gate_evidence_links(features=None, evidence_links=None, *, known_source_ids=None, claim_decision_grade=False, require_features=True, min_links_per_feature=1, refuse_rubric_fail=True, refuse_broken_sources=True)

Refuse decision-grade use of features without evidence provenance.

Public case: arXiv 2608.06366 Tracing the Heart: An Evidence-Linked Pipeline for Heart-Failure Feature Engineering. Derived features without structural integrity, rubric compliance, and provenance are not decision-grade. Foghorn already ages wiki sources and requires activity evidence_ptrs; this gate covers feature engineering provenance.

Rules:

1. claim_decision_grade with zero features when required → FAIL_LOUD
2. Empty features + empty links (non-claim, require_features) → FAIL_LOUD
3. Any feature with fewer than min_links_per_feature links → FAIL (or FAIL_LOUD when claim_decision_grade and zero links total)
4. Evidence pointer with empty source_id → FAIL
5. source_id not in known_source_ids (when provided) → FAIL
6. rubric_ok=False when refuse_rubric_fail → FAIL
7. Fully linked, rubric-ok features → PASS

Parameters:

Name	Type	Description	Default
features	Sequence[Any] | None	Feature records (dicts or :class:FeatureRecord).	None
evidence_links	Sequence[Any] | None	Provenance edges (dicts or :class:EvidenceLink).	None
known_source_ids	Sequence[str] | None	Optional inventory of valid source artifact ids.	None
claim_decision_grade	bool	Features claimed ready for clinical/decision use.	False
require_features	bool	Empty feature inventory → FAIL_LOUD when claiming or when True and no links either.	True
min_links_per_feature	int	Minimum evidence edges per feature (default 1).	1
refuse_rubric_fail	bool	Rubric non-compliance → FAIL.	True
refuse_broken_sources	bool	Unknown source_id → FAIL when inventory given.	True

Source code in src/foghorn/evidence.py

def gate_evidence_links(
    features: Sequence[Any] | None = None,
    evidence_links: Sequence[Any] | None = None,
    *,
    known_source_ids: Sequence[str] | None = None,
    claim_decision_grade: bool = False,
    require_features: bool = True,
    min_links_per_feature: int = 1,
    refuse_rubric_fail: bool = True,
    refuse_broken_sources: bool = True,
) -> GateOutcome:
    """Refuse decision-grade use of features without evidence provenance.

    Public case: arXiv 2608.06366 *Tracing the Heart: An Evidence-Linked
    Pipeline for Heart-Failure Feature Engineering*. Derived features without
    structural integrity, rubric compliance, and provenance are not
    decision-grade. Foghorn already ages wiki sources and requires activity
    evidence_ptrs; this gate covers **feature engineering** provenance.

    Rules:

    1. ``claim_decision_grade`` with zero features when required → **FAIL_LOUD**
    2. Empty features + empty links (non-claim, require_features) → **FAIL_LOUD**
    3. Any feature with fewer than ``min_links_per_feature`` links → **FAIL**
       (or **FAIL_LOUD** when claim_decision_grade and zero links total)
    4. Evidence pointer with empty ``source_id`` → **FAIL**
    5. ``source_id`` not in ``known_source_ids`` (when provided) → **FAIL**
    6. ``rubric_ok=False`` when ``refuse_rubric_fail`` → **FAIL**
    7. Fully linked, rubric-ok features → **PASS**

    Args:
        features: Feature records (dicts or :class:`FeatureRecord`).
        evidence_links: Provenance edges (dicts or :class:`EvidenceLink`).
        known_source_ids: Optional inventory of valid source artifact ids.
        claim_decision_grade: Features claimed ready for clinical/decision use.
        require_features: Empty feature inventory → FAIL_LOUD when claiming
            or when True and no links either.
        min_links_per_feature: Minimum evidence edges per feature (default 1).
        refuse_rubric_fail: Rubric non-compliance → FAIL.
        refuse_broken_sources: Unknown source_id → FAIL when inventory given.
    """
    try:
        summary = analyze_evidence_links(
            features,
            evidence_links,
            known_source_ids=known_source_ids,
            min_links_per_feature=min_links_per_feature,
        )
    except (TypeError, ValueError) as exc:
        return _fail_loud(
            f"EVIDENCE-LINK: invalid feature/link payload: {exc}",
            source_count=0,
        )

    n_feat = int(summary["feature_count"])
    n_links = int(summary["evidence_link_count"])
    unlinked = tuple(summary["unlinked_feature_ids"])
    under = tuple(summary["underlinked_feature_ids"])
    broken = tuple(summary["broken_source_ptrs"])
    empty_src = tuple(summary["empty_source_ptrs"])
    rubric_fail = tuple(summary["rubric_fail_ids"])

    if claim_decision_grade and require_features and n_feat == 0:
        return _fail_loud(
            "EVIDENCE-LINK: claim_decision_grade with zero features - "
            "phantom evidence-linked pipeline (arXiv 2608.06366); refuse "
            "decision-grade use without engineered feature inventory",
            source_count=0,
        )

    if require_features and n_feat == 0 and n_links == 0:
        return _fail_loud(
            "EVIDENCE-LINK: empty features and empty evidence links - "
            "nothing to ground; compile provenance before use",
            source_count=0,
        )

    if claim_decision_grade and n_feat > 0 and n_links == 0:
        return _fail_loud(
            f"EVIDENCE-LINK: {n_feat} decision-grade feature(s) with zero "
            f"evidence links - refuse unprovenanced feature engineering "
            f"(arXiv 2608.06366 Tracing the Heart)",
            source_count=n_feat,
            stale_source_ids=unlinked,
        )

    if unlinked or under:
        bad = list(unlinked) + [u for u in under if u not in unlinked]
        return _fail(
            f"EVIDENCE-LINK: {len(bad)} feature(s) lack required provenance "
            f"(min_links={min_links_per_feature}): {bad[:8]}"
            + ("…" if len(bad) > 8 else ""),
            source_count=n_feat,
            stale_source_ids=tuple(bad),
        )

    if empty_src:
        return _fail(
            f"EVIDENCE-LINK: {len(empty_src)} evidence pointer(s) with empty "
            f"source_id - refuse broken provenance",
            source_count=n_feat,
        )

    if refuse_broken_sources and broken:
        ids = [b.get("source_id", "") for b in broken]
        return _fail(
            f"EVIDENCE-LINK: {len(broken)} evidence pointer(s) reference "
            f"unknown sources {ids[:8]} - refuse dangling provenance",
            source_count=n_feat,
            stale_source_ids=tuple(str(x) for x in ids if x),
        )

    if refuse_rubric_fail and rubric_fail:
        return _fail(
            f"EVIDENCE-LINK: {len(rubric_fail)} feature(s) failed rubric/"
            f"structure checks: {list(rubric_fail)[:8]} - refuse "
            f"non-compliant engineered features",
            source_count=n_feat,
            stale_source_ids=rubric_fail,
        )

    return GateOutcome(
        ok=True,
        verdict="PASS",
        reason=(
            f"EVIDENCE-LINK ok: features={n_feat} links={n_links} "
            f"claim_decision_grade={claim_decision_grade}"
        ),
        exit_code=0,
        source_count=n_feat,
        human_required=False,
        stale_source_ids=(),
    )

is_evidence_kind(kind, *, extra=None)

True if kind marks a provenance/evidence edge.

Source code in src/foghorn/evidence.py

def is_evidence_kind(kind: str, *, extra: Iterable[str] | None = None) -> bool:
    """True if *kind* marks a provenance/evidence edge."""
    k = _canon(kind)
    if not k:
        return False
    allowed = set(DEFAULT_EVIDENCE_KINDS)
    if extra:
        allowed |= {_canon(x) for x in extra}
    return k in allowed or "evidence" in k or k.startswith("source_")

is_feature_kind(kind, *, extra=None)

True if kind marks an engineered/derived feature.

Source code in src/foghorn/evidence.py

def is_feature_kind(kind: str, *, extra: Iterable[str] | None = None) -> bool:
    """True if *kind* marks an engineered/derived feature."""
    k = _canon(kind)
    if not k:
        return False
    allowed = set(DEFAULT_FEATURE_KINDS)
    if extra:
        allowed |= {_canon(x) for x in extra}
    return k in allowed or k.endswith("_feature") or k.startswith("feature_")

export_graphviz(repo)

Export the fact → decision dependency graph in Graphviz DOT format.

Each fact and decision is a node; directed edges run from facts to the decisions that depend on them. The resulting DOT string can be rendered with dot -Tsvg graph.dot > graph.svg.

Parameters:

Name	Type	Description	Default
repo	WorldRepo	The repository to graph.	required

Returns:

Type	Description
str	A Graphviz DOT string.

Source code in src/foghorn/export.py

def export_graphviz(repo: WorldRepo) -> str:
    """Export the fact → decision dependency graph in Graphviz DOT format.

    Each fact and decision is a node; directed edges run from facts to the
    decisions that depend on them. The resulting DOT string can be rendered
    with ``dot -Tsvg graph.dot > graph.svg``.

    Args:
        repo: The repository to graph.

    Returns:
        A Graphviz DOT string.
    """
    store = repo.store
    facts = store.list_facts()
    decisions = store.list_decisions()

    lines: list[str] = [
        "digraph foghorn {",
        "  rankdir=LR;",
        '  node [fontname="Helvetica", fontsize=10];',
        "",
        "  // Facts",
    ]

    for fact in facts:
        label = f"{fact.subject}\\n{fact.predicate}\\n{fact.object}"
        label = label.replace('"', '\\"')
        lines.append(
            f'  "fact_{fact.id}" [shape=ellipse, style=filled, fillcolor="#d0e8ff",'
            f' label="{label}"];'
        )

    lines.append("")
    lines.append("  // Decisions")

    for dec in decisions:
        label = dec.label.replace('"', '\\"')
        lines.append(
            f'  "dec_{dec.id}" [shape=box, style=filled, fillcolor="#ffe0b0", label="{label}"];'
        )

    lines.append("")
    lines.append("  // Edges (fact -> decision)")

    for dec in decisions:
        for fid in dec.fact_ids:
            lines.append(f'  "fact_{fid}" -> "dec_{dec.id}";')

    lines.append("}")
    return "\n".join(lines)

export_json(repo)

Export the entire repository state as a JSON string.

Exports all facts, decisions, and commits currently known to the store. The resulting JSON is suitable for archiving, migration, or seeding a fresh repository via :func:import_json.

Parameters:

Name	Type	Description	Default
repo	WorldRepo	The repository to export.	required

Returns:

Type	Description
str	A JSON string with keys "facts", "decisions", and "commits".

Source code in src/foghorn/export.py

def export_json(repo: WorldRepo) -> str:
    """Export the entire repository state as a JSON string.

    Exports all facts, decisions, and commits currently known to the store.
    The resulting JSON is suitable for archiving, migration, or seeding a
    fresh repository via :func:`import_json`.

    Args:
        repo: The repository to export.

    Returns:
        A JSON string with keys ``"facts"``, ``"decisions"``, and ``"commits"``.
    """
    store = repo.store

    facts = [f.to_dict() for f in store.list_facts()]
    decisions = [d.to_dict() for d in store.list_decisions()]
    commits = [c.to_dict() for c in store.log()]

    return json.dumps(
        {
            "foghorn_export_version": 1,
            "facts": facts,
            "decisions": decisions,
            "commits": commits,
        },
        indent=2,
        sort_keys=True,
    )

import_json(path_or_str, target_repo)

Import a JSON export into a target repository.

Imports all facts and decisions from the export. Each unique fact and decision is staged and committed to the target repository as a single "import" commit. If there is nothing new to stage, no commit is created.

Parameters:

Name	Type	Description	Default
path_or_str	str	Either a JSON string or a path to a JSON file.	required
target_repo	WorldRepo	The :class:~foghorn.repo.WorldRepo to import into.	required

Returns:

Type	Description
int	The total number of items (facts + decisions) imported.

Raises:

Type	Description
ValueError	If the JSON is not a valid foghorn export.
FileNotFoundError	If a path is given but does not exist.

Source code in src/foghorn/export.py

def import_json(path_or_str: str, target_repo: WorldRepo) -> int:
    """Import a JSON export into a target repository.

    Imports all facts and decisions from the export. Each unique fact and
    decision is staged and committed to the target repository as a single
    "import" commit.  If there is nothing new to stage, no commit is created.

    Args:
        path_or_str: Either a JSON string or a path to a JSON file.
        target_repo: The :class:`~foghorn.repo.WorldRepo` to import into.

    Returns:
        The total number of items (facts + decisions) imported.

    Raises:
        ValueError: If the JSON is not a valid foghorn export.
        FileNotFoundError: If a path is given but does not exist.
    """
    # Detect whether path_or_str is a file path or raw JSON.
    # Guard against very long strings that would cause OSError on Path.exists().
    raw: str
    if len(path_or_str) < 4096:
        p = Path(path_or_str)
        try:
            is_file = p.exists() and p.is_file()
        except OSError:
            is_file = False
        raw = p.read_text(encoding="utf-8") if is_file else path_or_str
    else:
        raw = path_or_str

    try:
        data = json.loads(raw)
    except json.JSONDecodeError as exc:
        raise ValueError(f"Invalid JSON: {exc}") from exc

    if not isinstance(data, dict):
        raise ValueError("Export JSON must be a top-level object.")

    version = data.get("foghorn_export_version", 0)
    if version != 1:
        raise ValueError(f"Unsupported export version: {version} (expected 1)")

    from foghorn.fact import Decision, Fact

    store = target_repo.store
    imported = 0

    for fact_dict in data.get("facts", []):
        fact = Fact.from_dict(fact_dict)
        # add_fact is idempotent (INSERT OR IGNORE)
        store.add_fact(fact)
        imported += 1

    for dec_dict in data.get("decisions", []):
        decision = Decision.from_dict(dec_dict)
        store.add_decision(decision)
        imported += 1

    if imported > 0 and store.staged_count() > 0:
        store.commit(f"Import: {imported} items from foghorn export")

    return imported

propagate_staleness(repo, changed_fact_ids)

Find all directly and transitively stale decisions for a set of changed facts.

The propagation graph is fact → decision. Decisions that share a common "fact" edge with other decisions (i.e., decisions whose ID appears in another decision's fact_ids list) are considered transitive dependents.

Parameters:

Name	Type	Description	Default
repo	WorldRepo	The :class:~foghorn.repo.WorldRepo to query.	required
changed_fact_ids	list[str]	IDs of facts that have changed.	required

Returns:

Name	Type	Description
A	PropagationResult	class:PropagationResult with direct and transitive stale decision labels.

Source code in src/foghorn/propagate.py

def propagate_staleness(repo: WorldRepo, changed_fact_ids: list[str]) -> PropagationResult:
    """Find all directly and transitively stale decisions for a set of changed facts.

    The propagation graph is fact → decision. Decisions that share a common "fact"
    edge with other decisions (i.e., decisions whose ID appears in another decision's
    ``fact_ids`` list) are considered transitive dependents.

    Args:
        repo: The :class:`~foghorn.repo.WorldRepo` to query.
        changed_fact_ids: IDs of facts that have changed.

    Returns:
        A :class:`PropagationResult` with direct and transitive stale decision labels.
    """
    if not changed_fact_ids:
        return PropagationResult(
            changed_fact_ids=[],
            impact_summary="No facts changed - nothing is stale.",
        )

    store = repo.store
    all_decisions = store.list_decisions()

    # Build a set of directly stale decision IDs (depend on a changed fact)
    directly_stale_ids: set[str] = set()
    directly_stale_labels: list[str] = []

    for decision in all_decisions:
        if (
            any(fid in changed_fact_ids for fid in decision.fact_ids)
            and decision.id not in directly_stale_ids
        ):
            directly_stale_ids.add(decision.id)
            directly_stale_labels.append(decision.label)

    # Build transitive closure: decisions whose fact_ids reference stale decision IDs
    # (some workflows record decision dependencies as fact IDs for cross-linking)
    transitively_stale_ids: set[str] = set()
    transitively_stale_labels: list[str] = []
    depth = 1 if directly_stale_ids else 0

    # BFS over decision→decision edges where a fact_id is a decision ID
    frontier = set(directly_stale_ids)
    visited: set[str] = set(directly_stale_ids)
    current_depth = 1

    while frontier:
        next_frontier: set[str] = set()
        for decision in all_decisions:
            if decision.id in visited:
                continue
            # Decision references a stale decision via its fact_ids
            if any(fid in frontier for fid in decision.fact_ids):
                transitively_stale_ids.add(decision.id)
                transitively_stale_labels.append(decision.label)
                next_frontier.add(decision.id)
        if not next_frontier:
            break
        visited |= next_frontier
        frontier = next_frontier
        current_depth += 1
        if current_depth > depth:
            depth = current_depth

    # Build human-readable summary
    n_direct = len(directly_stale_labels)
    n_transitive = len(transitively_stale_labels)
    n_facts = len(changed_fact_ids)

    if n_direct == 0:
        summary = f"{n_facts} fact(s) changed but no decisions depend on them directly."
    elif n_transitive == 0:
        summary = (
            f"{n_facts} fact(s) changed → {n_direct} decision(s) directly stale "
            f"(no transitive dependencies found)."
        )
    else:
        summary = (
            f"{n_facts} fact(s) changed → {n_direct} decision(s) directly stale, "
            f"{n_transitive} decision(s) transitively stale (depth {depth})."
        )

    return PropagationResult(
        changed_fact_ids=list(changed_fact_ids),
        directly_stale=directly_stale_labels,
        transitively_stale=transitively_stale_labels,
        propagation_depth=depth,
        impact_summary=summary,
    )

compute_staleness(store, changed_fact_ids)

Given a set of changed fact IDs, return staleness alerts for affected decisions.

For each Decision that depends on at least one changed fact, emit a StalenessAlert. The impact_score is the average confidence of the changed facts that the decision depended on.

Parameters:

Name	Type	Description	Default
store	WorldStore	The WorldStore to resolve Decisions from.	required
changed_fact_ids	set[str]	Set of Fact IDs that have been added or removed.	required

Returns:

Type	Description
list[StalenessAlert]	List of StalenessAlert, sorted by impact_score descending.

Source code in src/foghorn/staleness.py

def compute_staleness(
    store: WorldStore,
    changed_fact_ids: set[str],
) -> list[StalenessAlert]:
    """Given a set of changed fact IDs, return staleness alerts for affected decisions.

    For each Decision that depends on at least one changed fact, emit a
    StalenessAlert. The ``impact_score`` is the average confidence of the
    changed facts that the decision depended on.

    Args:
        store: The WorldStore to resolve Decisions from.
        changed_fact_ids: Set of Fact IDs that have been added or removed.

    Returns:
        List of StalenessAlert, sorted by impact_score descending.
    """
    if not changed_fact_ids:
        return []

    seen_decision_ids: set[str] = set()
    alerts: list[StalenessAlert] = []

    for fact_id in changed_fact_ids:
        decisions = store.get_decisions_for_fact(fact_id)
        for decision in decisions:
            if decision.id in seen_decision_ids:
                continue
            seen_decision_ids.add(decision.id)

            stale_ids = [fid for fid in decision.fact_ids if fid in changed_fact_ids]
            if not stale_ids:
                continue

            # Impact = mean confidence of the changed facts this decision relied on
            confidences: list[float] = []
            for fid in stale_ids:
                fact = store.get_fact(fid)
                if fact is not None:
                    confidences.append(fact.confidence)
            impact = sum(confidences) / len(confidences) if confidences else 0.5

            alerts.append(
                StalenessAlert(
                    decision_id=decision.id,
                    decision_label=decision.label,
                    stale_fact_ids=stale_ids,
                    impact_score=impact,
                )
            )

    alerts.sort(key=lambda a: a.impact_score, reverse=True)
    return alerts

diff_commits(store, commit_a, commit_b)

Compute the fact-level diff between two commits.

Parameters:

Name	Type	Description	Default
store	WorldStore	The WorldStore to resolve Fact objects from.	required
commit_a	WorldCommit | None	The base commit (None = empty state).	required
commit_b	WorldCommit	The head commit to compare against.	required

Returns:

Type	Description
DiffResult	DiffResult with added/removed facts and the union of changed IDs.

Source code in src/foghorn/staleness.py

def diff_commits(
    store: WorldStore,
    commit_a: WorldCommit | None,
    commit_b: WorldCommit,
) -> DiffResult:
    """Compute the fact-level diff between two commits.

    Args:
        store: The WorldStore to resolve Fact objects from.
        commit_a: The base commit (None = empty state).
        commit_b: The head commit to compare against.

    Returns:
        DiffResult with added/removed facts and the union of changed IDs.
    """
    ids_a = commit_a.fact_ids if commit_a else set()
    ids_b = commit_b.fact_ids

    added_ids = ids_b - ids_a
    removed_ids = ids_a - ids_b

    added_facts = [f for fid in added_ids if (f := store.get_fact(fid)) is not None]
    removed_facts = [f for fid in removed_ids if (f := store.get_fact(fid)) is not None]

    added_facts.sort(key=lambda f: f.recorded_at)
    removed_facts.sort(key=lambda f: f.recorded_at)

    return DiffResult(
        added_facts=added_facts,
        removed_facts=removed_facts,
        changed_fact_ids=added_ids | removed_ids,
        commit_a_id=commit_a.id if commit_a else None,
        commit_b_id=commit_b.id,
    )