popoto.fields.cyclic_decay_field

popoto.fields.cyclic_decay_field

CyclicDecayField — Temporal Rhythms + Homeostatic Pressure.

Extends DecayingSortedField with two additional temporal forces computed atomically in the same Lua script:

1. Cyclical resonance: Periodic boosts following cosine curves. A record about Q1 renewals resurfaces every January.

2. Homeostatic pressure: Urgency that builds linearly over time when an item goes unresolved. Discharged by resolve_pressure().

The effective score is: decay + cyclic_resonance + pressure

When cycles=[] and pressure_rate=0.0, behavior is identical to DecayingSortedField (the Lua script short-circuits on nil HGET lookups).

Companion Redis hashes store per-member cycle and pressure data

• $CyclicDecayF:{Model}:{field}:{partitions}:cycles — msgpack cycle tuples, [period, amplitude, phase] or, once a member has saved under this field (#698), [period, amplitude, phase, declared_baseline]. The optional 4th slot records the declared amplitude in force when the entry was last written, letting on_save tell "the developer edited the declaration" from "learning diverged from the declaration" — see CyclicDecayField.on_save.
• $CyclicDecayF:{Model}:{field}:{partitions}:pressure — msgpack pressure dict

Example

class Directive(Model): agent_id = KeyField() content = Field(type=str) relevance = CyclicDecayField( decay_rate=0.5, cycles=[(TemporalPeriod.QUARTERLY, 5.0, 0)], pressure_rate=0.1, )

top = Directive.query.filter(agent_id="agent-1").top_by_decay("relevance", n=10) directive.resolve_pressure("relevance")

CyclicDecayField

Bases: DecayingSortedField

A DecayingSortedField with cyclical resonance and homeostatic pressure.

Extends the parent's power-law decay with two additional forces:

1. Cyclical resonance via cycles parameter: each cycle is a (period, amplitude, phase) tuple defining a cosine curve. The resonance contribution is amplitude * cos(2*pi*(now-phase)/period).

2. Homeostatic pressure via pressure_rate: linearly increasing urgency. Pressure = pressure_rate * unresolved_days. Reset by calling model.resolve_pressure(field_name).

When cycles=[] and pressure_rate=0.0, behavior is identical to DecayingSortedField.

Ranking is deterministic: equal effective-scored members are ordered by member key (redis_key) ascending, byte-wise, broken inside the Lua script before top-N truncation.

Parameters:

Name	Type	Description	Default
decay_rate		Controls how fast scores drop. Higher = faster decay. Default 0.5. Must be > 0. (Inherited from DecayingSortedField.)	required
base_score_field		Name of a companion field whose value multiplies the decay curve. When None, base score is 1.0. (Inherited.)	required
cycles		List of (period, amplitude, phase) tuples defining cyclical resonance curves. period is in seconds (use TemporalPeriod constants). amplitude is the peak boost. phase is a time offset in seconds. Default [].	required
pressure_rate		Rate at which urgency builds per unresolved day. Default 0.0 (no pressure). Must be >= 0.	required
partition_by		Partition the sorted set by key field values. Inherited from SortedFieldMixin.	required

Example

from popoto.fields.constants import TemporalPeriod

class Directive(Model): agent_id = KeyField() content = Field(type=str) relevance = CyclicDecayField( decay_rate=0.5, cycles=[(TemporalPeriod.QUARTERLY, 5.0, 0)], pressure_rate=0.1, )

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

class CyclicDecayField(DecayingSortedField):
    """A DecayingSortedField with cyclical resonance and homeostatic pressure.

    Extends the parent's power-law decay with two additional forces:

    1. **Cyclical resonance** via ``cycles`` parameter: each cycle is a
       ``(period, amplitude, phase)`` tuple defining a cosine curve.
       The resonance contribution is ``amplitude * cos(2*pi*(now-phase)/period)``.

    2. **Homeostatic pressure** via ``pressure_rate``: linearly increasing
       urgency. Pressure = ``pressure_rate * unresolved_days``.
       Reset by calling ``model.resolve_pressure(field_name)``.

    When ``cycles=[]`` and ``pressure_rate=0.0``, behavior is identical
    to ``DecayingSortedField``.

    Ranking is deterministic: equal effective-scored members are ordered
    by member key (redis_key) ascending, byte-wise, broken inside the Lua
    script before top-N truncation.

    Args:
        decay_rate: Controls how fast scores drop. Higher = faster decay.
            Default 0.5. Must be > 0. (Inherited from DecayingSortedField.)
        base_score_field: Name of a companion field whose value multiplies
            the decay curve. When None, base score is 1.0. (Inherited.)
        cycles: List of ``(period, amplitude, phase)`` tuples defining
            cyclical resonance curves. ``period`` is in seconds (use
            ``TemporalPeriod`` constants). ``amplitude`` is the peak boost.
            ``phase`` is a time offset in seconds. Default ``[]``.
        pressure_rate: Rate at which urgency builds per unresolved day.
            Default ``0.0`` (no pressure). Must be >= 0.
        partition_by: Partition the sorted set by key field values.
            Inherited from SortedFieldMixin.

    Example:
        from popoto.fields.constants import TemporalPeriod

        class Directive(Model):
            agent_id = KeyField()
            content = Field(type=str)
            relevance = CyclicDecayField(
                decay_rate=0.5,
                cycles=[(TemporalPeriod.QUARTERLY, 5.0, 0)],
                pressure_rate=0.1,
            )
    """

    # Export/import: two companion hashes hold state a plain re-save cannot
    # reconstruct. Per-member cycle amplitudes are LEARNED (mutated by
    # strengthen_cycle / weaken_cycle) and diverge from the class-level
    # ``cycles`` defaults; ``pressure.last_resolved`` is genuine independent
    # state whose age is the whole point of homeostatic pressure.
    roundtrip_policy: str = "carry"

    @classmethod
    def export_state(cls, model_instance, field_name, field_value, **kwargs):
        """Export the per-member cycles and pressure companion data.

        Returns:
            ``{"cycles": [[period, amplitude, phase], ...],
                "pressure": {"rate": float, "last_resolved": float}}``
            with either key omitted when that companion hash has no entry for
            this instance, or ``None`` when neither does.
        """
        field = model_instance._meta.fields.get(field_name)
        if not isinstance(field, CyclicDecayField):
            return None

        member_key = model_instance.db_key.redis_key
        state = {}

        cycles_raw = get_REDIS_DB().hget(
            field.get_cycles_hash_key(model_instance, field_name), member_key
        )
        if cycles_raw:
            try:
                cycles = msgpack.unpackb(cycles_raw, raw=False)
            except Exception:
                logger.warning(
                    f"Could not decode cycles data for {member_key}; "
                    f"skipping cycles export of {field_name}"
                )
                cycles = None
            if isinstance(cycles, (list, tuple)):
                normalized = []
                for cycle in cycles:
                    cycle = list(cycle)
                    # #699 Risk 1: integral amplitudes/phases round-trip from
                    # the Lua scripts as msgpack integers (Lua 5.1 has one
                    # number type). Coerce whole-number slots back to float
                    # here so exporters see the same types as pre-#699;
                    # period (slot 0) is left alone since it may
                    # legitimately be a string. Only int values are
                    # touched — floats, strings and bools pass through, so
                    # a malformed entry can never raise out of this read
                    # path.
                    if (
                        len(cycle) > 1
                        and isinstance(cycle[1], int)
                        and not isinstance(cycle[1], bool)
                    ):
                        cycle[1] = float(cycle[1])
                    if (
                        len(cycle) > 2
                        and isinstance(cycle[2], int)
                        and not isinstance(cycle[2], bool)
                    ):
                        cycle[2] = float(cycle[2])
                    normalized.append(cycle)
                state["cycles"] = normalized

        pressure_raw = get_REDIS_DB().hget(
            field.get_pressure_hash_key(model_instance, field_name), member_key
        )
        if pressure_raw:
            try:
                pressure = msgpack.unpackb(pressure_raw, raw=False)
            except Exception:
                logger.warning(
                    f"Could not decode pressure data for {member_key}; "
                    f"skipping pressure export of {field_name}"
                )
                pressure = None
            if isinstance(pressure, dict):
                state["pressure"] = {
                    "rate": float(pressure.get("rate", 0.0) or 0.0),
                    "last_resolved": float(pressure.get("last_resolved", 0.0) or 0.0),
                }

        return state or None

    @classmethod
    def import_state(cls, model_instance, field_name, state, **kwargs):
        """Restore per-member cycles and pressure companion data after import.

        Ordering note -- the transfer driver calls ``import_state`` *after*
        ``save()``, and that is still required, though since #679 the reason
        has narrowed.

        ``on_save`` no longer clobbers learned amplitudes; it preserves
        whatever is already stored for the member. But on an import the record
        is *new*, so there is nothing stored yet and ``on_save`` legitimately
        writes the class-level defaults -- and it seeds
        ``pressure.last_resolved`` to ``now`` whenever the entry is fresh,
        which is exactly the value the import must replace. So these writes
        still have to land on top of ``on_save``'s, and inverting the order
        would still silently discard both the imported amplitudes and the
        accumulated pressure age.

        **Cycle entries deliberately do not carry the declared-baseline slot
        (#698).** A stored cycle entry may have a 4th element,
        ``declared_baseline`` — the declared amplitude in force in the
        deployment that *wrote* the entry (see ``on_save``). That is
        deployment-local by definition, so this method rebuilds every cycle as
        a 3-element ``[period, amplitude, phase]`` and never carries slot 3
        through, even when the exported state has it. Carrying the exporter's
        baseline into a target whose ``field.cycles`` declares a different
        amplitude for that period would make the first post-import ``save()``
        see ``baseline != declared`` and fire a reset — destroying exactly the
        learned amplitude ``roundtrip_policy = "carry"`` exists to preserve,
        with no import-time signal. Dropping it instead means the imported
        entry is "baseline unknown": it preserves the learned amplitude
        unconditionally and acquires a fresh baseline, from the *importing*
        deployment's declaration, on its next ordinary save. The cost is one
        missed detection window if the declaration was also edited between
        export and that first post-import save — the same trade already
        accepted for a pre-upgrade record (see ``on_save``'s Risk 2 in
        docs/plans/sdlc-698.md) and self-healing the same way.
        """
        if not state:
            return None

        field = model_instance._meta.fields.get(field_name)
        if not isinstance(field, CyclicDecayField):
            return None

        member_key = model_instance.db_key.redis_key

        cycles = state.get("cycles")
        if cycles:
            normalized = []
            for cycle in cycles:
                cycle = list(cycle)
                period, amplitude = cycle[0], cycle[1]
                phase = cycle[2] if len(cycle) > 2 else 0
                # Deliberately 3-element, not widened to carry a 4th slot
                # (#698 / see the docstring above): the declared baseline is
                # deployment-local and must never come from the exporter.
                normalized.append([period, amplitude, phase])
            get_REDIS_DB().hset(
                field.get_cycles_hash_key(model_instance, field_name),
                member_key,
                msgpack.packb(normalized),
            )

        pressure = state.get("pressure")
        if pressure:
            get_REDIS_DB().hset(
                field.get_pressure_hash_key(model_instance, field_name),
                member_key,
                msgpack.packb(
                    {
                        "rate": float(pressure.get("rate", 0.0) or 0.0),
                        "last_resolved": float(
                            pressure.get("last_resolved", 0.0) or 0.0
                        ),
                    }
                ),
            )
        return None

    def __init__(self, **kwargs):
        self.cycles = kwargs.pop("cycles", [])
        self.pressure_rate = kwargs.pop("pressure_rate", 0.0)

        # Validate cycles
        for cycle in self.cycles:
            if len(cycle) < 2 or len(cycle) > 3:
                raise ModelException(
                    f"Each cycle must be (period, amplitude) or "
                    f"(period, amplitude, phase), got {cycle}"
                )
            period, amplitude = cycle[0], cycle[1]
            if period <= 0:
                raise ModelException(f"Cycle period must be > 0 (got {period})")
            if amplitude < 0:
                raise ModelException(f"Cycle amplitude must be >= 0 (got {amplitude})")

        # Validate pressure_rate
        if self.pressure_rate < 0:
            raise ModelException(
                f"pressure_rate must be >= 0 (got {self.pressure_rate})"
            )

        super().__init__(**kwargs)

    def get_cycles_hash_key(self, model_instance, field_name):
        """Build the Redis key for the cycles companion hash.

        Public API for external callers that need direct Redis access to
        cycle data (e.g., bulk inspection, custom cycle updates, monitoring).

        Pattern: $CyclicDecayF:{Model}:{field}:{partitions}:cycles
        """
        ss_key = self.get_partitioned_sortedset_db_key(model_instance, field_name)
        return ss_key.redis_key + ":cycles"

    def get_pressure_hash_key(self, model_instance, field_name):
        """Build the Redis key for the pressure companion hash.

        Public API for external callers that need direct Redis access to
        pressure data (e.g., bulk pressure resets, monitoring dashboards).

        Pattern: $CyclicDecayF:{Model}:{field}:{partitions}:pressure
        """
        ss_key = self.get_partitioned_sortedset_db_key(model_instance, field_name)
        return ss_key.redis_key + ":pressure"

    def rank_decayed(
        self,
        zset_key: str,
        *,
        now: float,
        n: Optional[int] = None,
        confidence: Optional[tuple[str, str, str]] = None,
        validity: Optional[tuple[str, str, str]] = None,
        decay_rate: Optional[float] = None,
        base_score_field: Optional[str] = None,
    ) -> "list[Any]":
        """Evaluate the cyclic decay script over one sorted set (#648).

        Overrides :meth:`DecayingSortedField.rank_decayed` because this fork of
        the decay math uses an **incompatible KEYS layout**: cycles at
        ``KEYS[2]`` and pressure at ``KEYS[3]``, which pushes the confidence
        hash to ``KEYS[4]``. Both scripts carry a comment forbidding a "unify"
        on ``KEYS[2]`` -- reusing index 2 here would ``cmsgpack.unpack`` the
        cycles array as a confidence dict, a silent corrupt read rather than a
        clean crash. Keeping the two layouts in two class bodies, rather than
        behind a flag in one, is what makes that mistake unavailable instead of
        merely discouraged.

        The companion hash keys are the partition ZSET key plus a suffix (the
        same derivation as :meth:`get_cycles_hash_key` /
        :meth:`get_cycles_hash_key_from_parts`), so they follow from
        ``zset_key`` alone. The confidence hash does not -- it lives under its
        own ``$ConfidencF:`` prefix -- so it arrives resolved in ``confidence``.

        ``validity`` is accepted and **deliberately ignored**: ``KEYS`` 1-4 are
        taken here and the script's header forbids renumbering, so this script
        has no validity gate. That gap is an explicit No-Go, pinned by
        ``tests/test_validity_field.py::TestCyclicDecayGatingGap`` and recorded
        under "Known limitations" in
        ``docs/features/validity-and-supersession.md``. The parameter is kept in
        the signature so callers stay polymorphic; if you ever gate this script,
        update all three places.

        Args and return value are otherwise as
        :meth:`DecayingSortedField.rank_decayed`.
        """
        conf_hash_key, conf_s, conf_c0 = (
            MODULATION_DISABLED if confidence is None else confidence
        )
        if n is None:
            n = int(get_REDIS_DB().zcard(zset_key))
            if not n:
                return []
        effective_rate = self.decay_rate if decay_rate is None else decay_rate
        if base_score_field is None:
            base_score_field = self.base_score_field or ""

        return run_lua(
            get_REDIS_DB(),
            CYCLIC_DECAY_LUA,
            # numkeys: zset + cycles + pressure + confidence (KEYS[4]).
            # Passing the confidence key without bumping this would shunt it
            # into ARGV and silently disable modulation.
            4,
            zset_key,
            zset_key + ":cycles",
            zset_key + ":pressure",
            conf_hash_key,
            str(now),
            str(effective_rate),
            str(n),
            base_score_field,
            conf_s,
            conf_c0,
        )

    @classmethod
    def get_cycles_hash_key_from_parts(cls, model_class, field_name, *partition_values):
        """Build cycles hash key from model class and explicit partition values.

        Public API for query paths and external callers that have partition
        values but not a model instance.
        """
        ss_key = cls.get_sortedset_db_key(model_class, field_name, *partition_values)
        return ss_key.redis_key + ":cycles"

    @classmethod
    def get_pressure_hash_key_from_parts(
        cls, model_class, field_name, *partition_values
    ):
        """Build pressure hash key from model class and explicit partition values.

        Public API for query paths and external callers that have partition
        values but not a model instance.
        """
        ss_key = cls.get_sortedset_db_key(model_class, field_name, *partition_values)
        return ss_key.redis_key + ":pressure"

    @classmethod
    def on_save(cls, model_instance, field_name, field_value, pipeline=None, **kwargs):
        """Store timestamp (parent) then store cycle/pressure companion data.

        Both companion hashes follow the same rule: **declared parameters are
        refreshed from the field; learned state is preserved.**

        For cycles, ``period`` and ``phase`` are declarative and re-read from
        ``field.cycles`` on every save. ``amplitude`` is learned (mutated by
        ``strengthen_cycle`` / ``weaken_cycle``) but is now merged with a
        **three-way rule** rather than a two-way one (#698): each stored entry
        may carry an optional 4th slot, ``declared_baseline`` — the declared
        amplitude that was in force the last time ``on_save`` wrote this entry.
        Comparing the *incoming* declared amplitude against that baseline (not
        against the learned value) distinguishes "the developer edited the
        declaration" from "learning diverged from the declaration":

        - baseline absent (a legacy 3-element entry, or a non-numeric slot 3) →
          "baseline unknown" → preserve the learned amplitude exactly as before
          #698, and record the declared amplitude as the new baseline.
        - baseline equals the incoming declared amplitude → the declaration has
          not moved → preserve the learned amplitude, re-record the same
          baseline.
        - baseline differs from the incoming declared amplitude → the developer
          edited the declaration → **discard the learned amplitude**, adopt the
          declared value, record it as the new baseline, and emit one
          ``logger.info`` naming the model, field, member key, period, old
          baseline, new declared value and the discarded learned amplitude.

        The comparison is exact float equality, never a tolerance — both sides
        are the same Python float, round-tripped through msgpack, which
        preserves IEEE doubles exactly. Stored cycles are matched to declared
        cycles by period, FIFO within duplicate periods, with the amplitude and
        its baseline popped together as one decision; a declared period with
        nothing stored takes the declared amplitude (baseline unknown), and a
        stored period no longer declared is dropped. Before #679 this branch
        overwrote the whole entry with the declared defaults, silently erasing
        everything the strengthen/weaken calls had accumulated; #679 then made
        the merge unconditional the other way, so an edited declaration could
        never win. #698 adds the missing third input (the baseline) so the
        merge can tell the two cases apart. A record with no recorded baseline
        needs two saves to honor an edit made after this change ships: the
        first save records the baseline, and only an edit made after that save
        is detected — see the field's module docs / docs/features/cyclic-decay-field.md.

        For pressure, ``rate`` is declarative and ``last_resolved`` is learned:
        on first save (no existing entry) the full dict is written with
        ``last_resolved=now``; on subsequent saves only the rate is updated.

        Both companion hashes are read-modified-written by one Lua script,
        ``CYCLES_MERGE_LUA`` (#699), run eagerly against the live connection
        regardless of a caller-supplied ``pipeline`` — the merge decision
        (which amplitude/rate wins) has to be known synchronously to emit the
        #698 reset log, and a pipeline-queued script's return value is not
        available until ``execute()``, which ``save()`` does not surface.
        Running it eagerly is also what closes the race this exists for:
        without it, two concurrent ``save()`` calls — or a ``save()`` racing
        ``strengthen_cycle()``/``weaken_cycle()``/``resolve_pressure()`` —
        each read-then-write the hash independently and the last writer
        clobbers the other's update; the script makes the whole
        read-decide-write sequence one atomic server-side step.
        """
        # Call parent to store timestamp in sorted set
        result = super().on_save(
            model_instance, field_name, field_value, pipeline=pipeline, **kwargs
        )

        field = model_instance._meta.fields[field_name]
        if not isinstance(field, CyclicDecayField):
            return result

        member_key = model_instance.db_key.redis_key
        cycles_hash_key = field.get_cycles_hash_key(model_instance, field_name)
        pressure_hash_key = field.get_pressure_hash_key(model_instance, field_name)

        # Build ARGV: member, N, then N (period, amplitude, phase) triples,
        # then pressure_rate, then now. Periods may be TemporalPeriod
        # aliases (strings) or raw numbers — EVAL only accepts string/number
        # ARGV, so every slot is stringified; the script's coerce_period()
        # converts numeric-looking strings back before storage (B1).
        argv: list[Any] = [member_key, str(len(field.cycles))]
        for cycle in field.cycles:
            period, declared_amplitude = cycle[0], cycle[1]
            phase = cycle[2] if len(cycle) > 2 else 0
            argv.extend([str(period), str(declared_amplitude), str(phase)])
        argv.append(str(field.pressure_rate))
        argv.append(str(time.time()))

        packed_report = run_lua(
            get_REDIS_DB(),
            CYCLES_MERGE_LUA,
            2,
            cycles_hash_key,
            pressure_hash_key,
            *argv,
        )
        decode_failed, resets = msgpack.unpackb(packed_report, raw=False)

        if decode_failed:
            # Mirror export_state's handler: #679 exists because this state
            # was destroyed silently. Do not add a second mute path — fall
            # back to declared defaults, but say so.
            logger.warning(
                f"Could not decode cycles data for {member_key}; "
                f"falling back to declared amplitudes for {field_name}"
            )

        for period, old_baseline, declared_amplitude, learned_amplitude in resets:
            # cmsgpack packs an integral Lua number as a msgpack integer
            # (Lua 5.1 has one number type), so a whole-number amplitude or
            # baseline round-trips as int rather than float. Coerce back to
            # float here so the log line matches the pre-#699 Python-float
            # formatting; period is left alone since it may legitimately be
            # a non-numeric string. Coercion is display-only and defensive:
            # a hand-written/migrated payload can carry a non-numeric value
            # that still decodes as valid msgpack, and the pre-#699 code
            # logged such values fine — never raise out of save() here.
            try:
                old_baseline = float(old_baseline)
                declared_amplitude = float(declared_amplitude)
                learned_amplitude = float(learned_amplitude)
            except (TypeError, ValueError):
                pass
            # The developer edited the declared amplitude — the declaration
            # wins. The learned amplitude is discarded by design (#698); this
            # destroys real state, so it is logged loudly.
            logger.info(
                f"CyclicDecayField declared amplitude changed for "
                f"{model_instance.__class__.__name__}.{field_name} "
                f"member={member_key} period={period!r}: "
                f"declared baseline {old_baseline!r} -> "
                f"{declared_amplitude!r}; discarded learned "
                f"amplitude {learned_amplitude!r}"
            )

        return result

    @classmethod
    def on_delete(
        cls, model_instance, field_name, field_value, pipeline=None, **kwargs
    ):
        """Remove companion hash entries then delegate to parent."""
        field = model_instance._meta.fields[field_name]

        if isinstance(field, CyclicDecayField):
            member_key = (
                kwargs.get("saved_redis_key") or model_instance.db_key.redis_key
            )
            cycles_hash_key = field.get_cycles_hash_key(model_instance, field_name)
            pressure_hash_key = field.get_pressure_hash_key(model_instance, field_name)

            db = (
                pipeline
                if isinstance(pipeline, redis.client.Pipeline)
                else get_REDIS_DB()
            )
            db.hdel(cycles_hash_key, member_key)
            db.hdel(pressure_hash_key, member_key)

        # Delegate to parent for sorted set cleanup
        return super().on_delete(
            model_instance, field_name, field_value, pipeline=pipeline, **kwargs
        )

export_state(model_instance, field_name, field_value, **kwargs) classmethod

Export the per-member cycles and pressure companion data.

Returns:

Type	Description
	{"cycles": [[period, amplitude, phase], ...], "pressure": {"rate": float, "last_resolved": float}}
	with either key omitted when that companion hash has no entry for
	this instance, or None when neither does.

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

@classmethod
def export_state(cls, model_instance, field_name, field_value, **kwargs):
    """Export the per-member cycles and pressure companion data.

    Returns:
        ``{"cycles": [[period, amplitude, phase], ...],
            "pressure": {"rate": float, "last_resolved": float}}``
        with either key omitted when that companion hash has no entry for
        this instance, or ``None`` when neither does.
    """
    field = model_instance._meta.fields.get(field_name)
    if not isinstance(field, CyclicDecayField):
        return None

    member_key = model_instance.db_key.redis_key
    state = {}

    cycles_raw = get_REDIS_DB().hget(
        field.get_cycles_hash_key(model_instance, field_name), member_key
    )
    if cycles_raw:
        try:
            cycles = msgpack.unpackb(cycles_raw, raw=False)
        except Exception:
            logger.warning(
                f"Could not decode cycles data for {member_key}; "
                f"skipping cycles export of {field_name}"
            )
            cycles = None
        if isinstance(cycles, (list, tuple)):
            normalized = []
            for cycle in cycles:
                cycle = list(cycle)
                # #699 Risk 1: integral amplitudes/phases round-trip from
                # the Lua scripts as msgpack integers (Lua 5.1 has one
                # number type). Coerce whole-number slots back to float
                # here so exporters see the same types as pre-#699;
                # period (slot 0) is left alone since it may
                # legitimately be a string. Only int values are
                # touched — floats, strings and bools pass through, so
                # a malformed entry can never raise out of this read
                # path.
                if (
                    len(cycle) > 1
                    and isinstance(cycle[1], int)
                    and not isinstance(cycle[1], bool)
                ):
                    cycle[1] = float(cycle[1])
                if (
                    len(cycle) > 2
                    and isinstance(cycle[2], int)
                    and not isinstance(cycle[2], bool)
                ):
                    cycle[2] = float(cycle[2])
                normalized.append(cycle)
            state["cycles"] = normalized

    pressure_raw = get_REDIS_DB().hget(
        field.get_pressure_hash_key(model_instance, field_name), member_key
    )
    if pressure_raw:
        try:
            pressure = msgpack.unpackb(pressure_raw, raw=False)
        except Exception:
            logger.warning(
                f"Could not decode pressure data for {member_key}; "
                f"skipping pressure export of {field_name}"
            )
            pressure = None
        if isinstance(pressure, dict):
            state["pressure"] = {
                "rate": float(pressure.get("rate", 0.0) or 0.0),
                "last_resolved": float(pressure.get("last_resolved", 0.0) or 0.0),
            }

    return state or None

import_state(model_instance, field_name, state, **kwargs) classmethod

Restore per-member cycles and pressure companion data after import.

Ordering note -- the transfer driver calls import_state after save(), and that is still required, though since #679 the reason has narrowed.

on_save no longer clobbers learned amplitudes; it preserves whatever is already stored for the member. But on an import the record is new, so there is nothing stored yet and on_save legitimately writes the class-level defaults -- and it seeds pressure.last_resolved to now whenever the entry is fresh, which is exactly the value the import must replace. So these writes still have to land on top of on_save's, and inverting the order would still silently discard both the imported amplitudes and the accumulated pressure age.

Cycle entries deliberately do not carry the declared-baseline slot (#698). A stored cycle entry may have a 4th element, declared_baseline — the declared amplitude in force in the deployment that wrote the entry (see on_save). That is deployment-local by definition, so this method rebuilds every cycle as a 3-element [period, amplitude, phase] and never carries slot 3 through, even when the exported state has it. Carrying the exporter's baseline into a target whose field.cycles declares a different amplitude for that period would make the first post-import save() see baseline != declared and fire a reset — destroying exactly the learned amplitude roundtrip_policy = "carry" exists to preserve, with no import-time signal. Dropping it instead means the imported entry is "baseline unknown": it preserves the learned amplitude unconditionally and acquires a fresh baseline, from the importing deployment's declaration, on its next ordinary save. The cost is one missed detection window if the declaration was also edited between export and that first post-import save — the same trade already accepted for a pre-upgrade record (see on_save's Risk 2 in docs/plans/sdlc-698.md) and self-healing the same way.

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

@classmethod
def import_state(cls, model_instance, field_name, state, **kwargs):
    """Restore per-member cycles and pressure companion data after import.

    Ordering note -- the transfer driver calls ``import_state`` *after*
    ``save()``, and that is still required, though since #679 the reason
    has narrowed.

    ``on_save`` no longer clobbers learned amplitudes; it preserves
    whatever is already stored for the member. But on an import the record
    is *new*, so there is nothing stored yet and ``on_save`` legitimately
    writes the class-level defaults -- and it seeds
    ``pressure.last_resolved`` to ``now`` whenever the entry is fresh,
    which is exactly the value the import must replace. So these writes
    still have to land on top of ``on_save``'s, and inverting the order
    would still silently discard both the imported amplitudes and the
    accumulated pressure age.

    **Cycle entries deliberately do not carry the declared-baseline slot
    (#698).** A stored cycle entry may have a 4th element,
    ``declared_baseline`` — the declared amplitude in force in the
    deployment that *wrote* the entry (see ``on_save``). That is
    deployment-local by definition, so this method rebuilds every cycle as
    a 3-element ``[period, amplitude, phase]`` and never carries slot 3
    through, even when the exported state has it. Carrying the exporter's
    baseline into a target whose ``field.cycles`` declares a different
    amplitude for that period would make the first post-import ``save()``
    see ``baseline != declared`` and fire a reset — destroying exactly the
    learned amplitude ``roundtrip_policy = "carry"`` exists to preserve,
    with no import-time signal. Dropping it instead means the imported
    entry is "baseline unknown": it preserves the learned amplitude
    unconditionally and acquires a fresh baseline, from the *importing*
    deployment's declaration, on its next ordinary save. The cost is one
    missed detection window if the declaration was also edited between
    export and that first post-import save — the same trade already
    accepted for a pre-upgrade record (see ``on_save``'s Risk 2 in
    docs/plans/sdlc-698.md) and self-healing the same way.
    """
    if not state:
        return None

    field = model_instance._meta.fields.get(field_name)
    if not isinstance(field, CyclicDecayField):
        return None

    member_key = model_instance.db_key.redis_key

    cycles = state.get("cycles")
    if cycles:
        normalized = []
        for cycle in cycles:
            cycle = list(cycle)
            period, amplitude = cycle[0], cycle[1]
            phase = cycle[2] if len(cycle) > 2 else 0
            # Deliberately 3-element, not widened to carry a 4th slot
            # (#698 / see the docstring above): the declared baseline is
            # deployment-local and must never come from the exporter.
            normalized.append([period, amplitude, phase])
        get_REDIS_DB().hset(
            field.get_cycles_hash_key(model_instance, field_name),
            member_key,
            msgpack.packb(normalized),
        )

    pressure = state.get("pressure")
    if pressure:
        get_REDIS_DB().hset(
            field.get_pressure_hash_key(model_instance, field_name),
            member_key,
            msgpack.packb(
                {
                    "rate": float(pressure.get("rate", 0.0) or 0.0),
                    "last_resolved": float(
                        pressure.get("last_resolved", 0.0) or 0.0
                    ),
                }
            ),
        )
    return None

get_cycles_hash_key(model_instance, field_name)

Build the Redis key for the cycles companion hash.

Public API for external callers that need direct Redis access to cycle data (e.g., bulk inspection, custom cycle updates, monitoring).

Pattern: $CyclicDecayF:{Model}:{field}:{partitions}:cycles

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

def get_cycles_hash_key(self, model_instance, field_name):
    """Build the Redis key for the cycles companion hash.

    Public API for external callers that need direct Redis access to
    cycle data (e.g., bulk inspection, custom cycle updates, monitoring).

    Pattern: $CyclicDecayF:{Model}:{field}:{partitions}:cycles
    """
    ss_key = self.get_partitioned_sortedset_db_key(model_instance, field_name)
    return ss_key.redis_key + ":cycles"

get_pressure_hash_key(model_instance, field_name)

Build the Redis key for the pressure companion hash.

Public API for external callers that need direct Redis access to pressure data (e.g., bulk pressure resets, monitoring dashboards).

Pattern: $CyclicDecayF:{Model}:{field}:{partitions}:pressure

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

def get_pressure_hash_key(self, model_instance, field_name):
    """Build the Redis key for the pressure companion hash.

    Public API for external callers that need direct Redis access to
    pressure data (e.g., bulk pressure resets, monitoring dashboards).

    Pattern: $CyclicDecayF:{Model}:{field}:{partitions}:pressure
    """
    ss_key = self.get_partitioned_sortedset_db_key(model_instance, field_name)
    return ss_key.redis_key + ":pressure"

rank_decayed(zset_key, *, now, n=None, confidence=None, validity=None, decay_rate=None, base_score_field=None)

Evaluate the cyclic decay script over one sorted set (#648).

Overrides :meth:DecayingSortedField.rank_decayed because this fork of the decay math uses an incompatible KEYS layout: cycles at KEYS[2] and pressure at KEYS[3], which pushes the confidence hash to KEYS[4]. Both scripts carry a comment forbidding a "unify" on KEYS[2] -- reusing index 2 here would cmsgpack.unpack the cycles array as a confidence dict, a silent corrupt read rather than a clean crash. Keeping the two layouts in two class bodies, rather than behind a flag in one, is what makes that mistake unavailable instead of merely discouraged.

The companion hash keys are the partition ZSET key plus a suffix (the same derivation as :meth:get_cycles_hash_key / :meth:get_cycles_hash_key_from_parts), so they follow from zset_key alone. The confidence hash does not -- it lives under its own $ConfidencF: prefix -- so it arrives resolved in confidence.

validity is accepted and deliberately ignored: KEYS 1-4 are taken here and the script's header forbids renumbering, so this script has no validity gate. That gap is an explicit No-Go, pinned by tests/test_validity_field.py::TestCyclicDecayGatingGap and recorded under "Known limitations" in docs/features/validity-and-supersession.md. The parameter is kept in the signature so callers stay polymorphic; if you ever gate this script, update all three places.

Args and return value are otherwise as :meth:DecayingSortedField.rank_decayed.

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

def rank_decayed(
    self,
    zset_key: str,
    *,
    now: float,
    n: Optional[int] = None,
    confidence: Optional[tuple[str, str, str]] = None,
    validity: Optional[tuple[str, str, str]] = None,
    decay_rate: Optional[float] = None,
    base_score_field: Optional[str] = None,
) -> "list[Any]":
    """Evaluate the cyclic decay script over one sorted set (#648).

    Overrides :meth:`DecayingSortedField.rank_decayed` because this fork of
    the decay math uses an **incompatible KEYS layout**: cycles at
    ``KEYS[2]`` and pressure at ``KEYS[3]``, which pushes the confidence
    hash to ``KEYS[4]``. Both scripts carry a comment forbidding a "unify"
    on ``KEYS[2]`` -- reusing index 2 here would ``cmsgpack.unpack`` the
    cycles array as a confidence dict, a silent corrupt read rather than a
    clean crash. Keeping the two layouts in two class bodies, rather than
    behind a flag in one, is what makes that mistake unavailable instead of
    merely discouraged.

    The companion hash keys are the partition ZSET key plus a suffix (the
    same derivation as :meth:`get_cycles_hash_key` /
    :meth:`get_cycles_hash_key_from_parts`), so they follow from
    ``zset_key`` alone. The confidence hash does not -- it lives under its
    own ``$ConfidencF:`` prefix -- so it arrives resolved in ``confidence``.

    ``validity`` is accepted and **deliberately ignored**: ``KEYS`` 1-4 are
    taken here and the script's header forbids renumbering, so this script
    has no validity gate. That gap is an explicit No-Go, pinned by
    ``tests/test_validity_field.py::TestCyclicDecayGatingGap`` and recorded
    under "Known limitations" in
    ``docs/features/validity-and-supersession.md``. The parameter is kept in
    the signature so callers stay polymorphic; if you ever gate this script,
    update all three places.

    Args and return value are otherwise as
    :meth:`DecayingSortedField.rank_decayed`.
    """
    conf_hash_key, conf_s, conf_c0 = (
        MODULATION_DISABLED if confidence is None else confidence
    )
    if n is None:
        n = int(get_REDIS_DB().zcard(zset_key))
        if not n:
            return []
    effective_rate = self.decay_rate if decay_rate is None else decay_rate
    if base_score_field is None:
        base_score_field = self.base_score_field or ""

    return run_lua(
        get_REDIS_DB(),
        CYCLIC_DECAY_LUA,
        # numkeys: zset + cycles + pressure + confidence (KEYS[4]).
        # Passing the confidence key without bumping this would shunt it
        # into ARGV and silently disable modulation.
        4,
        zset_key,
        zset_key + ":cycles",
        zset_key + ":pressure",
        conf_hash_key,
        str(now),
        str(effective_rate),
        str(n),
        base_score_field,
        conf_s,
        conf_c0,
    )

get_cycles_hash_key_from_parts(model_class, field_name, *partition_values) classmethod

Build cycles hash key from model class and explicit partition values.

Public API for query paths and external callers that have partition values but not a model instance.

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

@classmethod
def get_cycles_hash_key_from_parts(cls, model_class, field_name, *partition_values):
    """Build cycles hash key from model class and explicit partition values.

    Public API for query paths and external callers that have partition
    values but not a model instance.
    """
    ss_key = cls.get_sortedset_db_key(model_class, field_name, *partition_values)
    return ss_key.redis_key + ":cycles"

get_pressure_hash_key_from_parts(model_class, field_name, *partition_values) classmethod

Build pressure hash key from model class and explicit partition values.

Public API for query paths and external callers that have partition values but not a model instance.

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

@classmethod
def get_pressure_hash_key_from_parts(
    cls, model_class, field_name, *partition_values
):
    """Build pressure hash key from model class and explicit partition values.

    Public API for query paths and external callers that have partition
    values but not a model instance.
    """
    ss_key = cls.get_sortedset_db_key(model_class, field_name, *partition_values)
    return ss_key.redis_key + ":pressure"

on_save(model_instance, field_name, field_value, pipeline=None, **kwargs) classmethod

Store timestamp (parent) then store cycle/pressure companion data.

Both companion hashes follow the same rule: declared parameters are refreshed from the field; learned state is preserved.

For cycles, period and phase are declarative and re-read from field.cycles on every save. amplitude is learned (mutated by strengthen_cycle / weaken_cycle) but is now merged with a three-way rule rather than a two-way one (#698): each stored entry may carry an optional 4th slot, declared_baseline — the declared amplitude that was in force the last time on_save wrote this entry. Comparing the incoming declared amplitude against that baseline (not against the learned value) distinguishes "the developer edited the declaration" from "learning diverged from the declaration":

• baseline absent (a legacy 3-element entry, or a non-numeric slot 3) → "baseline unknown" → preserve the learned amplitude exactly as before #698, and record the declared amplitude as the new baseline.
• baseline equals the incoming declared amplitude → the declaration has not moved → preserve the learned amplitude, re-record the same baseline.
• baseline differs from the incoming declared amplitude → the developer edited the declaration → discard the learned amplitude, adopt the declared value, record it as the new baseline, and emit one logger.info naming the model, field, member key, period, old baseline, new declared value and the discarded learned amplitude.

The comparison is exact float equality, never a tolerance — both sides are the same Python float, round-tripped through msgpack, which preserves IEEE doubles exactly. Stored cycles are matched to declared cycles by period, FIFO within duplicate periods, with the amplitude and its baseline popped together as one decision; a declared period with nothing stored takes the declared amplitude (baseline unknown), and a stored period no longer declared is dropped. Before #679 this branch overwrote the whole entry with the declared defaults, silently erasing everything the strengthen/weaken calls had accumulated; #679 then made the merge unconditional the other way, so an edited declaration could never win. #698 adds the missing third input (the baseline) so the merge can tell the two cases apart. A record with no recorded baseline needs two saves to honor an edit made after this change ships: the first save records the baseline, and only an edit made after that save is detected — see the field's module docs / docs/features/cyclic-decay-field.md.

For pressure, rate is declarative and last_resolved is learned: on first save (no existing entry) the full dict is written with last_resolved=now; on subsequent saves only the rate is updated.

Both companion hashes are read-modified-written by one Lua script, CYCLES_MERGE_LUA (#699), run eagerly against the live connection regardless of a caller-supplied pipeline — the merge decision (which amplitude/rate wins) has to be known synchronously to emit the

698 reset log, and a pipeline-queued script's return value is not

available until execute(), which save() does not surface. Running it eagerly is also what closes the race this exists for: without it, two concurrent save() calls — or a save() racing strengthen_cycle()/weaken_cycle()/resolve_pressure() — each read-then-write the hash independently and the last writer clobbers the other's update; the script makes the whole read-decide-write sequence one atomic server-side step.

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

@classmethod
def on_save(cls, model_instance, field_name, field_value, pipeline=None, **kwargs):
    """Store timestamp (parent) then store cycle/pressure companion data.

    Both companion hashes follow the same rule: **declared parameters are
    refreshed from the field; learned state is preserved.**

    For cycles, ``period`` and ``phase`` are declarative and re-read from
    ``field.cycles`` on every save. ``amplitude`` is learned (mutated by
    ``strengthen_cycle`` / ``weaken_cycle``) but is now merged with a
    **three-way rule** rather than a two-way one (#698): each stored entry
    may carry an optional 4th slot, ``declared_baseline`` — the declared
    amplitude that was in force the last time ``on_save`` wrote this entry.
    Comparing the *incoming* declared amplitude against that baseline (not
    against the learned value) distinguishes "the developer edited the
    declaration" from "learning diverged from the declaration":

    - baseline absent (a legacy 3-element entry, or a non-numeric slot 3) →
      "baseline unknown" → preserve the learned amplitude exactly as before
      #698, and record the declared amplitude as the new baseline.
    - baseline equals the incoming declared amplitude → the declaration has
      not moved → preserve the learned amplitude, re-record the same
      baseline.
    - baseline differs from the incoming declared amplitude → the developer
      edited the declaration → **discard the learned amplitude**, adopt the
      declared value, record it as the new baseline, and emit one
      ``logger.info`` naming the model, field, member key, period, old
      baseline, new declared value and the discarded learned amplitude.

    The comparison is exact float equality, never a tolerance — both sides
    are the same Python float, round-tripped through msgpack, which
    preserves IEEE doubles exactly. Stored cycles are matched to declared
    cycles by period, FIFO within duplicate periods, with the amplitude and
    its baseline popped together as one decision; a declared period with
    nothing stored takes the declared amplitude (baseline unknown), and a
    stored period no longer declared is dropped. Before #679 this branch
    overwrote the whole entry with the declared defaults, silently erasing
    everything the strengthen/weaken calls had accumulated; #679 then made
    the merge unconditional the other way, so an edited declaration could
    never win. #698 adds the missing third input (the baseline) so the
    merge can tell the two cases apart. A record with no recorded baseline
    needs two saves to honor an edit made after this change ships: the
    first save records the baseline, and only an edit made after that save
    is detected — see the field's module docs / docs/features/cyclic-decay-field.md.

    For pressure, ``rate`` is declarative and ``last_resolved`` is learned:
    on first save (no existing entry) the full dict is written with
    ``last_resolved=now``; on subsequent saves only the rate is updated.

    Both companion hashes are read-modified-written by one Lua script,
    ``CYCLES_MERGE_LUA`` (#699), run eagerly against the live connection
    regardless of a caller-supplied ``pipeline`` — the merge decision
    (which amplitude/rate wins) has to be known synchronously to emit the
    #698 reset log, and a pipeline-queued script's return value is not
    available until ``execute()``, which ``save()`` does not surface.
    Running it eagerly is also what closes the race this exists for:
    without it, two concurrent ``save()`` calls — or a ``save()`` racing
    ``strengthen_cycle()``/``weaken_cycle()``/``resolve_pressure()`` —
    each read-then-write the hash independently and the last writer
    clobbers the other's update; the script makes the whole
    read-decide-write sequence one atomic server-side step.
    """
    # Call parent to store timestamp in sorted set
    result = super().on_save(
        model_instance, field_name, field_value, pipeline=pipeline, **kwargs
    )

    field = model_instance._meta.fields[field_name]
    if not isinstance(field, CyclicDecayField):
        return result

    member_key = model_instance.db_key.redis_key
    cycles_hash_key = field.get_cycles_hash_key(model_instance, field_name)
    pressure_hash_key = field.get_pressure_hash_key(model_instance, field_name)

    # Build ARGV: member, N, then N (period, amplitude, phase) triples,
    # then pressure_rate, then now. Periods may be TemporalPeriod
    # aliases (strings) or raw numbers — EVAL only accepts string/number
    # ARGV, so every slot is stringified; the script's coerce_period()
    # converts numeric-looking strings back before storage (B1).
    argv: list[Any] = [member_key, str(len(field.cycles))]
    for cycle in field.cycles:
        period, declared_amplitude = cycle[0], cycle[1]
        phase = cycle[2] if len(cycle) > 2 else 0
        argv.extend([str(period), str(declared_amplitude), str(phase)])
    argv.append(str(field.pressure_rate))
    argv.append(str(time.time()))

    packed_report = run_lua(
        get_REDIS_DB(),
        CYCLES_MERGE_LUA,
        2,
        cycles_hash_key,
        pressure_hash_key,
        *argv,
    )
    decode_failed, resets = msgpack.unpackb(packed_report, raw=False)

    if decode_failed:
        # Mirror export_state's handler: #679 exists because this state
        # was destroyed silently. Do not add a second mute path — fall
        # back to declared defaults, but say so.
        logger.warning(
            f"Could not decode cycles data for {member_key}; "
            f"falling back to declared amplitudes for {field_name}"
        )

    for period, old_baseline, declared_amplitude, learned_amplitude in resets:
        # cmsgpack packs an integral Lua number as a msgpack integer
        # (Lua 5.1 has one number type), so a whole-number amplitude or
        # baseline round-trips as int rather than float. Coerce back to
        # float here so the log line matches the pre-#699 Python-float
        # formatting; period is left alone since it may legitimately be
        # a non-numeric string. Coercion is display-only and defensive:
        # a hand-written/migrated payload can carry a non-numeric value
        # that still decodes as valid msgpack, and the pre-#699 code
        # logged such values fine — never raise out of save() here.
        try:
            old_baseline = float(old_baseline)
            declared_amplitude = float(declared_amplitude)
            learned_amplitude = float(learned_amplitude)
        except (TypeError, ValueError):
            pass
        # The developer edited the declared amplitude — the declaration
        # wins. The learned amplitude is discarded by design (#698); this
        # destroys real state, so it is logged loudly.
        logger.info(
            f"CyclicDecayField declared amplitude changed for "
            f"{model_instance.__class__.__name__}.{field_name} "
            f"member={member_key} period={period!r}: "
            f"declared baseline {old_baseline!r} -> "
            f"{declared_amplitude!r}; discarded learned "
            f"amplitude {learned_amplitude!r}"
        )

    return result

on_delete(model_instance, field_name, field_value, pipeline=None, **kwargs) classmethod

Remove companion hash entries then delegate to parent.

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

@classmethod
def on_delete(
    cls, model_instance, field_name, field_value, pipeline=None, **kwargs
):
    """Remove companion hash entries then delegate to parent."""
    field = model_instance._meta.fields[field_name]

    if isinstance(field, CyclicDecayField):
        member_key = (
            kwargs.get("saved_redis_key") or model_instance.db_key.redis_key
        )
        cycles_hash_key = field.get_cycles_hash_key(model_instance, field_name)
        pressure_hash_key = field.get_pressure_hash_key(model_instance, field_name)

        db = (
            pipeline
            if isinstance(pipeline, redis.client.Pipeline)
            else get_REDIS_DB()
        )
        db.hdel(cycles_hash_key, member_key)
        db.hdel(pressure_hash_key, member_key)

    # Delegate to parent for sorted set cleanup
    return super().on_delete(
        model_instance, field_name, field_value, pipeline=pipeline, **kwargs
    )