Python API Reference

Top-level exports

import clickproof

clickproof

clickproof - persistent GUI behavioral facts for computer-use agents.

DecayProjection(fact_id, element, current_score, score_in_7_days, score_in_30_days, days_until_threshold, recommendation) dataclass

Decay projection for a single UIFact.

Attributes:

Name	Type	Description
fact_id	str	ID of the UIFact.
element	str	Semantic element description.
current_score	float	Score right now.
score_in_7_days	float	Projected score 7 days from now.
score_in_30_days	float	Projected score 30 days from now.
days_until_threshold	float	Days until score drops below min_score (0.0 if already below threshold or if it can never reach threshold).
recommendation	str	One of "ok", "re-validate", or "archive".

ClickAttempt(fact_id, target_element, hit, force_used=False, overlay_intercepted=False, observed_effect=True, agent_run_id='', notes='') dataclass

One computer-use click against a stored UIFact.

Farm OVERLAY-CLICK: Playwright force=True can hit an overlay (e.g. X #layers) and never throw - the agent thinks it clicked the target. Callers must report whether the intended element was hit.

is_miss property

True when the intended target did not receive a real click.

ClickOutcomeResult(ok, invalidated, miss_kind, score_before, score_after, confidence_after, observation_confirmed, fact_id, reason) dataclass

Result of recording a click attempt against a fact.

ClosedLoopError

Bases: ValueError

Raised when the gate refuses empty or unusable fact stores.

GateOutcome(ok, verdict, reason, exit_code, fact_count=0, usable_count=0, stale_count=0, min_score_seen=None, human_required=False, action=None, task=None, risk=None) dataclass

Result of a closed-loop read of a clickproof fact store or task gate.

Attributes:

Name	Type	Description
ok	bool	True only when a pipeline may continue (PASS).
verdict	str	PASS, FAIL, or FAIL_LOUD.
reason	str	Human-readable explanation (always non-empty).
exit_code	int	0 PASS, 1 FAIL (stale/low-confidence), 2 FAIL_LOUD (empty).
fact_count	int	Number of facts examined.
usable_count	int	Facts with score >= min_score.
stale_count	int	Facts with score < min_score.
min_score_seen	float | None	Lowest score among facts (None if empty).
human_required	bool	True when adversarial/out-of-scope needs human review.
action	str | None	Proposed action when task-alignment gated.
task	str | None	Declared task when task-alignment gated.
risk	str | None	safe / high_risk when classified.

to_dict()

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

Source code in clickproof/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,
        "fact_count": self.fact_count,
        "usable_count": self.usable_count,
        "stale_count": self.stale_count,
        "min_score_seen": self.min_score_seen,
        "human_required": self.human_required,
        "action": self.action,
        "task": self.task,
        "risk": self.risk,
    }

SessionMemory(session_id, app_name, app_version, loaded_fact_ids, bootstrap_text, loaded_at, usable_count, min_score) dataclass

Facts loaded for one computer-use agent session.

GUI-MEMORY: sessions that skip load while the store already holds usable facts for the app re-discover the UI every run - the farm failure mode.

ContextDecision(decision_id, choice, sparse_context=dict(), dominant_cues=dict(), attended_keys=(), choice_by_context=dict()) dataclass

One multimodal / GUI decision with context inventory.

Attributes:

Name	Type	Description
decision_id	str	Stable id for the choice.
choice	str	Selected option label (e.g. creative id, button path).
sparse_context	dict[str, Any]	Sparse decision-critical vars (market, locale, …).
dominant_cues	dict[str, Any]	High-volume product/visual signals.
attended_keys	tuple[str, ...]	Context keys the model claims to have used.
choice_by_context	Mapping[str, str]	Optional map context fingerprint → choice (for cross-context collapse detection across a batch).

CVEReport(sparse_keys_present, sparse_keys_missing, attended_sparse, ignored_sparse, dominant_only, cross_context_collapse, collapsed_contexts) dataclass

Analysis of contextual variable overestimation risk.

FactObservation(fact_id, observed_at, confirmed, agent_run_id='') dataclass

An observation that confirms or refutes a UIFact.

Attributes:

Name	Type	Description
fact_id	str	ID of the UIFact this observation pertains to.
observed_at	float	Unix timestamp when this observation was made.
confirmed	bool	True = fact still holds; False = fact no longer holds.
agent_run_id	str	Optional tracing identifier.
id	str	Content-addressed identifier - SHA-256[:16] of "{fact_id}|{observed_at}|{confirmed}".

to_dict()

Serialize to a JSON-compatible dict.

Source code in clickproof/fact.py

def to_dict(self) -> dict[str, Any]:
    """Serialize to a JSON-compatible dict."""
    return {
        "id": self.id,
        "fact_id": self.fact_id,
        "observed_at": self.observed_at,
        "confirmed": self.confirmed,
        "agent_run_id": self.agent_run_id,
    }

from_dict(d) classmethod

Deserialize from a dict produced by to_dict().

Source code in clickproof/fact.py

@classmethod
def from_dict(cls, d: dict[str, Any]) -> FactObservation:
    """Deserialize from a dict produced by to_dict()."""
    obs = cls(
        fact_id=d["fact_id"],
        observed_at=d["observed_at"],
        confirmed=d["confirmed"],
        agent_run_id=d.get("agent_run_id", ""),
    )
    return obs

UIFact(app_name, app_version, element, action, outcome, context='', confidence=1.0, recorded_at=time.time()) dataclass

A behavioral fact about a UI element in a specific app version.

UIFacts are the atoms of clickproof. Two UIFacts with the same app_name, app_version, element, and action always have the same ID.

Attributes:

Name	Type	Description
app_name	str	Application identifier, e.g. "salesforce", "gmail".
app_version	str	Version string, e.g. "2025.11", "unknown".
element	str	Semantic element description, e.g. "export-csv-button".
action	str	What to do: "click", "type", "navigate".
outcome	str	What happens: "opens-download-dialog", "error:not-found".
context	str	Optional UI context, e.g. "reports-page".
confidence	float	Initial confidence in [0.0, 1.0]. Default 1.0.
recorded_at	float	Unix timestamp when this fact was recorded.
id	str	Content-addressed identifier - SHA-256[:16] of "{app_name}|{app_version}|{element}|{action}".

to_dict()

Serialize to a JSON-compatible dict.

Source code in clickproof/fact.py

def to_dict(self) -> dict[str, Any]:
    """Serialize to a JSON-compatible dict."""
    return {
        "id": self.id,
        "app_name": self.app_name,
        "app_version": self.app_version,
        "element": self.element,
        "action": self.action,
        "outcome": self.outcome,
        "context": self.context,
        "confidence": self.confidence,
        "recorded_at": self.recorded_at,
    }

from_dict(d) classmethod

Deserialize from a dict produced by to_dict().

Source code in clickproof/fact.py

@classmethod
def from_dict(cls, d: dict[str, Any]) -> UIFact:
    """Deserialize from a dict produced by to_dict()."""
    fact = cls(
        app_name=d["app_name"],
        app_version=d["app_version"],
        element=d["element"],
        action=d["action"],
        outcome=d["outcome"],
        context=d.get("context", ""),
        confidence=d.get("confidence", 1.0),
        recorded_at=d.get("recorded_at", 0.0),
    )
    return fact

FactRetriever(store, scorer=None)

Retrieves and ranks relevant UIFacts for an agent session start.

Parameters:

Name	Type	Description	Default
store	FactStore	The FactStore to query.	required
scorer	FactScorer | None	Optional FactScorer; a default one is created if not provided.	None

Source code in clickproof/retriever.py

def __init__(self, store: FactStore, scorer: FactScorer | None = None) -> None:
    self._store = store
    self._scorer = scorer or FactScorer()

query(app_name, app_version=None, element=None, min_score=0.5)

Return (fact, score) pairs sorted by score descending.

Parameters:

Name	Type	Description	Default
app_name	str	Required - filter by application name.	required
app_version	str | None	Optional - filter to a specific version.	None
element	str | None	Optional - filter to a specific element (substring match).	None
min_score	float	Minimum score threshold; facts below this are excluded.	0.5

Source code in clickproof/retriever.py

def query(
    self,
    app_name: str,
    app_version: str | None = None,
    element: str | None = None,
    min_score: float = 0.5,
) -> list[tuple[UIFact, FactScore]]:
    """Return (fact, score) pairs sorted by score descending.

    Args:
        app_name: Required - filter by application name.
        app_version: Optional - filter to a specific version.
        element: Optional - filter to a specific element (substring match).
        min_score: Minimum score threshold; facts below this are excluded.
    """
    facts = self._store.list_facts(app_name=app_name, app_version=app_version)

    if element is not None:
        facts = [f for f in facts if element.lower() in f.element.lower()]

    scored: list[tuple[UIFact, FactScore]] = []
    for fact in facts:
        observations = self._store.get_observations(fact.id)
        fs = self._scorer.score(fact, observations)
        if fs.score >= min_score:
            scored.append((fact, fs))

    scored.sort(key=lambda pair: pair[1].score, reverse=True)
    return scored

bootstrap_context(app_name, app_version='unknown')

Return a text summary of known facts for agent context injection.

The returned string can be prepended to an agent's system prompt to give it a snapshot of what is known about the target application.

Parameters:

Name	Type	Description	Default
app_name	str	Application to summarize.	required
app_version	str	Optional version to scope the summary.	'unknown'

Source code in clickproof/retriever.py

def bootstrap_context(self, app_name: str, app_version: str = "unknown") -> str:
    """Return a text summary of known facts for agent context injection.

    The returned string can be prepended to an agent's system prompt to
    give it a snapshot of what is known about the target application.

    Args:
        app_name: Application to summarize.
        app_version: Optional version to scope the summary.
    """
    pairs = self.query(app_name=app_name, app_version=app_version, min_score=0.0)

    if not pairs:
        return f"No known UI facts for {app_name!r} (version: {app_version!r})."

    lines: list[str] = [
        f"# clickproof: Known UI facts for {app_name!r} (version: {app_version!r})",
        f"# {len(pairs)} fact(s) retrieved, sorted by confidence\n",
    ]

    for fact, score in pairs:
        ctx = f" [{fact.context}]" if fact.context else ""
        lines.append(
            f"- [{score.score:.2f}] {fact.element} --{fact.action}--> {fact.outcome}{ctx}"
        )

    return "\n".join(lines)

FactScore(fact_id, app_name, app_version, element, score, observation_count, confirmed_count, last_observed, staleness_days) dataclass

Confidence score for a UIFact given its observation history.

Attributes:

Name	Type	Description
fact_id	str	ID of the scored UIFact.
app_name	str	Application identifier.
app_version	str	Version string.
element	str	Semantic element description.
score	float	Current confidence in [0.0, 1.0].
observation_count	int	Total number of observations.
confirmed_count	int	Number of confirming observations.
last_observed	float	Unix timestamp of the most recent observation.
staleness_days	float	Days since the last observation.

last_seen_at property

Alias for last_observed - provided for backward compatibility.

to_dict()

Serialize to a JSON-compatible dict.

Source code in clickproof/scorer.py

def to_dict(self) -> dict[str, Any]:
    """Serialize to a JSON-compatible dict."""
    return {
        "fact_id": self.fact_id,
        "app_name": self.app_name,
        "app_version": self.app_version,
        "element": self.element,
        "score": round(self.score, 4),
        "observation_count": self.observation_count,
        "confirmed_count": self.confirmed_count,
        "last_observed": self.last_observed,
        "staleness_days": round(self.staleness_days, 2),
    }

FactScorer

Computes confidence scores from observation history.

Algorithm

1. Base: ratio of confirmed / total observations (uses initial confidence if no obs).
2. Decay: multiply by staleness factor: e^(-0.1 * staleness_days).
3. Boost: scale up with observation count (more observations = more confident). Final score = base_ratio * staleness_decay * _count_boost(count)

score(fact, observations)

Compute a FactScore for a single UIFact given its observations.

Source code in clickproof/scorer.py

def score(self, fact: UIFact, observations: list[FactObservation]) -> FactScore:
    """Compute a FactScore for a single UIFact given its observations."""
    now = time.time()
    count = len(observations)

    if count == 0:
        # No observations - use the initial confidence, apply mild staleness from recorded_at
        staleness_days = (now - fact.recorded_at) / 86400.0
        decay = math.exp(-0.1 * staleness_days)
        score = fact.confidence * decay
        return FactScore(
            fact_id=fact.id,
            app_name=fact.app_name,
            app_version=fact.app_version,
            element=fact.element,
            score=max(0.0, min(1.0, score)),
            observation_count=0,
            confirmed_count=0,
            last_observed=fact.recorded_at,
            staleness_days=staleness_days,
        )

    confirmed_count = sum(1 for o in observations if o.confirmed)
    last_observed = max(o.observed_at for o in observations)

    base_ratio = confirmed_count / count
    staleness_days = (now - last_observed) / 86400.0
    staleness_decay = math.exp(-0.1 * staleness_days)
    count_boost = _count_boost(count)

    score = base_ratio * staleness_decay * count_boost

    return FactScore(
        fact_id=fact.id,
        app_name=fact.app_name,
        app_version=fact.app_version,
        element=fact.element,
        score=max(0.0, min(1.0, score)),
        observation_count=count,
        confirmed_count=confirmed_count,
        last_observed=last_observed,
        staleness_days=staleness_days,
    )

batch_score(facts, store)

Score all facts using observations from the store.

Source code in clickproof/scorer.py

def batch_score(self, facts: list[UIFact], store: FactStore) -> list[FactScore]:
    """Score all facts using observations from the store."""
    results = []
    for fact in facts:
        observations = store.get_observations(fact.id)
        results.append(self.score(fact, observations))
    return results

FactStore(path)

SQLite-backed store for UIFacts and observations.

Parameters:

Name	Type	Description	Default
path	str | Path	Path to the SQLite database file. Use ":memory:" for an in-memory database (useful for testing).	required

Source code in clickproof/store.py

def __init__(self, path: str | Path) -> None:
    self._path = str(path)
    self._conn = sqlite3.connect(self._path)
    self._conn.execute("PRAGMA foreign_keys = ON")
    self._conn.row_factory = sqlite3.Row
    self._create_tables()

add_fact(fact)

Insert a UIFact. Silently ignores duplicates (same id).

Source code in clickproof/store.py

def add_fact(self, fact: UIFact) -> None:
    """Insert a UIFact. Silently ignores duplicates (same id)."""
    self._conn.execute(
        """
        INSERT OR IGNORE INTO ui_facts
            (id, app_name, app_version, element, action, outcome,
             context, confidence, recorded_at, extra)
        VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
        """,
        (
            fact.id,
            fact.app_name,
            fact.app_version,
            fact.element,
            fact.action,
            fact.outcome,
            fact.context,
            fact.confidence,
            fact.recorded_at,
            "{}",
        ),
    )
    self._conn.commit()

get_fact(fact_id)

Return a UIFact by id, or None if not found.

Source code in clickproof/store.py

def get_fact(self, fact_id: str) -> UIFact | None:
    """Return a UIFact by id, or None if not found."""
    row = self._conn.execute("SELECT * FROM ui_facts WHERE id = ?", (fact_id,)).fetchone()
    if row is None:
        return None
    return self._row_to_fact(row)

set_confidence(fact_id, confidence)

Update stored confidence for a fact (OVERLAY-CLICK decay / invalidate).

Returns True if a row was updated. Confidence is clamped to [0.0, 1.0].

Source code in clickproof/store.py

def set_confidence(self, fact_id: str, confidence: float) -> bool:
    """Update stored confidence for a fact (OVERLAY-CLICK decay / invalidate).

    Returns True if a row was updated. Confidence is clamped to [0.0, 1.0].
    """
    conf = max(0.0, min(1.0, float(confidence)))
    cur = self._conn.execute(
        "UPDATE ui_facts SET confidence = ? WHERE id = ?",
        (conf, fact_id),
    )
    self._conn.commit()
    return cur.rowcount > 0

list_facts(app_name=None, app_version=None)

Return all UIFacts, optionally filtered by app_name and/or app_version.

Source code in clickproof/store.py

def list_facts(
    self,
    app_name: str | None = None,
    app_version: str | None = None,
) -> list[UIFact]:
    """Return all UIFacts, optionally filtered by app_name and/or app_version."""
    query = "SELECT * FROM ui_facts WHERE 1=1"
    params: list[str] = []
    if app_name is not None:
        query += " AND app_name = ?"
        params.append(app_name)
    if app_version is not None:
        query += " AND app_version = ?"
        params.append(app_version)
    query += " ORDER BY recorded_at DESC"
    rows = self._conn.execute(query, params).fetchall()
    return [self._row_to_fact(r) for r in rows]

add_observation(obs)

Insert a FactObservation. Silently ignores duplicates (same id).

Source code in clickproof/store.py

def add_observation(self, obs: FactObservation) -> None:
    """Insert a FactObservation. Silently ignores duplicates (same id)."""
    self._conn.execute(
        """
        INSERT OR IGNORE INTO fact_observations
            (id, fact_id, observed_at, confirmed, agent_run_id)
        VALUES (?, ?, ?, ?, ?)
        """,
        (
            obs.id,
            obs.fact_id,
            obs.observed_at,
            int(obs.confirmed),
            obs.agent_run_id,
        ),
    )
    self._conn.commit()

get_observations(fact_id)

Return all observations for a given fact_id, ordered by observed_at.

Source code in clickproof/store.py

def get_observations(self, fact_id: str) -> list[FactObservation]:
    """Return all observations for a given fact_id, ordered by observed_at."""
    rows = self._conn.execute(
        "SELECT * FROM fact_observations WHERE fact_id = ? ORDER BY observed_at ASC",
        (fact_id,),
    ).fetchall()
    return [self._row_to_obs(r) for r in rows]

close()

Close the database connection.

Source code in clickproof/store.py

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

project_decay(store, scorer, app_name, min_score=0.5)

Return decay projections for all facts belonging to app_name.

Parameters:

Name	Type	Description	Default
store	FactStore	An open FactStore to read from.	required
scorer	FactScorer	A FactScorer used to compute the current score.	required
app_name	str	Application name to project for.	required
min_score	float	Threshold below which a fact is considered stale.	0.5

Returns:

Type	Description
list[DecayProjection]	List of :class:DecayProjection objects, one per fact.

Source code in clickproof/analytics.py

def project_decay(
    store: FactStore,
    scorer: FactScorer,
    app_name: str,
    min_score: float = 0.5,
) -> list[DecayProjection]:
    """Return decay projections for all facts belonging to *app_name*.

    Args:
        store: An open FactStore to read from.
        scorer: A FactScorer used to compute the current score.
        app_name: Application name to project for.
        min_score: Threshold below which a fact is considered stale.

    Returns:
        List of :class:`DecayProjection` objects, one per fact.
    """
    facts = store.list_facts(app_name=app_name)
    projections: list[DecayProjection] = []

    for fact in facts:
        observations = store.get_observations(fact.id)
        fact_score = scorer.score(fact, observations)
        current_score = fact_score.score
        staleness_days = fact_score.staleness_days

        count = fact_score.observation_count

        if count == 0:
            # No observations - scorer uses fact.confidence as the base ratio
            base_ratio = fact.confidence
            count_boost = 1.0
        else:
            confirmed_count = fact_score.confirmed_count
            base_ratio = confirmed_count / count
            count_boost = _count_boost(count)

        score_in_7_days = max(0.0, _project_score(base_ratio, count_boost, staleness_days, 7.0))
        score_in_30_days = max(0.0, _project_score(base_ratio, count_boost, staleness_days, 30.0))

        # Solve: base_ratio * exp(-0.1 * (staleness + x)) * count_boost = min_score
        # => -0.1 * (staleness + x) = ln(min_score / (base_ratio * count_boost))
        # => staleness + x = -ln(min_score / (base_ratio * count_boost)) / 0.1
        # => x = -ln(min_score / (base_ratio * count_boost)) / 0.1 - staleness
        peak = base_ratio * count_boost
        if min_score <= 0.0:
            # Score never reaches 0 (asymptotic), so the threshold is effectively never crossed
            days_until_threshold = 9999.0
        elif peak <= min_score:
            # Score can never reach the threshold even at days_since=0
            days_until_threshold = 0.0
        elif current_score < min_score:
            days_until_threshold = 0.0
        else:
            days_until_threshold = max(
                0.0,
                (-math.log(min_score / peak) / 0.1) - staleness_days,
            )

        if current_score < min_score:
            recommendation = "archive"
        elif score_in_7_days < min_score:
            recommendation = "re-validate"
        else:
            recommendation = "ok"

        projections.append(
            DecayProjection(
                fact_id=fact.id,
                element=fact.element,
                current_score=round(current_score, 6),
                score_in_7_days=round(score_in_7_days, 6),
                score_in_30_days=round(score_in_30_days, 6),
                days_until_threshold=round(days_until_threshold, 2),
                recommendation=recommendation,
            )
        )

    return projections

stale_facts(store, scorer, app_name, min_score=0.5)

Return facts whose current score is below min_score.

Parameters:

Name	Type	Description	Default
store	FactStore	An open FactStore to read from.	required
scorer	FactScorer	A FactScorer used to compute the current score.	required
app_name	str	Application name to filter by.	required
min_score	float	Score threshold; facts with score strictly below this are returned.	0.5

Returns:

Type	Description
list[UIFact]	List of :class:UIFact objects whose score < min_score.

Source code in clickproof/analytics.py

def stale_facts(
    store: FactStore,
    scorer: FactScorer,
    app_name: str,
    min_score: float = 0.5,
) -> list[UIFact]:
    """Return facts whose current score is below *min_score*.

    Args:
        store: An open FactStore to read from.
        scorer: A FactScorer used to compute the current score.
        app_name: Application name to filter by.
        min_score: Score threshold; facts with score strictly below this are returned.

    Returns:
        List of :class:`UIFact` objects whose score < *min_score*.
    """
    facts = store.list_facts(app_name=app_name)
    stale: list[UIFact] = []

    for fact in facts:
        observations = store.get_observations(fact.id)
        fact_score = scorer.score(fact, observations)
        if fact_score.score < min_score:
            stale.append(fact)

    return stale

export_bootstrap_pack(store, app_name)

Export a 'bootstrap pack' - minimal JSON with top-scored facts for an app.

Only facts with a score >= 0.5 are included, sorted by score descending, capped at the top 20.

Parameters:

Name	Type	Description	Default
store	FactStore	An open FactStore to read from.	required
app_name	str	Application whose facts to export.	required

Returns:

Type	Description
str	JSON string with keys version, app_name, bootstrap_pack,
str	facts, count.

Source code in clickproof/bulk.py

def export_bootstrap_pack(store: FactStore, app_name: str) -> str:
    """Export a 'bootstrap pack' - minimal JSON with top-scored facts for an app.

    Only facts with a score >= 0.5 are included, sorted by score descending,
    capped at the top 20.

    Args:
        store: An open FactStore to read from.
        app_name: Application whose facts to export.

    Returns:
        JSON string with keys ``version``, ``app_name``, ``bootstrap_pack``,
        ``facts``, ``count``.
    """
    scorer = FactScorer()
    retriever = FactRetriever(store, scorer)
    pairs = retriever.query(app_name=app_name, min_score=0.5)
    top = pairs[:20]

    payload = {
        "version": "1.0",
        "app_name": app_name,
        "bootstrap_pack": True,
        "facts": [fact.to_dict() for fact, _score in top],
        "count": len(top),
    }
    return json.dumps(payload, indent=2)

export_facts(store, app_name=None)

Export facts and observations as JSON.

Parameters:

Name	Type	Description	Default
store	FactStore	An open FactStore to read from.	required
app_name	str | None	If given, only export facts for this application.	None

Returns:

Type	Description
str	JSON string with keys version, app_name, facts, count.

Source code in clickproof/bulk.py

def export_facts(store: FactStore, app_name: str | None = None) -> str:
    """Export facts and observations as JSON.

    Args:
        store: An open FactStore to read from.
        app_name: If given, only export facts for this application.

    Returns:
        JSON string with keys ``version``, ``app_name``, ``facts``, ``count``.
    """
    facts = store.list_facts(app_name=app_name)
    entries = []
    for fact in facts:
        observations = store.get_observations(fact.id)
        entries.append(
            {
                "fact": fact.to_dict(),
                "observations": [obs.to_dict() for obs in observations],
            }
        )
    payload = {
        "version": "1.0",
        "app_name": app_name or "all",
        "facts": entries,
        "count": len(entries),
    }
    return json.dumps(payload, indent=2)

import_facts(store, json_str, merge_strategy='upsert')

Import facts from JSON.

Parameters:

Name	Type	Description	Default
store	FactStore	An open FactStore to write into.	required
json_str	str	JSON string produced by :func:export_facts.	required
merge_strategy	str	One of "upsert", "skip_existing", or "overwrite". "upsert" and "overwrite" both call :meth:FactStore.add_fact (which is itself an upsert). "skip_existing" skips facts whose id is already in the store.	'upsert'

Returns:

Type	Description
int	Number of facts imported (observations are always imported alongside).

Raises:

Type	Description
ValueError	If merge_strategy is not a recognised value.

Source code in clickproof/bulk.py

def import_facts(
    store: FactStore,
    json_str: str,
    merge_strategy: str = "upsert",
) -> int:
    """Import facts from JSON.

    Args:
        store: An open FactStore to write into.
        json_str: JSON string produced by :func:`export_facts`.
        merge_strategy: One of ``"upsert"``, ``"skip_existing"``, or
            ``"overwrite"``.  ``"upsert"`` and ``"overwrite"`` both call
            :meth:`FactStore.add_fact` (which is itself an upsert).
            ``"skip_existing"`` skips facts whose id is already in the store.

    Returns:
        Number of facts imported (observations are always imported alongside).

    Raises:
        ValueError: If *merge_strategy* is not a recognised value.
    """
    if merge_strategy not in {"upsert", "skip_existing", "overwrite"}:
        raise ValueError(
            f"Unknown merge_strategy {merge_strategy!r}. "
            "Choose 'upsert', 'skip_existing', or 'overwrite'."
        )

    data = json.loads(json_str)
    imported = 0

    for entry in data.get("facts", []):
        fact = UIFact.from_dict(entry["fact"])

        if merge_strategy == "skip_existing" and store.get_fact(fact.id) is not None:
            continue

        store.add_fact(fact)
        imported += 1

        for obs_dict in entry.get("observations", []):
            obs = FactObservation.from_dict(obs_dict)
            store.add_observation(obs)

    return imported

apply_click_outcome(store, attempt, *, scorer=None, miss_confidence_factor=0.25, invalidate_confidence=0.05)

Record click result: confirm on hit, refute + decay confidence on miss.

OVERLAY-CLICK product control

• hit + effect → confirmed observation (confidence preserved)
• miss / overlay / force-without-effect → refuted observation and hard confidence decay (miss_confidence_factor or floor at invalidate_confidence)

Returns:

Type	Description
ClickOutcomeResult	class:ClickOutcomeResult with before/after scores.

Source code in clickproof/closed_loop.py

def apply_click_outcome(
    store: FactStore,
    attempt: ClickAttempt,
    *,
    scorer: FactScorer | None = None,
    miss_confidence_factor: float = 0.25,
    invalidate_confidence: float = 0.05,
) -> ClickOutcomeResult:
    """Record click result: confirm on hit, refute + decay confidence on miss.

    OVERLAY-CLICK product control:
      * hit + effect → confirmed observation (confidence preserved)
      * miss / overlay / force-without-effect → refuted observation and
        hard confidence decay (``miss_confidence_factor`` or floor at
        ``invalidate_confidence``)

    Returns:
        :class:`ClickOutcomeResult` with before/after scores.
    """
    scorer = scorer or FactScorer()
    fact = store.get_fact(attempt.fact_id)
    if fact is None:
        return ClickOutcomeResult(
            ok=False,
            invalidated=False,
            miss_kind="unknown_fact",
            score_before=0.0,
            score_after=0.0,
            confidence_after=0.0,
            observation_confirmed=False,
            fact_id=attempt.fact_id,
            reason=f"fact not found: {attempt.fact_id}",
        )

    obs_before = store.get_observations(fact.id)
    score_before = scorer.score(fact, obs_before).score

    if not attempt.is_miss:
        store.add_observation(
            FactObservation(
                fact_id=fact.id,
                observed_at=time.time(),
                confirmed=True,
                agent_run_id=attempt.agent_run_id or "click",
            )
        )
        score_after = scorer.score(fact, store.get_observations(fact.id)).score
        return ClickOutcomeResult(
            ok=True,
            invalidated=False,
            miss_kind=None,
            score_before=score_before,
            score_after=score_after,
            confidence_after=fact.confidence,
            observation_confirmed=True,
            fact_id=fact.id,
            reason="click hit target; observation confirmed",
        )

    # Miss path - refute and decay
    kind = attempt.miss_kind or "miss"
    store.add_observation(
        FactObservation(
            fact_id=fact.id,
            observed_at=time.time(),
            confirmed=False,
            agent_run_id=attempt.agent_run_id or f"miss:{kind}",
        )
    )
    new_conf = max(invalidate_confidence, fact.confidence * miss_confidence_factor)
    store.set_confidence(fact.id, new_conf)
    updated = store.get_fact(fact.id) or fact
    updated.confidence = new_conf
    score_after = scorer.score(updated, store.get_observations(fact.id)).score

    return ClickOutcomeResult(
        ok=False,
        invalidated=True,
        miss_kind=kind,
        score_before=score_before,
        score_after=score_after,
        confidence_after=new_conf,
        observation_confirmed=False,
        fact_id=fact.id,
        reason=(
            f"OVERLAY-CLICK miss kind={kind}: force_used={attempt.force_used} "
            f"overlay={attempt.overlay_intercepted} hit={attempt.hit} "
            f"effect={attempt.observed_effect}; confidence {fact.confidence:.3f}→{new_conf:.3f} "
            f"score {score_before:.3f}→{score_after:.3f}"
        ),
    )

assert_click_ok(store, attempt, **kwargs)

Apply gate_click_attempt and raise :class:ClosedLoopError unless ok.

Source code in clickproof/closed_loop.py

def assert_click_ok(
    store: FactStore,
    attempt: ClickAttempt,
    **kwargs: Any,
) -> GateOutcome:
    """Apply gate_click_attempt and raise :class:`ClosedLoopError` unless ok."""
    outcome = gate_click_attempt(store, attempt, **kwargs)
    if not outcome.ok:
        raise ClosedLoopError(f"{outcome.verdict}: {outcome.reason}")
    return outcome

assert_session_bootstrapped(store, session, *, app_name, **kwargs)

Raise :class:ClosedLoopError unless session memory gate passes.

Source code in clickproof/closed_loop.py

def assert_session_bootstrapped(
    store: FactStore,
    session: SessionMemory | None,
    *,
    app_name: str,
    **kwargs: Any,
) -> GateOutcome:
    """Raise :class:`ClosedLoopError` unless session memory gate passes."""
    outcome = gate_session_memory(store, session, app_name=app_name, **kwargs)
    if not outcome.ok:
        raise ClosedLoopError(f"{outcome.verdict}: {outcome.reason}")
    return outcome

assert_task_aligned(declared_task, proposed_action, **kwargs)

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

Source code in clickproof/closed_loop.py

def assert_task_aligned(
    declared_task: str,
    proposed_action: str,
    **kwargs: Any,
) -> GateOutcome:
    """Raise :class:`ClosedLoopError` unless :func:`gate_task_alignment` is ok."""
    outcome = gate_task_alignment(declared_task, proposed_action, **kwargs)
    if not outcome.ok:
        raise ClosedLoopError(f"{outcome.verdict}: {outcome.reason}")
    return outcome

assert_usable_facts(source, **kwargs)

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

Source code in clickproof/closed_loop.py

def assert_usable_facts(
    source: FactStore | Sequence[UIFact] | str | Path,
    **kwargs: Any,
) -> GateOutcome:
    """Gate facts and raise :class:`ClosedLoopError` unless outcome is ok."""
    outcome = gate_facts(source, **kwargs)
    if not outcome.ok:
        raise ClosedLoopError(f"{outcome.verdict}: {outcome.reason}")
    return outcome

gate_click_attempt(store, attempt, *, scorer=None, min_score_after=0.5, apply=True, miss_confidence_factor=0.25)

Gate a click: OVERLAY-CLICK misses FAIL and invalidate the fact.

Parameters:

Name	Type	Description	Default
store	FactStore	Fact store containing the target fact.	required
attempt	ClickAttempt	Click report from the computer-use runtime.	required
min_score_after	float	After a hit, require score >= this for PASS.	0.5
apply	bool	If True, write refute/confirm + confidence decay to the store.	True
miss_confidence_factor	float	Multiplier applied to confidence on miss.	0.25

Returns:

Type	Description
GateOutcome	FAIL_LOUD if fact missing; FAIL on miss or post-hit unusable score;
GateOutcome	PASS only on verified hit with usable score.

Source code in clickproof/closed_loop.py

def gate_click_attempt(
    store: FactStore,
    attempt: ClickAttempt,
    *,
    scorer: FactScorer | None = None,
    min_score_after: float = 0.5,
    apply: bool = True,
    miss_confidence_factor: float = 0.25,
) -> GateOutcome:
    """Gate a click: OVERLAY-CLICK misses FAIL and invalidate the fact.

    Args:
        store: Fact store containing the target fact.
        attempt: Click report from the computer-use runtime.
        min_score_after: After a hit, require score >= this for PASS.
        apply: If True, write refute/confirm + confidence decay to the store.
        miss_confidence_factor: Multiplier applied to confidence on miss.

    Returns:
        FAIL_LOUD if fact missing; FAIL on miss or post-hit unusable score;
        PASS only on verified hit with usable score.
    """
    if apply:
        result = apply_click_outcome(
            store,
            attempt,
            scorer=scorer,
            miss_confidence_factor=miss_confidence_factor,
        )
    else:
        # Dry-run classification only
        fact = store.get_fact(attempt.fact_id)
        if fact is None:
            return _fail_loud(f"fact not found: {attempt.fact_id}")
        scorer = scorer or FactScorer()
        score = scorer.score(fact, store.get_observations(fact.id)).score
        if attempt.is_miss:
            return GateOutcome(
                ok=False,
                verdict="FAIL",
                reason=f"OVERLAY-CLICK miss (dry-run) kind={attempt.miss_kind}",
                exit_code=1,
                fact_count=1,
                usable_count=0,
                stale_count=1,
                min_score_seen=score,
            )
        usable = 1 if score >= min_score_after else 0
        return GateOutcome(
            ok=usable == 1,
            verdict="PASS" if usable else "FAIL",
            reason=f"dry-run hit score={score}",
            exit_code=0 if usable else 1,
            fact_count=1,
            usable_count=usable,
            stale_count=1 - usable,
            min_score_seen=score,
        )

    if result.miss_kind == "unknown_fact":
        return _fail_loud(result.reason)

    if result.invalidated or not result.ok:
        return GateOutcome(
            ok=False,
            verdict="FAIL",
            reason=result.reason,
            exit_code=1,
            fact_count=1,
            usable_count=0,
            stale_count=1,
            min_score_seen=result.score_after,
        )

    usable = 1 if result.score_after >= min_score_after else 0
    if usable == 0:
        return GateOutcome(
            ok=False,
            verdict="FAIL",
            reason=(
                f"click hit but score_after={result.score_after:.3f} "
                f"< min_score_after={min_score_after}"
            ),
            exit_code=1,
            fact_count=1,
            usable_count=0,
            stale_count=1,
            min_score_seen=result.score_after,
        )

    return GateOutcome(
        ok=True,
        verdict="PASS",
        reason=result.reason,
        exit_code=0,
        fact_count=1,
        usable_count=1,
        stale_count=0,
        min_score_seen=result.score_after,
    )

gate_facts(source, *, min_score=0.5, app_name=None, require_usable=True, scorer=None)

Read UI facts and fail loudly when the store is empty or unusable.

Parameters:

Name	Type	Description	Default
source	FactStore | Sequence[UIFact] | str | Path	Open :class:FactStore, path to a SQLite db, or a sequence of :class:UIFact (scores use fact.confidence when no store/obs).	required
min_score	float	Score threshold; facts strictly below this count as stale.	0.5
app_name	str | None	Optional filter when reading from a store.	None
require_usable	bool	If True, zero usable facts with some present is FAIL.	True
scorer	FactScorer | None	Optional :class:FactScorer; defaults to a new instance.	None

Returns:

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

Source code in clickproof/closed_loop.py

def gate_facts(
    source: FactStore | Sequence[UIFact] | str | Path,
    *,
    min_score: float = 0.5,
    app_name: str | None = None,
    require_usable: bool = True,
    scorer: FactScorer | None = None,
) -> GateOutcome:
    """Read UI facts and fail loudly when the store is empty or unusable.

    Args:
        source: Open :class:`FactStore`, path to a SQLite db, or a sequence of
            :class:`UIFact` (scores use fact.confidence when no store/obs).
        min_score: Score threshold; facts strictly below this count as stale.
        app_name: Optional filter when reading from a store.
        require_usable: If True, zero usable facts with some present is FAIL.
        scorer: Optional :class:`FactScorer`; defaults to a new instance.

    Returns:
        :class:`GateOutcome` - callers should ``sys.exit(outcome.exit_code)``.
    """
    scorer = scorer or FactScorer()
    owns = False
    store: FactStore | None = None
    try:
        scores: list[float] = []
        if isinstance(source, FactStore):
            store = source
            facts = store.list_facts(app_name=app_name)
            for fact in facts:
                obs = store.get_observations(fact.id)
                scores.append(scorer.score(fact, obs).score)
        elif isinstance(source, (str, Path)):
            path = Path(source)
            if str(source) != ":memory:" and not path.is_file():
                return _fail_loud(f"fact store not found: {path}")
            try:
                store = FactStore(path if str(source) != ":memory:" else ":memory:")
                owns = True
                facts = store.list_facts(app_name=app_name)
                for fact in facts:
                    obs = store.get_observations(fact.id)
                    scores.append(scorer.score(fact, obs).score)
            except Exception as exc:
                return _fail_loud(f"open fact store failed: {exc.__class__.__name__}: {exc}")
        else:
            facts = list(source)
            # Sequence path: no observations - score from fact.confidence via scorer
            for fact in facts:
                scores.append(scorer.score(fact, []).score)

        if len(facts) == 0:
            return _fail_loud(
                "empty facts - no load-bearing GUI behavioral facts to gate "
                "(write-only store is ornament)"
            )

        usable = sum(1 for s in scores if s >= min_score)
        stale = len(scores) - usable
        min_seen = min(scores) if scores else None

        if require_usable and usable == 0:
            return GateOutcome(
                ok=False,
                verdict="FAIL",
                reason=(
                    f"no usable facts above min_score={min_score}: "
                    f"fact_count={len(facts)} stale={stale} "
                    f"min_score_seen={min_seen}"
                ),
                exit_code=1,
                fact_count=len(facts),
                usable_count=0,
                stale_count=stale,
                min_score_seen=min_seen,
            )

        return GateOutcome(
            ok=True,
            verdict="PASS",
            reason=(
                f"facts ok: count={len(facts)} usable={usable} stale={stale} "
                f"min_score={min_score} min_seen={min_seen}"
            ),
            exit_code=0,
            fact_count=len(facts),
            usable_count=usable,
            stale_count=stale,
            min_score_seen=min_seen,
        )
    finally:
        if owns and store is not None:
            with contextlib.suppress(Exception):
                store.close()

gate_session_memory(store, session, *, app_name, app_version=None, min_score=0.5, require_load_when_known=True, scorer=None)

Gate session bootstrap against the durable fact store (GUI-MEMORY).

• Store has usable facts for app, session is None or empty load → FAIL (re-discover trap - known UI not injected).
• Store empty for app → FAIL_LOUD (nothing to remember; cold discover is expected but not a silent pass of "memory ok").
• Session loaded usable facts matching store → PASS.

Parameters:

Name	Type	Description	Default
store	FactStore	Durable :class:FactStore.	required
session	SessionMemory | None	Result of :func:load_session_memory, or None if agent skipped.	required
app_name	str	Application under automation.	required
require_load_when_known	bool	If True (default), skip-load with known facts fails.	True

Source code in clickproof/closed_loop.py

def gate_session_memory(
    store: FactStore,
    session: SessionMemory | None,
    *,
    app_name: str,
    app_version: str | None = None,
    min_score: float = 0.5,
    require_load_when_known: bool = True,
    scorer: FactScorer | None = None,
) -> GateOutcome:
    """Gate session bootstrap against the durable fact store (GUI-MEMORY).

    * Store has usable facts for app, session is ``None`` or empty load →
      **FAIL** (re-discover trap - known UI not injected).
    * Store empty for app → **FAIL_LOUD** (nothing to remember; cold discover
      is expected but not a silent pass of "memory ok").
    * Session loaded usable facts matching store → **PASS**.

    Args:
        store: Durable :class:`FactStore`.
        session: Result of :func:`load_session_memory`, or None if agent skipped.
        app_name: Application under automation.
        require_load_when_known: If True (default), skip-load with known facts fails.
    """
    known = store_usable_count(
        store,
        app_name,
        app_version=app_version,
        min_score=min_score,
        scorer=scorer,
    )

    if known == 0:
        return GateOutcome(
            ok=False,
            verdict="FAIL_LOUD",
            reason=(
                f"GUI-MEMORY: no usable facts for app {app_name!r} "
                f"(min_score={min_score}) - store empty; cold re-discover only, "
                f"not a memory pass"
            ),
            exit_code=2,
            fact_count=0,
            usable_count=0,
            stale_count=0,
            min_score_seen=None,
        )

    if session is None:
        if require_load_when_known:
            return GateOutcome(
                ok=False,
                verdict="FAIL",
                reason=(
                    f"GUI-MEMORY: store has {known} usable fact(s) for {app_name!r} "
                    f"but session never called load_session_memory - refusing "
                    f"cold re-discover"
                ),
                exit_code=1,
                fact_count=known,
                usable_count=0,
                stale_count=known,
                min_score_seen=None,
            )
        return GateOutcome(
            ok=False,
            verdict="FAIL",
            reason="session is None",
            exit_code=1,
            fact_count=known,
            usable_count=0,
            stale_count=known,
        )

    if session.app_name != app_name:
        return GateOutcome(
            ok=False,
            verdict="FAIL",
            reason=(f"GUI-MEMORY: session app {session.app_name!r} != gate app {app_name!r}"),
            exit_code=1,
            fact_count=known,
            usable_count=session.usable_count,
            stale_count=max(0, known - session.usable_count),
        )

    if session.is_empty and require_load_when_known:
        return GateOutcome(
            ok=False,
            verdict="FAIL",
            reason=(
                f"GUI-MEMORY: session {session.session_id!r} loaded 0 usable facts "
                f"but store has {known} for {app_name!r} - incomplete bootstrap"
            ),
            exit_code=1,
            fact_count=known,
            usable_count=0,
            stale_count=known,
        )

    return GateOutcome(
        ok=True,
        verdict="PASS",
        reason=(
            f"GUI-MEMORY ok: session {session.session_id!r} loaded "
            f"{session.usable_count}/{known} usable fact(s) for {app_name!r}"
        ),
        exit_code=0,
        fact_count=known,
        usable_count=session.usable_count,
        stale_count=max(0, known - session.usable_count),
        min_score_seen=None,
    )

gate_task_alignment(declared_task, proposed_action, *, allowed_actions=None, target=None, allowed_targets=None, refuse_high_risk_outside_allowlist=True, infer_from_task=True)

Block adversarial CUA actions behind a legitimate task (Invisible Ink).

Public case (arXiv 2608.02018): Invisible Ink Threats - Adversarial Goals Behind Legitimate Tasks in Computer-Use Agents. The user/task text is benign; UI injection or model drift proposes delete/export/auth that was never authorized by the task.

Rules:

1. Empty task or empty action → FAIL_LOUD
2. Build allowlist from allowed_actions and/or infer_allowlist_from_task
3. High-risk proposed action not in allowlist → FAIL (human_required)
4. Any proposed action not in allowlist when allowlist non-empty → FAIL
5. target not in allowed_targets when both set → FAIL
6. In-scope action → PASS

Parameters:

Name	Type	Description	Default
declared_task	str	User/task description for this CUA step/session.	required
proposed_action	str	Tool/click/verb the agent wants to run.	required
allowed_actions	Sequence[str] | None	Explicit allowlist (overrides pure inference if set; merged with inference when infer_from_task is True).	None
target	str | None	Optional UI target / path / element id.	None
allowed_targets	Sequence[str] | None	Optional target allowlist.	None
refuse_high_risk_outside_allowlist	bool	High-risk out-of-scope → FAIL.	True
infer_from_task	bool	Merge heuristic verbs from task text into allowlist.	True

Source code in clickproof/closed_loop.py

def gate_task_alignment(
    declared_task: str,
    proposed_action: str,
    *,
    allowed_actions: Sequence[str] | None = None,
    target: str | None = None,
    allowed_targets: Sequence[str] | None = None,
    refuse_high_risk_outside_allowlist: bool = True,
    infer_from_task: bool = True,
) -> GateOutcome:
    """Block adversarial CUA actions behind a legitimate task (Invisible Ink).

    Public case (arXiv 2608.02018): *Invisible Ink Threats - Adversarial Goals
    Behind Legitimate Tasks in Computer-Use Agents*. The user/task text is
    benign; UI injection or model drift proposes delete/export/auth that was
    never authorized by the task.

    Rules:

    1. Empty task or empty action → **FAIL_LOUD**
    2. Build allowlist from ``allowed_actions`` and/or ``infer_allowlist_from_task``
    3. High-risk proposed action not in allowlist → **FAIL** (``human_required``)
    4. Any proposed action not in allowlist when allowlist non-empty → **FAIL**
    5. ``target`` not in ``allowed_targets`` when both set → **FAIL**
    6. In-scope action → **PASS**

    Args:
        declared_task: User/task description for this CUA step/session.
        proposed_action: Tool/click/verb the agent wants to run.
        allowed_actions: Explicit allowlist (overrides pure inference if set;
            merged with inference when ``infer_from_task`` is True).
        target: Optional UI target / path / element id.
        allowed_targets: Optional target allowlist.
        refuse_high_risk_outside_allowlist: High-risk out-of-scope → FAIL.
        infer_from_task: Merge heuristic verbs from task text into allowlist.
    """
    task = (declared_task or "").strip()
    action = _canonical_action(proposed_action)
    task_preview = task[:80] if task else None

    if not task:
        return _fail_loud(
            "INVISIBLE-INK: empty declared_task - cannot align actions to a "
            "phantom goal (adversarial free-run)",
            human_required=True,
            action=action or None,
            task=None,
            risk="high_risk",
        )
    if not action:
        return _fail_loud(
            "INVISIBLE-INK: empty proposed_action - cannot gate phantom CUA step",
            human_required=True,
            action=None,
            task=task_preview,
            risk="high_risk",
        )

    allow = set(normalize_allowlist(allowed_actions))
    if infer_from_task:
        allow |= set(infer_allowlist_from_task(task))
    # Explicit allowlist alone is enough even if empty inference
    if allowed_actions is not None and not infer_from_task:
        allow = set(normalize_allowlist(allowed_actions))

    risk = "high_risk" if is_high_risk_cua_action(action) else "safe"
    head = action.split(":", 1)[0]
    in_allow = action in allow or head in allow

    if risk == "high_risk" and refuse_high_risk_outside_allowlist and not in_allow:
        return _fail(
            f"INVISIBLE-INK: high-risk action {action!r} not authorized by task "
            f"{task_preview!r} allowlist={sorted(allow)[:12]} - "
            f"adversarial goal behind legitimate task (arXiv 2608.02018)",
            human_required=True,
            action=action,
            task=task_preview,
            risk=risk,
        )

    if allow and not in_allow:
        return _fail(
            f"INVISIBLE-INK: action {action!r} outside task allowlist "
            f"{sorted(allow)[:12]} for task {task_preview!r}",
            human_required=True,
            action=action,
            task=task_preview,
            risk=risk,
        )

    if target is not None and allowed_targets is not None:
        targets = {str(t).strip() for t in allowed_targets if str(t).strip()}
        t = str(target).strip()
        if targets and t not in targets:
            return _fail(
                f"INVISIBLE-INK: target {t!r} not in allowed_targets "
                f"(task={task_preview!r}) - possible UI injection detour",
                human_required=True,
                action=action,
                task=task_preview,
                risk=risk,
            )

    return GateOutcome(
        ok=True,
        verdict="PASS",
        reason=(
            f"INVISIBLE-INK ok: action={action!r} risk={risk} "
            f"task={task_preview!r} allowlist_size={len(allow)}"
        ),
        exit_code=0,
        human_required=False,
        action=action,
        task=task_preview,
        risk=risk,
    )

infer_allowlist_from_task(task)

Heuristic allowlist from declared task text (no LLM).

Extracts simple verbs that appear as whole words and always includes a base set of safe navigation actions. High-risk verbs only enter the allowlist if the task text literally mentions them.

Source code in clickproof/closed_loop.py

def infer_allowlist_from_task(task: str) -> frozenset[str]:
    """Heuristic allowlist from declared task text (no LLM).

    Extracts simple verbs that appear as whole words and always includes a
    base set of safe navigation actions. High-risk verbs only enter the
    allowlist if the task text literally mentions them.
    """
    text = (task or "").lower()
    tokens = set(re.findall(r"[a-z_][a-z0-9_]*", text))
    allowed: set[str] = set(DEFAULT_SAFE_CUA_ACTIONS)
    for risk in DEFAULT_HIGH_RISK_CUA_ACTIONS:
        if risk in tokens or risk.replace("_", " ") in text:
            allowed.add(risk)
    # Common synonyms in task phrasing
    if "close" in tokens or "dismiss" in tokens:
        allowed |= {"close", "dismiss", "click", "cancel"}
    if "open" in tokens or "navigate" in tokens:
        allowed |= {"open", "navigate", "click"}
    if "fill" in tokens or "type" in tokens or "enter" in tokens:
        allowed |= {"type", "click", "select", "focus"}
        # "enter password" is high-risk - only if password mentioned
        if "password" in tokens or "credential" in tokens:
            allowed.add("enter_password")
    return frozenset(allowed)

is_high_risk_cua_action(action, *, extra=None)

True if action is high-risk for computer-use (delete/export/auth/…).

Source code in clickproof/closed_loop.py

def is_high_risk_cua_action(
    action: str,
    *,
    extra: Iterable[str] | None = None,
) -> bool:
    """True if *action* is high-risk for computer-use (delete/export/auth/…)."""
    a = _canonical_action(action)
    if not a:
        return True  # empty never safe
    banned = set(DEFAULT_HIGH_RISK_CUA_ACTIONS)
    if extra:
        banned |= {_canonical_action(x) for x in extra}
    head = a.split(":", 1)[0]
    return a in banned or head in banned

load_session_memory(store, app_name, *, app_version=None, session_id=None, min_score=0.5, scorer=None)

Load known UI facts into a session (bootstrap for computer-use agents).

This is the load-bearing writer→reader path for GUI-MEMORY: call at session start so the agent does not re-discover controls every run.

Source code in clickproof/closed_loop.py

def load_session_memory(
    store: FactStore,
    app_name: str,
    *,
    app_version: str | None = None,
    session_id: str | None = None,
    min_score: float = 0.5,
    scorer: FactScorer | None = None,
) -> SessionMemory:
    """Load known UI facts into a session (bootstrap for computer-use agents).

    This is the load-bearing *writer→reader* path for GUI-MEMORY: call at
    session start so the agent does not re-discover controls every run.
    """
    retriever = FactRetriever(store, scorer=scorer)
    pairs = retriever.query(
        app_name=app_name,
        app_version=app_version,
        min_score=min_score,
    )
    # bootstrap_context uses min_score=0.0 for text; we still only *count* usable
    text = retriever.bootstrap_context(
        app_name=app_name,
        app_version=app_version or "unknown",
    )
    ids = tuple(f.id for f, _ in pairs)
    return SessionMemory(
        session_id=session_id or _secrets.token_hex(4),
        app_name=app_name,
        app_version=app_version,
        loaded_fact_ids=ids,
        bootstrap_text=text,
        loaded_at=time.time(),
        usable_count=len(pairs),
        min_score=min_score,
    )

store_usable_count(store, app_name, *, app_version=None, min_score=0.5, scorer=None)

Count usable facts in the store for app_name (no session load).

Source code in clickproof/closed_loop.py

def store_usable_count(
    store: FactStore,
    app_name: str,
    *,
    app_version: str | None = None,
    min_score: float = 0.5,
    scorer: FactScorer | None = None,
) -> int:
    """Count usable facts in the store for *app_name* (no session load)."""
    retriever = FactRetriever(store, scorer=scorer)
    return len(retriever.query(app_name=app_name, app_version=app_version, min_score=min_score))

analyze_cve(decision, *, required_sparse_keys=None, extra_sparse_keys=None, batch=None)

Analyse one decision (and optional batch) for CVE failure modes.

Source code in clickproof/context_vars.py

def analyze_cve(
    decision: ContextDecision | dict[str, Any],
    *,
    required_sparse_keys: Iterable[str] | None = None,
    extra_sparse_keys: Iterable[str] | None = None,
    batch: Sequence[ContextDecision | dict[str, Any]] | None = None,
) -> CVEReport:
    """Analyse one decision (and optional batch) for CVE failure modes."""
    d = _as_decision(decision)
    required = [_norm_key(x) for x in (required_sparse_keys or ()) if str(x).strip()]
    sparse_keys = {
        _norm_key(k) for k in d.sparse_context if d.sparse_context.get(k) not in (None, "")
    }
    # also count keys that match sparse taxonomy even if empty later
    present = tuple(
        sorted(
            k
            for k in sparse_keys
            if is_sparse_context_key(k, extra=extra_sparse_keys) or k in required
        )
    )
    missing_list: list[str] = []
    for req in required:
        found = False
        for sk, sv in d.sparse_context.items():
            if _norm_key(sk) == req and sv not in (None, ""):
                found = True
                break
        if not found:
            missing_list.append(req)
    missing = tuple(missing_list)

    attended_norm = {_norm_key(a) for a in d.attended_keys}
    attended_sparse = tuple(sorted(k for k in present if k in attended_norm))
    ignored = tuple(sorted(k for k in present if k not in attended_norm))

    dominant_vals = {_norm_key(k): v for k, v in d.dominant_cues.items() if v not in (None, "")}
    dominant_only = bool(dominant_vals) and (len(present) == 0 or len(attended_sparse) == 0)

    # Cross-context collapse: multiple distinct sparse fingerprints map to same choice
    collapsed: list[str] = []
    by_ctx = dict(d.choice_by_context)
    if batch:
        for item in batch:
            bd = _as_decision(item)
            fp = context_fingerprint(bd.sparse_context)
            if fp:
                by_ctx[fp] = bd.choice
    if len(by_ctx) >= 2:
        # group by choice
        from collections import defaultdict

        groups: dict[str, list[str]] = defaultdict(list)
        for fp, ch in by_ctx.items():
            groups[str(ch)].append(fp)
        for _ch, fps in groups.items():
            if len(fps) >= 2:
                collapsed.extend(fps)

    return CVEReport(
        sparse_keys_present=present,
        sparse_keys_missing=missing,
        attended_sparse=attended_sparse,
        ignored_sparse=ignored,
        dominant_only=dominant_only,
        cross_context_collapse=bool(collapsed),
        collapsed_contexts=tuple(collapsed[:20]),
    )

assert_context_variables_ok(decision=None, **kwargs)

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

Source code in clickproof/context_vars.py

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

context_fingerprint(sparse)

Stable fingerprint of sparse context for cross-context collapse checks.

Source code in clickproof/context_vars.py

def context_fingerprint(sparse: Mapping[str, Any]) -> str:
    """Stable fingerprint of sparse context for cross-context collapse checks."""
    parts = []
    for k in sorted(sparse, key=lambda x: _norm_key(str(x))):
        v = sparse[k]
        if v is None or v == "":
            continue
        parts.append(f"{_norm_key(str(k))}={v}")
    return "|".join(parts) if parts else ""

gate_context_variables(decision=None, *, required_sparse_keys=None, extra_sparse_keys=None, batch=None, require_sparse_inventory=True, require_attended=True, refuse_dominant_only=True, refuse_cross_context_collapse=True)

Refuse decisions that ignore sparse context (CVE / GeoReward class).

Rules:

• No decision when inventory required → FAIL_LOUD
• Required sparse keys missing/empty → FAIL_LOUD
• Sparse present but none attended → FAIL (overestimation of dominant cues)
• Dominant-only decision (dominant cues, no sparse attend) → FAIL
• Cross-context collapse (same choice across distinct markets) → FAIL
• Sparse attended, no collapse → PASS

Source code in clickproof/context_vars.py

def gate_context_variables(
    decision: ContextDecision | dict[str, Any] | None = None,
    *,
    required_sparse_keys: Iterable[str] | None = None,
    extra_sparse_keys: Iterable[str] | None = None,
    batch: Sequence[ContextDecision | dict[str, Any]] | None = None,
    require_sparse_inventory: bool = True,
    require_attended: bool = True,
    refuse_dominant_only: bool = True,
    refuse_cross_context_collapse: bool = True,
) -> GateOutcome:
    """Refuse decisions that ignore sparse context (CVE / GeoReward class).

    Rules:

    * No decision when inventory required → **FAIL_LOUD**
    * Required sparse keys missing/empty → **FAIL_LOUD**
    * Sparse present but none attended → **FAIL** (overestimation of dominant cues)
    * Dominant-only decision (dominant cues, no sparse attend) → **FAIL**
    * Cross-context collapse (same choice across distinct markets) → **FAIL**
    * Sparse attended, no collapse → **PASS**
    """
    if decision is None and not batch:
        return GateOutcome(
            ok=False,
            verdict="FAIL_LOUD",
            reason=(
                "CVE/GeoReward: no decision payload — cannot gate contextual "
                "variables without a choice inventory (arXiv 2608.04504)"
            ),
            exit_code=2,
            human_required=True,
        )

    try:
        primary = decision
        if primary is None and batch:
            primary = batch[0]
        assert primary is not None
        d = _as_decision(primary)
        report = analyze_cve(
            d,
            required_sparse_keys=required_sparse_keys,
            extra_sparse_keys=extra_sparse_keys,
            batch=batch,
        )
    except (TypeError, ValueError) as exc:
        return GateOutcome(
            ok=False,
            verdict="FAIL_LOUD",
            reason=f"CVE/GeoReward: invalid decision payload: {exc}",
            exit_code=2,
            human_required=True,
        )

    if require_sparse_inventory:
        has_sparse = bool(report.sparse_keys_present) or bool(
            d.sparse_context and any(v not in (None, "") for v in d.sparse_context.values())
        )
        if required_sparse_keys:
            if report.sparse_keys_missing:
                return GateOutcome(
                    ok=False,
                    verdict="FAIL_LOUD",
                    reason=(
                        f"CVE/GeoReward: required sparse context missing/empty "
                        f"keys={list(report.sparse_keys_missing)[:8]} — "
                        "refuse market/locale decision without context inventory"
                    ),
                    exit_code=2,
                    human_required=True,
                    action=d.choice or d.decision_id,
                )
        elif not has_sparse:
            return GateOutcome(
                ok=False,
                verdict="FAIL_LOUD",
                reason=(
                    "CVE/GeoReward: sparse_context empty — decision-critical "
                    "variables (market/region/locale) not supplied; dominant "
                    "cues alone are not load-bearing"
                ),
                exit_code=2,
                human_required=True,
                action=d.choice or d.decision_id,
            )

    if refuse_cross_context_collapse and report.cross_context_collapse:
        return GateOutcome(
            ok=False,
            verdict="FAIL",
            reason=(
                f"CVE/GeoReward: cross-context collapse — same choice across "
                f"distinct sparse contexts count={len(report.collapsed_contexts)} "
                f"fps={list(report.collapsed_contexts)[:4]} — sparse variables "
                "underestimated (arXiv 2608.04504)"
            ),
            exit_code=1,
            human_required=True,
            action=d.choice or d.decision_id,
        )

    if refuse_dominant_only and report.dominant_only:
        return GateOutcome(
            ok=False,
            verdict="FAIL",
            reason=(
                f"CVE/GeoReward: dominant-cue overestimation decision_id={d.decision_id} "
                f"attended_sparse={list(report.attended_sparse)} "
                f"ignored={list(report.ignored_sparse)[:6]} — refuse collapse to "
                "product/visual-only signal"
            ),
            exit_code=1,
            human_required=True,
            action=d.choice or d.decision_id,
        )

    if require_attended and report.sparse_keys_present and not report.attended_sparse:
        return GateOutcome(
            ok=False,
            verdict="FAIL",
            reason=(
                f"CVE/GeoReward: sparse keys present {list(report.sparse_keys_present)[:6]} "
                f"but none in attended_keys — model ignored decision-critical context"
            ),
            exit_code=1,
            human_required=True,
            action=d.choice or d.decision_id,
        )

    return GateOutcome(
        ok=True,
        verdict="PASS",
        reason=(
            f"CVE/GeoReward ok: decision={d.decision_id} "
            f"sparse={list(report.sparse_keys_present)} "
            f"attended={list(report.attended_sparse)} collapse=false"
        ),
        exit_code=0,
        human_required=False,
        action=d.choice or d.decision_id,
    )

is_sparse_context_key(key, *, extra=None)

True if key is a sparse decision-critical context variable.

Source code in clickproof/context_vars.py

def is_sparse_context_key(
    key: str,
    *,
    extra: Iterable[str] | None = None,
) -> bool:
    """True if key is a sparse decision-critical context variable."""
    k = _norm_key(key)
    if not k:
        return False
    keys = set(DEFAULT_SPARSE_CONTEXT_KEYS)
    if extra:
        keys |= {_norm_key(x) for x in extra}
    if k in keys:
        return True
    return any(k.startswith(s + "_") or k.endswith("_" + s) for s in keys)

to_markdown(facts_scores)

Format facts and scores as a Markdown table.

Parameters:

Name	Type	Description	Default
facts_scores	list[tuple[UIFact, FactScore]]	List of (UIFact, FactScore) pairs.	required

Returns:

Type	Description
str	Markdown string with a header and table.

Source code in clickproof/report.py

def to_markdown(facts_scores: list[tuple[UIFact, FactScore]]) -> str:
    """Format facts and scores as a Markdown table.

    Args:
        facts_scores: List of (UIFact, FactScore) pairs.

    Returns:
        Markdown string with a header and table.
    """
    lines = [
        "## clickproof - UI Behavioral Facts",
        "",
        f"_{len(facts_scores)} fact(s) retrieved_",
        "",
        "| Score | App | Version | Element | Action | Outcome | Obs |",
        "|------:|-----|---------|---------|--------|---------|----:|",
    ]

    for fact, score in facts_scores:
        lines.append(
            f"| {score.score:.2f}"
            f" | {fact.app_name}"
            f" | {fact.app_version}"
            f" | {fact.element}"
            f" | {fact.action}"
            f" | {fact.outcome}"
            f" | {score.observation_count} |"
        )

    return "\n".join(lines)