`popoto.fields.observation`¶

`popoto.fields.observation` ¶

ObservationProtocol + RecallProposal — Outcome-driven memory effects.

Provides lifecycle hooks for passive behavioral inference on memory models. The application layer reports how the agent used retrieved memories; the ORM applies effects atomically.

Three hooks

on_read(instance): Fire when query hydrates an instance. Delegates to AccessTrackerMixin staging.
on_surfaced(instances, reason): Fire when proactive system pushes memories into agent context. Creates RecallProposal entries.
on_context_used(instances, outcome_map): Fire when application reports how the agent responded. Applies effects based on outcome.

Five outcomes

acted: Memory content appeared in agent's response. Strengthen.
dismissed: Agent explicitly ignored/rejected. Weaken.
deferred: Agent didn't address it. No effects, pressure builds.
contradicted: Agent explicitly contradicted. Aggressively weaken.
used: Agent consumed the memory (read + reasoned) but did not act on it in the response. Confirms the staged read and auto-resolves predictions with a moderate error, but does NOT touch ConfidenceField, CyclicDecayField, or DecayingSortedField. Observable distinction from deferred: used records a confirmed-read trace, deferred discards staged reads.

After agent processes memories:¶

outcome_map = { memory1.db_key.redis_key: "acted", memory2.db_key.redis_key: "dismissed", } ObservationProtocol.on_context_used(memories, outcome_map)

`ACTED_CONFIDENCE_SIGNAL = Defaults.ACTED_CONFIDENCE_SIGNAL` `module-attribute` ¶

Confidence signal sent to ConfidenceField on 'acted' outcome. Higher values corroborate the memory more strongly. Optimal range: [0.5, 1.0]. Insensitive within this range (nDCG stable).

`CONTRADICTED_CONFIDENCE_SIGNAL = Defaults.CONTRADICTED_CONFIDENCE_SIGNAL` `module-attribute` ¶

Confidence signal sent to ConfidenceField on 'contradicted' outcome. Lower values contradict the memory more aggressively. Optimal range: [0.05, 0.3]. Insensitive within this range (nDCG stable).

`ACTED_CYCLE_STRENGTHEN_FACTOR = Defaults.ACTED_CYCLE_STRENGTHEN_FACTOR` `module-attribute` ¶

CyclicDecayField amplification factor on 'acted' outcome. CLIFF EFFECT: values < 1.0 cause a 23% nDCG drop in temporal scheduling. Must be >= 1.0. Optimal range: [1.0, 2.0]. Default 1.2 is safe.

`DISMISSED_CYCLE_WEAKEN_FACTOR = Defaults.DISMISSED_CYCLE_WEAKEN_FACTOR` `module-attribute` ¶

CyclicDecayField damping factor on 'dismissed' outcome. Values < 1.0 weaken the cycle amplitude. Optimal range: [0.3, 1.0]. Insensitive within this range.

`CONTRADICTED_CYCLE_WEAKEN_FACTOR = Defaults.CONTRADICTED_CYCLE_WEAKEN_FACTOR` `module-attribute` ¶

CyclicDecayField aggressive damping factor on 'contradicted' outcome. Values < 1.0 weaken the cycle amplitude; lower = more aggressive. Optimal range: [0.3, 0.8]. Insensitive within this range.

`AUTO_DISCHARGE_CONFIDENCE_THRESHOLD = Defaults.AUTO_DISCHARGE_CONFIDENCE_THRESHOLD` `module-attribute` ¶

Below this confidence level, pressure is auto-resolved on contradicted outcome. Very low confidence records should not continue building pressure. Optimal range: [0.05, 0.3]. Insensitive within this range.

`CONFIDENCE_EPSILON = Defaults.CONFIDENCE_EPSILON` `module-attribute` ¶

Internal float-boundary tolerance for the auto-discharge comparison. Confidence values within this epsilon of the threshold are NOT below it. Not user config.

`ObservationProtocol` ¶

Lifecycle hooks for passive behavioral inference on memory models.

All methods are static — the protocol is a stateless coordinator that dispatches effects based on outcome type.

Source code in src/popoto/fields/observation.py

class ObservationProtocol:
    """Lifecycle hooks for passive behavioral inference on memory models.

    All methods are static — the protocol is a stateless coordinator that
    dispatches effects based on outcome type.
    """

    @staticmethod
    def on_read(instance, pipeline=None):
        """Fire when query hydrates an instance. Delegates to AccessTrackerMixin staging.

        If the instance's model uses AccessTrackerMixin, this calls
        ``instance.on_read()``. Otherwise it's a no-op.

        Args:
            instance: A Model instance that was just read from Redis.
            pipeline: Optional Redis pipeline for batch operations.
        """
        if hasattr(instance, "on_read") and callable(instance.on_read):
            instance.on_read(pipeline=pipeline)

    @staticmethod
    def on_surfaced(instances, reason="proactive", partition=None, pipeline=None):
        """Fire when proactive system pushes memories into agent context.

        Creates RecallProposal entries for tracking. Side-effect-free on
        the memories themselves.

        Args:
            instances: List of Model instances being surfaced.
            reason: Why the memories were surfaced. Default "proactive".
            partition: Optional partition key for multi-agent setups.
            pipeline: Optional Redis pipeline for batch operations.
        """
        if not instances:
            return
        RecallProposal.create_batch(
            instances, reason=reason, partition=partition, pipeline=pipeline
        )

    @staticmethod
    def on_context_used(instances, outcome_map, pipeline=None):
        """Fire when application reports how agent responded to surfaced memories.

        For each instance, looks up its outcome in outcome_map and applies
        the corresponding effects atomically.

        Args:
            instances: List of Model instances that were in the agent's context.
            outcome_map: Dict mapping instance Redis keys (str) to outcome
                strings: "acted", "used", "dismissed", "deferred",
                "contradicted". Instances not in the map default to "deferred".
            pipeline: Optional Redis pipeline for batch operations.

        Note:
            This method validates ``outcome_map`` strictly against
            ``VALID_OUTCOMES``. Application-specific outcomes (e.g. a custom
            ``"echoed"`` label) must be coerced to one of the five valid
            values before calling, otherwise a ``ValueError`` is raised.
            See ``docs/features/observation-protocol.md`` (where the
            protocol lives) for guidance on mapping bespoke outcomes into
            the canonical vocabulary.

        Signalling a correction with ``contradicted``:
            ``outcome_map`` is ``{redis_key: outcome}`` and has no slot for a
            second instance, so a caller that knows *what* corrected a memory
            names it by setting the private ``_superseded_by`` attribute on the
            contradicted instance before reporting the outcome::

                stale._superseded_by = corrected
                ObservationProtocol.on_context_used(
                    [stale], {stale.db_key.redis_key: "contradicted"}
                )

            On a model declaring a ValidityField this closes ``stale``'s
            validity interval and writes the supersession edge to ``corrected``
            (issue #580). Without the attribute — and on every model with no
            ValidityField — ``contradicted`` behaves exactly as before. See
            ``_apply_supersession``.

        Unsaved instances:
            An unsaved member of ``instances`` is skipped, for every outcome.
            Its field effects raise internally and are swallowed, it
            contributes no commands, and the remaining members of the batch
            still get their full effects — one unsaved instance never aborts
            the batch (issue #583). The model methods themselves
            (``touch``, ``confirm_access``, ``strengthen_cycle``,
            ``weaken_cycle``, ``resolve_pressure``) are unchanged and still
            raise ``TypeError`` for direct callers; the degradation belongs
            to this protocol layer only.

        Raises:
            ValueError: If any outcome string is not a valid outcome. Being
                unsaved is never a reason this method raises.
        """
        if not instances:
            return

        # Validate all outcomes upfront
        for pk, outcome in outcome_map.items():
            if outcome not in VALID_OUTCOMES:
                raise ValueError(
                    f"Invalid outcome '{outcome}' for key '{pk}'. "
                    f"Valid outcomes: {sorted(VALID_OUTCOMES)}"
                )

        backend_items = [i for i in instances if non_redis_backend(i) is not None]
        if backend_items:
            _on_context_used_on_backend(backend_items, outcome_map, pipeline)
            instances = [i for i in instances if non_redis_backend(i) is None]

        for instance in instances:
            pk = _get_instance_key(instance)
            outcome = outcome_map.get(pk, "deferred")
            _apply_outcome(instance, outcome, pipeline=pipeline)

            # Resolve any pending proposal for this instance
            RecallProposal.resolve(instance, outcome, pipeline=pipeline)

`on_read(instance, pipeline=None)` `staticmethod` ¶

Fire when query hydrates an instance. Delegates to AccessTrackerMixin staging.

If the instance's model uses AccessTrackerMixin, this calls instance.on_read(). Otherwise it's a no-op.

Parameters:

Name	Type	Description	Default
`instance`		A Model instance that was just read from Redis.	required
`pipeline`		Optional Redis pipeline for batch operations.	`None`

Source code in src/popoto/fields/observation.py

@staticmethod
def on_read(instance, pipeline=None):
    """Fire when query hydrates an instance. Delegates to AccessTrackerMixin staging.

    If the instance's model uses AccessTrackerMixin, this calls
    ``instance.on_read()``. Otherwise it's a no-op.

    Args:
        instance: A Model instance that was just read from Redis.
        pipeline: Optional Redis pipeline for batch operations.
    """
    if hasattr(instance, "on_read") and callable(instance.on_read):
        instance.on_read(pipeline=pipeline)

`on_surfaced(instances, reason='proactive', partition=None, pipeline=None)` `staticmethod` ¶

Fire when proactive system pushes memories into agent context.

Creates RecallProposal entries for tracking. Side-effect-free on the memories themselves.

Parameters:

Name	Description	Default
`instances`	List of Model instances being surfaced.	required
`reason`	Why the memories were surfaced. Default "proactive".	`'proactive'`
`partition`	Optional partition key for multi-agent setups.	`None`
`pipeline`	Optional Redis pipeline for batch operations.	`None`

Source code in src/popoto/fields/observation.py

@staticmethod
def on_surfaced(instances, reason="proactive", partition=None, pipeline=None):
    """Fire when proactive system pushes memories into agent context.

    Creates RecallProposal entries for tracking. Side-effect-free on
    the memories themselves.

    Args:
        instances: List of Model instances being surfaced.
        reason: Why the memories were surfaced. Default "proactive".
        partition: Optional partition key for multi-agent setups.
        pipeline: Optional Redis pipeline for batch operations.
    """
    if not instances:
        return
    RecallProposal.create_batch(
        instances, reason=reason, partition=partition, pipeline=pipeline
    )

`on_context_used(instances, outcome_map, pipeline=None)` `staticmethod` ¶

Fire when application reports how agent responded to surfaced memories.

For each instance, looks up its outcome in outcome_map and applies the corresponding effects atomically.

Parameters:

Name	Description	Default
`instances`	List of Model instances that were in the agent's context.	required
`outcome_map`	Dict mapping instance Redis keys (str) to outcome strings: "acted", "used", "dismissed", "deferred", "contradicted". Instances not in the map default to "deferred".	required
`pipeline`	Optional Redis pipeline for batch operations.	`None`

Note

This method validates outcome_map strictly against VALID_OUTCOMES. Application-specific outcomes (e.g. a custom "echoed" label) must be coerced to one of the five valid values before calling, otherwise a ValueError is raised. See docs/features/observation-protocol.md (where the protocol lives) for guidance on mapping bespoke outcomes into the canonical vocabulary.

Signalling a correction with contradicted: outcome_map is {redis_key: outcome} and has no slot for a second instance, so a caller that knows what corrected a memory names it by setting the private _superseded_by attribute on the contradicted instance before reporting the outcome::

    stale._superseded_by = corrected
    ObservationProtocol.on_context_used(
        [stale], {stale.db_key.redis_key: "contradicted"}
    )

On a model declaring a ValidityField this closes ``stale``'s
validity interval and writes the supersession edge to ``corrected``
(issue #580). Without the attribute — and on every model with no
ValidityField — ``contradicted`` behaves exactly as before. See
``_apply_supersession``.

Unsaved instances

An unsaved member of instances is skipped, for every outcome. Its field effects raise internally and are swallowed, it contributes no commands, and the remaining members of the batch still get their full effects — one unsaved instance never aborts the batch (issue #583). The model methods themselves (touch, confirm_access, strengthen_cycle, weaken_cycle, resolve_pressure) are unchanged and still raise TypeError for direct callers; the degradation belongs to this protocol layer only.

Raises:

Type	Description
`ValueError`	If any outcome string is not a valid outcome. Being unsaved is never a reason this method raises.

Source code in src/popoto/fields/observation.py

@staticmethod
def on_context_used(instances, outcome_map, pipeline=None):
    """Fire when application reports how agent responded to surfaced memories.

    For each instance, looks up its outcome in outcome_map and applies
    the corresponding effects atomically.

    Args:
        instances: List of Model instances that were in the agent's context.
        outcome_map: Dict mapping instance Redis keys (str) to outcome
            strings: "acted", "used", "dismissed", "deferred",
            "contradicted". Instances not in the map default to "deferred".
        pipeline: Optional Redis pipeline for batch operations.

    Note:
        This method validates ``outcome_map`` strictly against
        ``VALID_OUTCOMES``. Application-specific outcomes (e.g. a custom
        ``"echoed"`` label) must be coerced to one of the five valid
        values before calling, otherwise a ``ValueError`` is raised.
        See ``docs/features/observation-protocol.md`` (where the
        protocol lives) for guidance on mapping bespoke outcomes into
        the canonical vocabulary.

    Signalling a correction with ``contradicted``:
        ``outcome_map`` is ``{redis_key: outcome}`` and has no slot for a
        second instance, so a caller that knows *what* corrected a memory
        names it by setting the private ``_superseded_by`` attribute on the
        contradicted instance before reporting the outcome::

            stale._superseded_by = corrected
            ObservationProtocol.on_context_used(
                [stale], {stale.db_key.redis_key: "contradicted"}
            )

        On a model declaring a ValidityField this closes ``stale``'s
        validity interval and writes the supersession edge to ``corrected``
        (issue #580). Without the attribute — and on every model with no
        ValidityField — ``contradicted`` behaves exactly as before. See
        ``_apply_supersession``.

    Unsaved instances:
        An unsaved member of ``instances`` is skipped, for every outcome.
        Its field effects raise internally and are swallowed, it
        contributes no commands, and the remaining members of the batch
        still get their full effects — one unsaved instance never aborts
        the batch (issue #583). The model methods themselves
        (``touch``, ``confirm_access``, ``strengthen_cycle``,
        ``weaken_cycle``, ``resolve_pressure``) are unchanged and still
        raise ``TypeError`` for direct callers; the degradation belongs
        to this protocol layer only.

    Raises:
        ValueError: If any outcome string is not a valid outcome. Being
            unsaved is never a reason this method raises.
    """
    if not instances:
        return

    # Validate all outcomes upfront
    for pk, outcome in outcome_map.items():
        if outcome not in VALID_OUTCOMES:
            raise ValueError(
                f"Invalid outcome '{outcome}' for key '{pk}'. "
                f"Valid outcomes: {sorted(VALID_OUTCOMES)}"
            )

    backend_items = [i for i in instances if non_redis_backend(i) is not None]
    if backend_items:
        _on_context_used_on_backend(backend_items, outcome_map, pipeline)
        instances = [i for i in instances if non_redis_backend(i) is None]

    for instance in instances:
        pk = _get_instance_key(instance)
        outcome = outcome_map.get(pk, "deferred")
        _apply_outcome(instance, outcome, pipeline=pipeline)

        # Resolve any pending proposal for this instance
        RecallProposal.resolve(instance, outcome, pipeline=pipeline)

`RecallProposal` ¶

Internal tracking for proactively surfaced memories.

Key pattern: $RP:{ClassName}:pending:{partition} -> ZSET scored by surfaced_at Statuses: pending -> acted | used | dismissed | deferred | contradicted | expired TTL: default 3600s (1 hour). Unresolved proposals treated as deferred.

This is internal ORM infrastructure, not a user-facing Model.

Source code in src/popoto/fields/observation.py

class RecallProposal:
    """Internal tracking for proactively surfaced memories.

    Key pattern: $RP:{ClassName}:pending:{partition} -> ZSET scored by surfaced_at
    Statuses: pending -> acted | used | dismissed | deferred | contradicted | expired
    TTL: default 3600s (1 hour). Unresolved proposals treated as deferred.

    This is internal ORM infrastructure, not a user-facing Model.
    """

    DEFAULT_TTL = 3600

    @classmethod
    def _pending_key(cls, model_class, partition=None):
        """Build the Redis key for pending proposals.

        Args:
            model_class: The Model class (or instance's class).
            partition: Optional partition key.

        Returns:
            str: Redis key like '$RP:ClassName:pending:partition'
        """
        class_name = (
            model_class.__name__
            if isinstance(model_class, type)
            else type(model_class).__name__
        )
        part = partition or "default"
        return f"$RP:{class_name}:pending:{part}"

    @classmethod
    def create_batch(cls, instances, reason="proactive", partition=None, pipeline=None):
        """Create pending proposals for a batch of instances.

        Args:
            instances: List of Model instances being surfaced.
            reason: Why the memories were surfaced.
            partition: Optional partition key.
            pipeline: Optional Redis pipeline for batch operations.
        """
        if not instances:
            return

        now = time.time()
        model_class = type(instances[0])
        backend = non_redis_backend(model_class)
        if backend is not None:
            backend.field_call(
                model_class._meta.spec,
                "_recall",
                "add",
                partition or "default",
                [_get_instance_key(i) for i in instances],
                now,
                uow=_uow_of(pipeline, backend),
            )
            return
        key = cls._pending_key(model_class, partition)

        db = pipeline if pipeline is not None else get_REDIS_DB()

        # ZADD each instance with score=now
        members = {}
        for instance in instances:
            member_key = _get_instance_key(instance)
            # Store member key with reason as the value scored by time
            members[member_key] = now

        if members:
            db.zadd(key, members)

    @classmethod
    def resolve(cls, instance, outcome, partition=None, pipeline=None):
        """Remove a resolved proposal from the pending set.

        Idempotent — returns 0 if already removed (e.g., by expiration).

        Args:
            instance: The Model instance whose proposal to resolve.
            outcome: The outcome string (for logging).
            partition: Optional partition key.
            pipeline: Optional Redis pipeline for batch operations.

        Returns:
            int: Number of members removed (0 or 1).
        """
        model_class = type(instance)
        backend = non_redis_backend(model_class)
        if backend is not None:
            removed = backend.field_call(
                model_class._meta.spec,
                "_recall",
                "remove",
                partition or "default",
                _get_instance_key(instance),
                uow=_uow_of(pipeline, backend),
            )
            return pipeline if pipeline is not None else removed
        key = cls._pending_key(model_class, partition)
        member_key = _get_instance_key(instance)

        if pipeline is not None:
            pipeline.zrem(key, member_key)
            return pipeline
        else:
            return get_REDIS_DB().zrem(key, member_key)

    @classmethod
    def expire_stale(cls, model_class, partition=None, ttl=None, pipeline=None):
        """Remove proposals older than TTL. Returns expired member keys.

        Args:
            model_class: The Model class to check.
            partition: Optional partition key.
            ttl: TTL in seconds. Default DEFAULT_TTL (3600).
            pipeline: Optional Redis pipeline for batch operations.

        Returns:
            list: List of expired member key strings.
        """
        if ttl is None:
            ttl = cls.DEFAULT_TTL

        backend = non_redis_backend(model_class)
        if backend is not None:
            return backend.field_call(
                model_class._meta.spec,
                "_recall",
                "expire",
                partition or "default",
                time.time() - ttl,
                uow=_uow_of(pipeline, backend),
            )
        key = cls._pending_key(model_class, partition)
        cutoff = time.time() - ttl

        # Get members with scores below cutoff (older than TTL)
        expired = get_REDIS_DB().zrangebyscore(key, "-inf", cutoff)

        if expired:
            if pipeline is not None:
                pipeline.zremrangebyscore(key, "-inf", cutoff)
            else:
                get_REDIS_DB().zremrangebyscore(key, "-inf", cutoff)

        # Decode bytes to strings
        return [m.decode() if isinstance(m, bytes) else m for m in expired]

    @classmethod
    def get_pending(cls, model_class, partition=None):
        """Return all pending proposals as (member_key, surfaced_at) pairs.

        Args:
            model_class: The Model class to check.
            partition: Optional partition key.

        Returns:
            list: List of (member_key_str, surfaced_at_float) tuples.
        """
        backend = non_redis_backend(model_class)
        if backend is not None:
            return backend.field_call(
                model_class._meta.spec, "_recall", "pending", partition or "default"
            )
        key = cls._pending_key(model_class, partition)
        results = get_REDIS_DB().zrange(key, 0, -1, withscores=True)
        return [
            (m.decode() if isinstance(m, bytes) else m, score) for m, score in results
        ]

`create_batch(instances, reason='proactive', partition=None, pipeline=None)` `classmethod` ¶

Create pending proposals for a batch of instances.

Parameters:

Name	Description	Default
`instances`	List of Model instances being surfaced.	required
`reason`	Why the memories were surfaced.	`'proactive'`
`partition`	Optional partition key.	`None`
`pipeline`	Optional Redis pipeline for batch operations.	`None`

Source code in src/popoto/fields/observation.py

@classmethod
def create_batch(cls, instances, reason="proactive", partition=None, pipeline=None):
    """Create pending proposals for a batch of instances.

    Args:
        instances: List of Model instances being surfaced.
        reason: Why the memories were surfaced.
        partition: Optional partition key.
        pipeline: Optional Redis pipeline for batch operations.
    """
    if not instances:
        return

    now = time.time()
    model_class = type(instances[0])
    backend = non_redis_backend(model_class)
    if backend is not None:
        backend.field_call(
            model_class._meta.spec,
            "_recall",
            "add",
            partition or "default",
            [_get_instance_key(i) for i in instances],
            now,
            uow=_uow_of(pipeline, backend),
        )
        return
    key = cls._pending_key(model_class, partition)

    db = pipeline if pipeline is not None else get_REDIS_DB()

    # ZADD each instance with score=now
    members = {}
    for instance in instances:
        member_key = _get_instance_key(instance)
        # Store member key with reason as the value scored by time
        members[member_key] = now

    if members:
        db.zadd(key, members)

`resolve(instance, outcome, partition=None, pipeline=None)` `classmethod` ¶

Remove a resolved proposal from the pending set.

Idempotent — returns 0 if already removed (e.g., by expiration).

Parameters:

Name	Description	Default
`instance`	The Model instance whose proposal to resolve.	required
`outcome`	The outcome string (for logging).	required
`partition`	Optional partition key.	`None`
`pipeline`	Optional Redis pipeline for batch operations.	`None`

Returns:

Name	Type	Description
`int`		Number of members removed (0 or 1).

Source code in src/popoto/fields/observation.py

@classmethod
def resolve(cls, instance, outcome, partition=None, pipeline=None):
    """Remove a resolved proposal from the pending set.

    Idempotent — returns 0 if already removed (e.g., by expiration).

    Args:
        instance: The Model instance whose proposal to resolve.
        outcome: The outcome string (for logging).
        partition: Optional partition key.
        pipeline: Optional Redis pipeline for batch operations.

    Returns:
        int: Number of members removed (0 or 1).
    """
    model_class = type(instance)
    backend = non_redis_backend(model_class)
    if backend is not None:
        removed = backend.field_call(
            model_class._meta.spec,
            "_recall",
            "remove",
            partition or "default",
            _get_instance_key(instance),
            uow=_uow_of(pipeline, backend),
        )
        return pipeline if pipeline is not None else removed
    key = cls._pending_key(model_class, partition)
    member_key = _get_instance_key(instance)

    if pipeline is not None:
        pipeline.zrem(key, member_key)
        return pipeline
    else:
        return get_REDIS_DB().zrem(key, member_key)

`expire_stale(model_class, partition=None, ttl=None, pipeline=None)` `classmethod` ¶

Remove proposals older than TTL. Returns expired member keys.

Parameters:

Name	Description	Default
`model_class`	The Model class to check.	required
`partition`	Optional partition key.	`None`
`ttl`	TTL in seconds. Default DEFAULT_TTL (3600).	`None`
`pipeline`	Optional Redis pipeline for batch operations.	`None`

Returns:

Name	Type	Description
`list`		List of expired member key strings.

Source code in src/popoto/fields/observation.py

@classmethod
def expire_stale(cls, model_class, partition=None, ttl=None, pipeline=None):
    """Remove proposals older than TTL. Returns expired member keys.

    Args:
        model_class: The Model class to check.
        partition: Optional partition key.
        ttl: TTL in seconds. Default DEFAULT_TTL (3600).
        pipeline: Optional Redis pipeline for batch operations.

    Returns:
        list: List of expired member key strings.
    """
    if ttl is None:
        ttl = cls.DEFAULT_TTL

    backend = non_redis_backend(model_class)
    if backend is not None:
        return backend.field_call(
            model_class._meta.spec,
            "_recall",
            "expire",
            partition or "default",
            time.time() - ttl,
            uow=_uow_of(pipeline, backend),
        )
    key = cls._pending_key(model_class, partition)
    cutoff = time.time() - ttl

    # Get members with scores below cutoff (older than TTL)
    expired = get_REDIS_DB().zrangebyscore(key, "-inf", cutoff)

    if expired:
        if pipeline is not None:
            pipeline.zremrangebyscore(key, "-inf", cutoff)
        else:
            get_REDIS_DB().zremrangebyscore(key, "-inf", cutoff)

    # Decode bytes to strings
    return [m.decode() if isinstance(m, bytes) else m for m in expired]

`get_pending(model_class, partition=None)` `classmethod` ¶

Return all pending proposals as (member_key, surfaced_at) pairs.

Parameters:

Name	Type	Description	Default
`model_class`		The Model class to check.	required
`partition`		Optional partition key.	`None`

Returns:

Name	Type	Description
`list`		List of (member_key_str, surfaced_at_float) tuples.

Source code in src/popoto/fields/observation.py

@classmethod
def get_pending(cls, model_class, partition=None):
    """Return all pending proposals as (member_key, surfaced_at) pairs.

    Args:
        model_class: The Model class to check.
        partition: Optional partition key.

    Returns:
        list: List of (member_key_str, surfaced_at_float) tuples.
    """
    backend = non_redis_backend(model_class)
    if backend is not None:
        return backend.field_call(
            model_class._meta.spec, "_recall", "pending", partition or "default"
        )
    key = cls._pending_key(model_class, partition)
    results = get_REDIS_DB().zrange(key, 0, -1, withscores=True)
    return [
        (m.decode() if isinstance(m, bytes) else m, score) for m, score in results
    ]

popoto.fields.observation¶