| """Router-free causal temporal budgeting for 2D streaming video. |
| |
| The controller deliberately has no task, benchmark, answer-space, or protocol |
| input. It receives a single chronological stream of continuous frame signals, |
| reserves a recent high-fidelity window, and spends the remaining fixed model |
| budget on high-value historical transitions. Query compatibility may enter as |
| one continuous score, but it cannot select a different policy or frame budget. |
| """ |
|
|
| from __future__ import annotations |
|
|
| import hashlib |
| import json |
| import math |
| from dataclasses import asdict, dataclass |
| from typing import Any, Sequence |
|
|
|
|
| POLICY_ID = "unified_causal_temporal_budget_v0" |
| TRANSITION_PAIR_POLICY_ID = "unified_causal_transition_pair_budget_v0" |
|
|
|
|
| class TemporalBudgetError(ValueError): |
| """Raised when the controller input violates its causal public contract.""" |
|
|
|
|
| @dataclass(frozen=True) |
| class TemporalBudgetConfig: |
| max_model_frames: int = 20 |
| recent_frames: int = 4 |
| novelty_weight: float = 0.20 |
| curvature_weight: float = 0.45 |
| query_weight: float = 0.25 |
| recency_weight: float = 0.10 |
|
|
| def validate(self) -> None: |
| if self.max_model_frames <= 0: |
| raise TemporalBudgetError("max_model_frames must be positive") |
| if self.recent_frames <= 0: |
| raise TemporalBudgetError("recent_frames must be positive") |
| if self.recent_frames > self.max_model_frames: |
| raise TemporalBudgetError("recent window exceeds model frame budget") |
| weights = ( |
| self.novelty_weight, |
| self.curvature_weight, |
| self.query_weight, |
| self.recency_weight, |
| ) |
| if any(not math.isfinite(value) or value < 0 for value in weights): |
| raise TemporalBudgetError("controller weights must be finite and non-negative") |
| if sum(weights) <= 0: |
| raise TemporalBudgetError("at least one controller weight must be positive") |
|
|
| @property |
| def sha256(self) -> str: |
| payload = json.dumps( |
| {"policy_id": POLICY_ID, **asdict(self)}, |
| sort_keys=True, |
| separators=(",", ":"), |
| ) |
| return hashlib.sha256(payload.encode("utf-8")).hexdigest() |
|
|
|
|
| @dataclass(frozen=True) |
| class FrameSignal: |
| frame_index: int |
| timestamp: float |
| novelty: float |
| curvature: float |
| query_relevance: float = 0.0 |
|
|
| def validate(self) -> None: |
| if self.frame_index < 0: |
| raise TemporalBudgetError("frame index must be non-negative") |
| values = ( |
| self.timestamp, |
| self.novelty, |
| self.curvature, |
| self.query_relevance, |
| ) |
| if any(not math.isfinite(value) for value in values): |
| raise TemporalBudgetError("frame signals must be finite") |
|
|
|
|
| def _rank_normalize(values: Sequence[float]) -> list[float]: |
| """Average-tie rank normalization in [0, 1], invariant to score scale.""" |
|
|
| if not values: |
| return [] |
| if len(values) == 1: |
| return [0.0] |
| order = sorted(range(len(values)), key=lambda index: (values[index], index)) |
| ranks = [0.0] * len(values) |
| offset = 0 |
| while offset < len(order): |
| end = offset + 1 |
| while end < len(order) and values[order[end]] == values[order[offset]]: |
| end += 1 |
| average = (offset + end - 1) / 2.0 |
| normalized = average / (len(order) - 1) |
| for position in range(offset, end): |
| ranks[order[position]] = normalized |
| offset = end |
| return ranks |
|
|
|
|
| def _cosine_distance(left: Sequence[float], right: Sequence[float]) -> float: |
| if len(left) != len(right) or not left: |
| raise TemporalBudgetError("feature vectors must have the same non-zero width") |
| dot = sum(a * b for a, b in zip(left, right, strict=True)) |
| left_norm = math.sqrt(sum(value * value for value in left)) |
| right_norm = math.sqrt(sum(value * value for value in right)) |
| if left_norm <= 1e-12 or right_norm <= 1e-12: |
| return 0.0 |
| cosine = max(-1.0, min(1.0, dot / (left_norm * right_norm))) |
| return 1.0 - cosine |
|
|
|
|
| def trajectory_change_scores( |
| features: Sequence[Sequence[float]], |
| ) -> tuple[list[float], list[float]]: |
| """Compute first-order novelty and second-order trajectory curvature. |
| |
| Curvature is the cosine distance between consecutive displacement vectors, |
| matching the scale-free geometric signal used by curvature-aware streaming |
| memory without committing to a particular visual feature backend. |
| """ |
|
|
| if not features: |
| return [], [] |
| width = len(features[0]) |
| if width <= 0 or any(len(feature) != width for feature in features): |
| raise TemporalBudgetError("features must form a non-empty rectangular matrix") |
| vectors = [[float(value) for value in feature] for feature in features] |
| if any(not math.isfinite(value) for row in vectors for value in row): |
| raise TemporalBudgetError("features must be finite") |
|
|
| novelty = [0.0] |
| novelty.extend( |
| _cosine_distance(vectors[index - 1], vectors[index]) |
| for index in range(1, len(vectors)) |
| ) |
| curvature = [0.0] * len(vectors) |
| for index in range(1, len(vectors) - 1): |
| previous_delta = [ |
| current - previous |
| for previous, current in zip( |
| vectors[index - 1], vectors[index], strict=True |
| ) |
| ] |
| next_delta = [ |
| future - current |
| for current, future in zip( |
| vectors[index], vectors[index + 1], strict=True |
| ) |
| ] |
| curvature[index] = _cosine_distance(previous_delta, next_delta) |
| return novelty, curvature |
|
|
|
|
| def build_frame_signals( |
| *, |
| frame_indices: Sequence[int], |
| timestamps: Sequence[float], |
| features: Sequence[Sequence[float]], |
| query_relevance: Sequence[float] | None = None, |
| ) -> list[FrameSignal]: |
| if not (len(frame_indices) == len(timestamps) == len(features)): |
| raise TemporalBudgetError("indices, timestamps, and features must align") |
| if query_relevance is None: |
| query_relevance = [0.0] * len(frame_indices) |
| if len(query_relevance) != len(frame_indices): |
| raise TemporalBudgetError("query relevance must align with frames") |
| novelty, curvature = trajectory_change_scores(features) |
| return [ |
| FrameSignal( |
| frame_index=int(index), |
| timestamp=float(timestamp), |
| novelty=float(novelty_value), |
| curvature=float(curvature_value), |
| query_relevance=float(relevance), |
| ) |
| for index, timestamp, novelty_value, curvature_value, relevance in zip( |
| frame_indices, |
| timestamps, |
| novelty, |
| curvature, |
| query_relevance, |
| strict=True, |
| ) |
| ] |
|
|
|
|
| def select_temporal_budget( |
| signals: Sequence[FrameSignal], |
| *, |
| causal_frontier: float, |
| config: TemporalBudgetConfig = TemporalBudgetConfig(), |
| ) -> dict[str, Any]: |
| """Select one deterministic, chronologically returned causal frame set.""" |
|
|
| config.validate() |
| if not math.isfinite(causal_frontier) or causal_frontier < 0: |
| raise TemporalBudgetError("causal frontier must be finite and non-negative") |
| ordered = sorted(signals, key=lambda item: (item.timestamp, item.frame_index)) |
| if not ordered: |
| raise TemporalBudgetError("temporal budget requires at least one frame") |
| for signal in ordered: |
| signal.validate() |
| if signal.timestamp > causal_frontier + 1e-3: |
| raise TemporalBudgetError("candidate frame crosses the causal frontier") |
| indices = [signal.frame_index for signal in ordered] |
| if len(indices) != len(set(indices)): |
| raise TemporalBudgetError("candidate frame indices must be unique") |
| if any( |
| right.timestamp + 1e-9 < left.timestamp |
| for left, right in zip(ordered, ordered[1:]) |
| ): |
| raise TemporalBudgetError("candidate timestamps must be chronological") |
|
|
| novelty = _rank_normalize([signal.novelty for signal in ordered]) |
| curvature = _rank_normalize([signal.curvature for signal in ordered]) |
| relevance = _rank_normalize([signal.query_relevance for signal in ordered]) |
| recency = _rank_normalize([signal.timestamp for signal in ordered]) |
| scored: list[dict[str, Any]] = [] |
| for signal, n_value, c_value, q_value, r_value in zip( |
| ordered, novelty, curvature, relevance, recency, strict=True |
| ): |
| score = ( |
| config.novelty_weight * n_value |
| + config.curvature_weight * c_value |
| + config.query_weight * q_value |
| + config.recency_weight * r_value |
| ) |
| scored.append( |
| { |
| "frame_index": signal.frame_index, |
| "timestamp": signal.timestamp, |
| "score": score, |
| "components": { |
| "novelty_rank": n_value, |
| "curvature_rank": c_value, |
| "query_relevance_rank": q_value, |
| "recency_rank": r_value, |
| }, |
| } |
| ) |
|
|
| recent_count = min(config.recent_frames, len(scored)) |
| recent_ids = {row["frame_index"] for row in scored[-recent_count:]} |
| remaining = max(0, config.max_model_frames - recent_count) |
| historical = [row for row in scored if row["frame_index"] not in recent_ids] |
| chosen_history = sorted( |
| historical, |
| key=lambda row: (-row["score"], -row["timestamp"], row["frame_index"]), |
| )[:remaining] |
| chosen_ids = recent_ids | {row["frame_index"] for row in chosen_history} |
| selected = [row for row in scored if row["frame_index"] in chosen_ids] |
| for row in selected: |
| row["allocation"] = ( |
| "recent_high_fidelity" |
| if row["frame_index"] in recent_ids |
| else "selected_history" |
| ) |
|
|
| return { |
| "policy_id": POLICY_ID, |
| "policy_config": asdict(config), |
| "policy_config_sha256": config.sha256, |
| "benchmark_task_or_group_router": False, |
| "question_keyword_router": False, |
| "single_continuous_scoring_rule": True, |
| "causal_frontier": float(causal_frontier), |
| "candidate_frames_inspected": len(scored), |
| "model_visible_frames": len(selected), |
| "extra_frames_inspected": max(0, len(scored) - len(selected)), |
| "selected_frame_indices": [row["frame_index"] for row in selected], |
| "selected_timestamps": [row["timestamp"] for row in selected], |
| "selected": selected, |
| "future_frame_violations": 0, |
| } |
|
|
|
|
| def select_transition_pair_budget( |
| signals: Sequence[FrameSignal], |
| *, |
| causal_frontier: float, |
| config: TemporalBudgetConfig = TemporalBudgetConfig(), |
| ) -> dict[str, Any]: |
| """Select task-blind historical before/after pairs plus the recent window. |
| |
| ``select_temporal_budget`` spends every historical slot independently. A |
| high-curvature frame can therefore be shown without its immediately |
| preceding state, which is a poor interface for action change, occurrence, |
| and temporal-order decisions. This variant keeps the same scoring rule, |
| frame budget, and recent reservation, but greedily spends historical slots |
| on ``(previous, anchor)`` pairs. Unused slots are filled by the original |
| global ranking. The question, task, benchmark, and model prediction never |
| enter selection. |
| """ |
|
|
| reference = select_temporal_budget( |
| signals, causal_frontier=causal_frontier, config=config |
| ) |
| ordered = sorted(signals, key=lambda item: (item.timestamp, item.frame_index)) |
| recent_count = min(config.recent_frames, len(ordered)) |
| recent_ids = {signal.frame_index for signal in ordered[-recent_count:]} |
| history = [signal for signal in ordered if signal.frame_index not in recent_ids] |
| history_budget = max(0, config.max_model_frames - recent_count) |
|
|
| novelty = _rank_normalize([signal.novelty for signal in ordered]) |
| curvature = _rank_normalize([signal.curvature for signal in ordered]) |
| relevance = _rank_normalize([signal.query_relevance for signal in ordered]) |
| recency = _rank_normalize([signal.timestamp for signal in ordered]) |
| score_by_id = { |
| signal.frame_index: ( |
| config.novelty_weight * n_value |
| + config.curvature_weight * c_value |
| + config.query_weight * q_value |
| + config.recency_weight * r_value |
| ) |
| for signal, n_value, c_value, q_value, r_value in zip( |
| ordered, novelty, curvature, relevance, recency, strict=True |
| ) |
| } |
| history_position = { |
| signal.frame_index: position for position, signal in enumerate(history) |
| } |
| ranked_anchors = sorted( |
| history, |
| key=lambda signal: ( |
| -score_by_id[signal.frame_index], |
| -signal.timestamp, |
| signal.frame_index, |
| ), |
| ) |
| chosen: set[int] = set() |
| anchors: set[int] = set() |
| contexts: set[int] = set() |
| for anchor in ranked_anchors: |
| if len(chosen) >= history_budget: |
| break |
| position = history_position[anchor.frame_index] |
| context = history[position - 1] if position > 0 else None |
| pair = ( |
| [context.frame_index, anchor.frame_index] |
| if context is not None |
| else [anchor.frame_index] |
| ) |
| additions = [frame_id for frame_id in pair if frame_id not in chosen] |
| if not additions or len(chosen) + len(additions) > history_budget: |
| continue |
| chosen.update(additions) |
| anchors.add(anchor.frame_index) |
| if context is not None: |
| contexts.add(context.frame_index) |
|
|
| for signal in ranked_anchors: |
| if len(chosen) >= history_budget: |
| break |
| chosen.add(signal.frame_index) |
|
|
| chosen_ids = recent_ids | chosen |
| selected = [] |
| for signal in ordered: |
| if signal.frame_index not in chosen_ids: |
| continue |
| if signal.frame_index in recent_ids: |
| allocation = "recent_high_fidelity" |
| elif signal.frame_index in anchors: |
| allocation = "transition_anchor" |
| elif signal.frame_index in contexts: |
| allocation = "transition_context" |
| else: |
| allocation = "selected_history_fill" |
| selected.append( |
| { |
| "frame_index": signal.frame_index, |
| "timestamp": signal.timestamp, |
| "score": score_by_id[signal.frame_index], |
| "allocation": allocation, |
| } |
| ) |
|
|
| policy_payload = json.dumps( |
| {"policy_id": TRANSITION_PAIR_POLICY_ID, **asdict(config)}, |
| sort_keys=True, |
| separators=(",", ":"), |
| ) |
| return { |
| "policy_id": TRANSITION_PAIR_POLICY_ID, |
| "policy_config": asdict(config), |
| "policy_config_sha256": hashlib.sha256(policy_payload.encode("utf-8")).hexdigest(), |
| "reference_policy_config_sha256": reference["policy_config_sha256"], |
| "benchmark_task_or_group_router": False, |
| "question_keyword_router": False, |
| "single_continuous_scoring_rule": True, |
| "causal_frontier": float(causal_frontier), |
| "candidate_frames_inspected": len(ordered), |
| "model_visible_frames": len(selected), |
| "extra_frames_inspected": max(0, len(ordered) - len(selected)), |
| "selected_frame_indices": [row["frame_index"] for row in selected], |
| "selected_timestamps": [row["timestamp"] for row in selected], |
| "selected": selected, |
| "transition_pairs": sum( |
| row["allocation"] == "transition_context" for row in selected |
| ), |
| "future_frame_violations": 0, |
| } |
|
|
|
|
| __all__ = [ |
| "FrameSignal", |
| "POLICY_ID", |
| "TRANSITION_PAIR_POLICY_ID", |
| "TemporalBudgetConfig", |
| "TemporalBudgetError", |
| "build_frame_signals", |
| "select_temporal_budget", |
| "select_transition_pair_budget", |
| "trajectory_change_scores", |
| ] |
|
|
