refactor: session_lifecycle_payload 自查最近活动，删除 5 处重复调用

routes/thinking.py

@@ -472,10 +472,7 @@ def _run_stage3_forum_turn(
     payload = result.to_public_dict()
     payload['user_goal'] = _stage3_user_goal(ts.id)
     payload['forum_state'] = _stage3_forum_state(ts.id)
-    payload['session_lifecycle'] = session_lifecycle_payload(
-        ts,
-        last_activity_at=latest_session_activity([ts.id]).get(ts.id),
-    )
+    payload['session_lifecycle'] = session_lifecycle_payload(ts)
     return ts, target_role, payload
 
 
@@ -665,10 +662,7 @@ def _stage3_payload_with_goal(payload, thinking_session, runtime=None):
     result = dict(payload or {})
     getter = getattr(runtime, 'public_user_goal', None)
     result['user_goal'] = getter() if callable(getter) else _stage3_user_goal(thinking_session.id)
-    result['session_lifecycle'] = session_lifecycle_payload(
-        thinking_session,
-        last_activity_at=latest_session_activity([thinking_session.id]).get(thinking_session.id),
-    )
+    result['session_lifecycle'] = session_lifecycle_payload(thinking_session)
     return result
 
 
@@ -979,10 +973,7 @@ def start_session():
         if existing:
             if _ensure_stage3_initial_prompt(existing, assignment, preset):
                 db.session.commit()
-            lifecycle = session_lifecycle_payload(
-                existing,
-                last_activity_at=latest_session_activity([existing.id]).get(existing.id),
-            )
+            lifecycle = session_lifecycle_payload(existing)
             forum_history = _stage3_forum_history(existing.id)
             forum_state = _stage3_forum_state(existing.id)
 
@@ -1111,10 +1102,7 @@ def start_session():
 
         # 记录日志
         _log_event(new_session.id, 1, 'session_start', 'student', '开始引导式学习')
-        lifecycle = session_lifecycle_payload(
-            new_session,
-            last_activity_at=new_session.started_at,
-        )
+        lifecycle = session_lifecycle_payload(new_session)
 
         return jsonify({
             'success': True,
@@ -2348,10 +2336,7 @@ def get_session_status(session_id):
         # this new endpoint does not reveal that a session id exists.
         return jsonify({'error': '会话不存在或无权访问'}), 403
 
-    lifecycle = session_lifecycle_payload(
-        thinking_session,
-        last_activity_at=latest_session_activity([thinking_session.id]).get(thinking_session.id),
-    )
+    lifecycle = session_lifecycle_payload(thinking_session)
     return jsonify({'success': True, 'session': lifecycle})
 
 @thinking.route('/api/session/<int:session_id>/log')

services/session_lifecycle.py

@@ -15,6 +15,11 @@
 LIFECYCLE_STATUSES = ("active", "idle", "completed", "abandoned")
 TERMINAL_STATUSES = {"completed", "abandoned"}
 
+# 区分“调用者省略 last_activity_at”（payload 自查一次最近活动）与
+# “显式传入 None”（批量查询未命中的会话，保留回退 started_at 的既有
+# 行为且不执行查询，避免教师概览等列表产生 N+1）。
+_LAST_ACTIVITY_UNSET = object()
+
 
 def _utc_now(value=None) -> datetime:
     value = value or datetime.utcnow()
@@ -186,9 +191,21 @@ def _stage_progress(session, lifecycle_status):
     }
 
 
-def session_lifecycle_payload(session, *, now=None, last_activity_at=None):
-    """Build a safe, content-free lifecycle payload for UI and teacher views."""
+def session_lifecycle_payload(session, *, now=None, last_activity_at=_LAST_ACTIVITY_UNSET):
+    """Build a safe, content-free lifecycle payload for UI and teacher views.
+
+    ``last_activity_at`` 通常由批量场景的调用者一次性查出再传入（未命中的
+    会话显式传 None，保留回退 started_at 的既有行为，不触发查询，避免
+    N+1）；单 session 调用者省略该参数时，函数内部基于 ``session.id``
+    自查一次最近活动时间，调用者不必在每个调用点重复写
+    ``last_activity_at=latest_session_activity([id]).get(id)``，也不会因为
+    漏传而退化成用 started_at 判 idle（刚活动的会话被误判空闲）。
+    """
     current_time = _utc_now(now)
+    if last_activity_at is _LAST_ACTIVITY_UNSET:
+        session_id = getattr(session, "id", None)
+        if session_id is not None:
+            last_activity_at = latest_session_activity([session_id]).get(session_id)
     lifecycle_status = session_lifecycle_status(
         session,
         now=current_time,

tests/test_session_lifecycle.py

@@ -197,3 +197,95 @@ def test_can_view_session_limits_teacher_to_owner_or_managed_class():
     assert can_view_session(admin, session) is True
     assert can_view_session(teacher, session) is True
     assert can_view_session(other_teacher, session) is False
+
+
+def test_payload_self_fetches_latest_activity_when_omitted(monkeypatch):
+    """单 session 调用者不传 last_activity_at 时，payload 应自查最近活动。
+
+    维护痛点：此前每个单 session 调用者都要手写
+    ``last_activity_at=latest_session_activity([id]).get(id)``，漏传就会
+    退化成用 started_at 判 idle（刚活动的会话被误判 idle）。改为 payload
+    内部自查后，调用者不再需要重复这行。
+    """
+    from services import session_lifecycle
+
+    calls = []
+    logged_activity = UTC_NOW - timedelta(minutes=2)
+
+    def fake_latest_activity(ids):
+        calls.append(list(ids))
+        return {7: logged_activity}
+
+    monkeypatch.setattr(
+        session_lifecycle, "latest_session_activity", fake_latest_activity
+    )
+
+    payload = session_lifecycle.session_lifecycle_payload(
+        _session(),  # id=7, started_at=UTC_NOW-5min
+        now=UTC_NOW,
+    )
+
+    # 用 session.id 自查一次
+    assert calls == [[7]]
+    # last_activity_at 取日志时间，而非 started_at
+    assert payload["last_activity_at"] == logged_activity.strftime("%Y-%m-%dT%H:%M:%SZ")
+    # activity_age 基于日志时间计算（约 2 分钟），不是 started_at 的 5 分钟
+    assert 110 <= payload["activity_age_seconds"] <= 130
+
+
+def test_payload_keeps_provided_activity_without_extra_query(monkeypatch):
+    """批量场景显式传 last_activity_at 时，不再自查，避免 N+1。"""
+    from services import session_lifecycle
+
+    calls = []
+
+    def fake_latest_activity(ids):
+        calls.append(list(ids))
+        return {}
+
+    monkeypatch.setattr(
+        session_lifecycle, "latest_session_activity", fake_latest_activity
+    )
+
+    provided = UTC_NOW - timedelta(minutes=3)
+    payload = session_lifecycle.session_lifecycle_payload(
+        _session(),
+        now=UTC_NOW,
+        last_activity_at=provided,
+    )
+
+    assert calls == []  # 没有触发自查
+    assert payload["last_activity_at"] == provided.strftime("%Y-%m-%dT%H:%M:%SZ")
+
+
+def test_payload_explicit_none_activity_skips_query_and_falls_back_to_started_at(monkeypatch):
+    """显式传入 last_activity_at=None 不得触发自查。
+
+    批量调用者先 latest_session_activity([...]) 一次查出映射，未命中的
+    会话得到 None 后会显式传入。若 payload 把显式 None 当成"省略参数"
+    再次自查，教师概览等列表就会为每个无活动记录的会话额外发一次查询
+    （N+1）。显式 None 必须保留改前行为：不查询，按现有规则回退到
+    started_at。
+    """
+    from services import session_lifecycle
+
+    calls = []
+
+    def fake_latest_activity(ids):
+        calls.append(list(ids))
+        return {}
+
+    monkeypatch.setattr(
+        session_lifecycle, "latest_session_activity", fake_latest_activity
+    )
+
+    payload = session_lifecycle.session_lifecycle_payload(
+        _session(),  # started_at = UTC_NOW - 5min
+        now=UTC_NOW,
+        last_activity_at=None,
+    )
+
+    assert calls == []  # 显式 None 与省略参数不同：不执行查询
+    # 回退到改前既有规则：以 started_at 作为活动时间
+    assert payload["last_activity_at"] == (UTC_NOW - timedelta(minutes=5)).strftime("%Y-%m-%dT%H:%M:%SZ")
+    assert 290 <= payload["activity_age_seconds"] <= 310