fix(gateway): keep internal wakes from extending sessions
This commit is contained in:
embwl0x
+13
-2
@@ -16860,7 +16860,10 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
||||
# on error. Let the user drive the next turn.
|
||||
if _final_text.strip():
|
||||
try:
|
||||
session_entry = await self.async_session_store.get_or_create_session(source)
|
||||
session_entry = await self.async_session_store.get_or_create_session(
|
||||
source,
|
||||
touch_activity=not is_internal,
|
||||
)
|
||||
except Exception:
|
||||
session_entry = None
|
||||
if session_entry is not None:
|
||||
@@ -17710,7 +17713,14 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
||||
)
|
||||
return
|
||||
else:
|
||||
session_entry = await self.async_session_store.get_or_create_session(source)
|
||||
# Internal wakes must observe reset policy without becoming user
|
||||
# activity themselves. Otherwise periodic Kanban/process
|
||||
# notifications keep the stable routing key alive across every
|
||||
# daily/idle boundary.
|
||||
session_entry = await self.async_session_store.get_or_create_session(
|
||||
source,
|
||||
touch_activity=not bool(getattr(event, "internal", False)),
|
||||
)
|
||||
session_key = session_entry.session_key
|
||||
if not strict_session and pinned_session_id:
|
||||
resolved_entry = await self._resolve_async_delegation_session(
|
||||
@@ -19511,6 +19521,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
||||
await self.async_session_store.update_session(
|
||||
session_entry.session_key,
|
||||
last_prompt_tokens=agent_result.get("last_prompt_tokens", 0),
|
||||
touch_activity=not bool(getattr(event, "internal", False)),
|
||||
)
|
||||
|
||||
# Re-baseline the cached agent's message_count snapshot now that
|
||||
|
||||
+29
-10
@@ -2429,12 +2429,15 @@ class SessionStore:
|
||||
self,
|
||||
source: SessionSource,
|
||||
force_new: bool = False,
|
||||
touch_activity: bool = True,
|
||||
) -> SessionEntry:
|
||||
"""Single-flight session lookup/create per routing key.
|
||||
|
||||
Calls for different keys remain concurrent. Overlapping calls for the
|
||||
same key share the owner's result, including concurrent ``force_new``
|
||||
deliveries, so only one routing transition and SQLite row is created.
|
||||
``touch_activity=False`` still evaluates reset policy but preserves the
|
||||
prior user-activity clock when an internal/system event reuses a session.
|
||||
"""
|
||||
session_key = self._generate_session_key(source)
|
||||
inflight_lock = getattr(self, "_inflight_lock", None)
|
||||
@@ -2457,10 +2460,16 @@ class SessionStore:
|
||||
if slot.error is not None:
|
||||
raise slot.error
|
||||
assert slot.result is not None
|
||||
if touch_activity:
|
||||
self.update_session(slot.result.session_key)
|
||||
return slot.result
|
||||
|
||||
try:
|
||||
result = self._get_or_create_session_impl(source, force_new=force_new)
|
||||
result = self._get_or_create_session_impl(
|
||||
source,
|
||||
force_new=force_new,
|
||||
touch_activity=touch_activity,
|
||||
)
|
||||
slot.result = result
|
||||
return result
|
||||
except BaseException as exc:
|
||||
@@ -2475,6 +2484,7 @@ class SessionStore:
|
||||
self,
|
||||
source: SessionSource,
|
||||
force_new: bool = False,
|
||||
touch_activity: bool = True,
|
||||
) -> SessionEntry:
|
||||
"""Perform one session routing transition for the single-flight owner.
|
||||
|
||||
@@ -2648,10 +2658,12 @@ class SessionStore:
|
||||
_needs_recover = True
|
||||
elif entry.session_id != _stale_session_id:
|
||||
# Another thread handled this entry during our lock-free
|
||||
# window. Treat as healthy -- bump updated_at and save.
|
||||
entry.updated_at = now
|
||||
_needs_save = True
|
||||
_metadata_only_save = not _healed
|
||||
# window. Treat as healthy; internal/system events preserve
|
||||
# the prior user-activity clock used by reset policy.
|
||||
if touch_activity:
|
||||
entry.updated_at = now
|
||||
_needs_save = touch_activity or _healed
|
||||
_metadata_only_save = touch_activity and not _healed
|
||||
else:
|
||||
# Stale check clean. Apply reset decision.
|
||||
if _reset_reason:
|
||||
@@ -2664,9 +2676,10 @@ class SessionStore:
|
||||
entry = None
|
||||
_needs_recover = True
|
||||
else:
|
||||
entry.updated_at = now
|
||||
_needs_save = True
|
||||
_metadata_only_save = not _healed
|
||||
if touch_activity:
|
||||
entry.updated_at = now
|
||||
_needs_save = touch_activity or _healed
|
||||
_metadata_only_save = touch_activity and not _healed
|
||||
else:
|
||||
if not force_new:
|
||||
_needs_recover = True
|
||||
@@ -2824,14 +2837,20 @@ class SessionStore:
|
||||
self,
|
||||
session_key: str,
|
||||
last_prompt_tokens: int = None,
|
||||
touch_activity: bool = True,
|
||||
) -> None:
|
||||
"""Update lightweight session metadata after an interaction."""
|
||||
"""Update lightweight session metadata after an interaction.
|
||||
|
||||
Internal/system turns can persist token metadata without advancing the
|
||||
user-activity clock that drives idle and daily reset policy.
|
||||
"""
|
||||
with self._lock:
|
||||
self._ensure_loaded_locked()
|
||||
entry = self._entries.get(session_key)
|
||||
if entry is None:
|
||||
return
|
||||
entry.updated_at = _now()
|
||||
if touch_activity:
|
||||
entry.updated_at = _now()
|
||||
if last_prompt_tokens is not None:
|
||||
entry.last_prompt_tokens = last_prompt_tokens
|
||||
# Snapshot peer fields while still holding _lock: a concurrent
|
||||
|
||||
@@ -8,6 +8,7 @@ Verifies that:
|
||||
- resume_pending_expired auto-reset sets the correct reason and DB end_reason
|
||||
"""
|
||||
|
||||
import json
|
||||
from datetime import datetime, timedelta
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
@@ -111,6 +112,52 @@ class TestSessionEntryReason:
|
||||
assert entry2.reset_had_activity is True
|
||||
|
||||
|
||||
class TestInternalActivityResetPolicy:
|
||||
def test_internal_turn_does_not_advance_activity_clock(self, tmp_path):
|
||||
store = _make_store(
|
||||
SessionResetPolicy(mode="idle", idle_minutes=60),
|
||||
tmp_path,
|
||||
)
|
||||
source = _make_source()
|
||||
entry = store.get_or_create_session(source)
|
||||
prior_activity = datetime.now() - timedelta(minutes=10)
|
||||
entry.updated_at = prior_activity
|
||||
store._save()
|
||||
|
||||
reused = store.get_or_create_session(source, touch_activity=False)
|
||||
store.update_session(
|
||||
reused.session_key,
|
||||
last_prompt_tokens=123,
|
||||
touch_activity=False,
|
||||
)
|
||||
|
||||
assert reused.session_id == entry.session_id
|
||||
assert reused.updated_at == prior_activity
|
||||
assert reused.last_prompt_tokens == 123
|
||||
rows = store._db.load_gateway_routing_entries(
|
||||
scope=store._routing_scope()
|
||||
)
|
||||
durable = json.loads(rows[reused.session_key])
|
||||
assert durable["updated_at"] == prior_activity.isoformat()
|
||||
assert durable["last_prompt_tokens"] == 123
|
||||
|
||||
def test_internal_turn_still_triggers_due_reset(self, tmp_path):
|
||||
store = _make_store(
|
||||
SessionResetPolicy(mode="idle", idle_minutes=1),
|
||||
tmp_path,
|
||||
)
|
||||
source = _make_source()
|
||||
entry = store.get_or_create_session(source)
|
||||
entry.updated_at = datetime.now() - timedelta(minutes=5)
|
||||
store._save()
|
||||
|
||||
reset = store.get_or_create_session(source, touch_activity=False)
|
||||
|
||||
assert reset.session_id != entry.session_id
|
||||
assert reset.was_auto_reset is True
|
||||
assert reset.auto_reset_reason == "idle"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# SessionResetPolicy notify config
|
||||
# ---------------------------------------------------------------------------
|
||||
@@ -282,4 +329,3 @@ class TestResumePendingExpiredAutoReset:
|
||||
db.promote_to_session_reset.assert_called_once()
|
||||
_, ended_reason = db.promote_to_session_reset.call_args.args
|
||||
assert ended_reason == "idle"
|
||||
|
||||
|
||||
Reference in New Issue
Block a user