Skip to content

Implement Phase 4 reasoning trigger gate with cooldown #137

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
Devnil434 merged 2 commits into from
Jun 2, 2026
+78 −117
Merged

Implement Phase 4 reasoning trigger gate with cooldown #137

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
3035870
feat(memory): implement Phase 4 reasoning trigger gate
dilshadalikhan2004 May 29, 2026
f9b0ac0
Merge branch 'main' into feat/reasoning-trigger-gate
Devnil434 May 30, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
18 changes: 12 additions & 6 deletions libs/config/settings.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,6 +30,14 @@ class Settings(BaseSettings):
tracker_max_cosine_distance: float = 0.4
camera_id: str = "cam_01"

# Action classifier settings
lingering_threshold_sec: float = 5.0
movement_threshold_px: float = 15.0
near_keypad_dist_px: float = 75.0
keypad_center_x: float = 500.0
keypad_center_y: float = 500.0

# Reasoning trigger settings
reasoning_dwell_threshold_seconds: float = 5.0
reasoning_cooldown_seconds: float = 5.0

Expand All @@ -45,12 +53,10 @@ class Settings(BaseSettings):
kafka_bootstrap_servers: str = "localhost:9092"
kafka_topic: str = "track-events"

@property
def REDIS_URL(self) -> str:
"""Backward-compatible uppercase alias for redis_url."""
return self.redis_url

model_config = SettingsConfigDict(env_file=".env", env_file_encoding="utf-8")
model_config = SettingsConfigDict(
env_file=".env",
env_file_encoding="utf-8",
)


settings = Settings()
90 changes: 32 additions & 58 deletions services/memory/trigger.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,93 +2,67 @@

from time import monotonic

from libs.schemas.tracking import TrackLifecycleEvent
from libs.config.settings import settings

from libs.schemas.memory import (
TrackSequence,
ActionHint,
)

_reasoning_cooldowns: dict[int, float] = {}

SUSPICIOUS_ACTIONS = {
"LINGERING",
"NEAR_KEYPAD",
"REPEATED_APPROACH",
ActionHint.LINGERING,
ActionHint.NEAR_KEYPAD,
ActionHint.REPEATED_APPROACH,
}


"""
Clear cooldown state for a track after reasoning completes.
"""

def reset_cooldown(track_id: int) -> None:
"""Clear cooldown state after reasoning completes."""
"""
Clear cooldown state after reasoning completes.
"""
_reasoning_cooldowns.pop(track_id, None)

"""
Determine whether VLM/LLM reasoning should be triggered
for a suspicious track sequence.

Conditions:
- track must be inside a restricted zone
- dwell time must exceed configured threshold
- at least one suspicious action must exist
- track must not be inside cooldown window
"""

def should_trigger_reasoning(
event: TrackLifecycleEvent,
suspicious_actions: set[str],
seq: TrackSequence,
) -> bool:
"""
Determine whether VLM/LLM reasoning should be triggered
for a suspicious track sequence.
Determine whether VLM/LLM reasoning should be triggered.

Conditions:
- track must be inside a restricted zone
- dwell time must exceed configured threshold
- at least one suspicious action must exist
- track must not be inside cooldown window
- Track is inside a restricted zone
- Dwell time exceeds configured threshold
- At least one suspicious action exists
- Track is not inside cooldown window
"""

if settings.reasoning_dwell_threshold_seconds < 0:
raise ValueError(
"reasoning_dwell_threshold_seconds must be >= 0"
)

if settings.reasoning_cooldown_seconds < 0:
raise ValueError(
"reasoning_cooldown_seconds must be >= 0"
)
if not seq.events:
return False

# Must be inside at least one restricted zone
if not event.zones_present:
# Zone check
if not seq.zones_visited:
return False
Comment on lines +43 to 45
Copy link
Copy Markdown

@coderabbitai coderabbitai Bot May 29, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟡 Minor | ⚡ Quick win

🧩 Analysis chain

🏁 Script executed:

#!/bin/bash
# Inspect how zones_visited / zones_key is populated to see whether it is restricted-only.
rg -nP -C4 '\bzones_visited\b' --type=py
rg -nP -C4 '_zones_key|zones_present|restricted' --type=py

Repository: Devnil434/Eagle

Length of output: 50371

Fix restricted-zone gating in should_trigger_reasoning

  • services/memory/trigger.py only checks that seq.zones_visited is non-empty (if not seq.zones_visited: return False), so it will fire for any visited zone.
  • services/memory/memory.py populates zones_visited by collecting event.zone values with no restricted-only filtering; tests/test_memory.py::test_zones_visited_populated shows "safe_corridor" and "restricted_door" both ending up in seq.zones_visited.
    Update the docstring/intent to match “any visited zone” or filter seq.zones_visited against the restricted-zone set before allowing the trigger to proceed.
🤖 Prompt for AI Agents 
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@services/memory/trigger.py` around lines 43 - 45, The current gating in
should_trigger_reasoning only ensures seq.zones_visited is non-empty so it fires
for any visited zone; either update the function docstring to state it
intentionally triggers on any visited zone, or change the gating to require an
intersection with the restricted-zone set (e.g., filter seq.zones_visited
against the restricted zones provided by Memory or a RESTRICTED_ZONES constant)
before returning True; locate should_trigger_reasoning in
services/memory/trigger.py and use the same restricted-zone identifier used in
services/memory/memory.py (or the Memory.restricted_zones attribute) to perform
the filtered check.


# Dwell time must exceed configured threshold
if (
event.dwell_time_seconds
< settings.reasoning_dwell_threshold_seconds
):
# Dwell threshold
if seq.total_dwell < settings.reasoning_dwell_threshold_seconds:
return False

# At least one suspicious action must exist
if not (SUSPICIOUS_ACTIONS & suspicious_actions):
# Suspicious action check
has_suspicious_action = any(event.action_hint in SUSPICIOUS_ACTIONS for event in seq.events)

if not has_suspicious_action:
return False

now = monotonic()

# Cooldown protection
last_trigger = _reasoning_cooldowns.get(event.track_id)
# Cooldown check
last_trigger = _reasoning_cooldowns.get(seq.track_id)

if (
last_trigger is not None
and (
now - last_trigger
< settings.reasoning_cooldown_seconds
)
):
if last_trigger is not None and (now - last_trigger < settings.reasoning_cooldown_seconds):
return False

# Store latest trigger timestamp
_reasoning_cooldowns[event.track_id] = now
# Start cooldown
_reasoning_cooldowns[seq.track_id] = now

return True
return True
87 changes: 34 additions & 53 deletions tests/test_trigger.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,15 +1,17 @@
import pytest

from libs.config.settings import settings
from libs.schemas.memory import (
TrackSequence,
TrackEvent,
ActionHint,
)

from services.memory.trigger import (
should_trigger_reasoning,
reset_cooldown,
)

from libs.schemas.tracking import (
TrackLifecycleEvent,
TrackState,
)

@pytest.fixture(autouse=True)
def fixed_reasoning_gate_config():
Expand All @@ -25,107 +27,86 @@ def fixed_reasoning_gate_config():
settings.reasoning_cooldown_seconds = old_cooldown


def make_event(
def make_sequence(
dwell: float = 10.0,
zones: list[str] | None = None,
track_id: int = 1,
action: ActionHint = ActionHint.LINGERING,
):
return TrackLifecycleEvent(
event=TrackState.LOST,
event = TrackEvent(
track_id=track_id,
frame_id=1,
zones_present=zones if zones is not None else ["restricted_zone"],
timestamp_ms=1000,
action_hint=action,
dwell_time_seconds=dwell,
)

return TrackSequence(
track_id=track_id,
events=[event],
total_dwell=dwell,
zones_visited=(zones if zones is not None else ["restricted_zone"]),
)


def test_returns_false_without_zone():
event = make_event(zones=[])
seq = make_sequence(zones=[])

result = should_trigger_reasoning(
event,
{"LINGERING"},
)
result = should_trigger_reasoning(seq)

assert result is False


def test_returns_false_below_dwell_threshold():
event = make_event(dwell=1.0)
seq = make_sequence(dwell=1.0)

result = should_trigger_reasoning(
event,
{"LINGERING"},
)
result = should_trigger_reasoning(seq)

assert result is False


def test_returns_false_without_suspicious_actions():
event = make_event()
seq = make_sequence(action=ActionHint.WALKING)

result = should_trigger_reasoning(
event,
{"NORMAL_WALKING"},
)
result = should_trigger_reasoning(seq)

assert result is False


def test_returns_true_for_valid_suspicious_sequence():
event = make_event(track_id=100)
seq = make_sequence(track_id=100)

reset_cooldown(100)

result = should_trigger_reasoning(
event,
{"LINGERING"},
)
result = should_trigger_reasoning(seq)

assert result is True


def test_returns_false_during_cooldown():
event = make_event(track_id=200)
seq = make_sequence(track_id=200)

reset_cooldown(200)

first = should_trigger_reasoning(
event,
{"LINGERING"},
)

second = should_trigger_reasoning(
event,
{"LINGERING"},
)
first = should_trigger_reasoning(seq)
second = should_trigger_reasoning(seq)

assert first is True
assert second is False


def test_reset_cooldown_allows_retrigger():
event = make_event(track_id=300)
seq = make_sequence(track_id=300)

reset_cooldown(300)

first = should_trigger_reasoning(
event,
{"LINGERING"},
)

second = should_trigger_reasoning(
event,
{"LINGERING"},
)
first = should_trigger_reasoning(seq)
second = should_trigger_reasoning(seq)

reset_cooldown(300)

third = should_trigger_reasoning(
event,
{"LINGERING"},
)
third = should_trigger_reasoning(seq)

assert first is True
assert second is False
assert third is True
assert third is True
Loading