feat(server): ingest v2事件级归属+session_resume+stay区间并集重算+响应明细契约

Co-Authored-By: Claude Code <noreply@anthropic.com>

server/tracker/api/collect.py

@@ -1,6 +1,8 @@
 import json
 from datetime import datetime, timezone as dt_tz
 from urllib.parse import urlsplit
+from uuid import uuid4
+from django.utils import timezone as dj_timezone
 from rest_framework.views import APIView
 from rest_framework.response import Response
 from rest_framework.permissions import AllowAny
@@ -42,20 +44,21 @@ class CollectBatchView(APIView):
     authentication_classes = []
 
     def post(self, request):
+        request_id = uuid4().hex[:32]   # 每次接收一个：响应回传 + RawBatch 同值落库（对账/去重排查）
         origin = request.headers.get("Origin") or ""
         cl = request.headers.get("Content-Length") or "0"
         if not cl.isdecimal() or int(cl) > 262144:   # isdecimal：isdigit 放过 "²"（int 仍崩），isdecimal 与 int() 精确对齐
-            return self._deny(400, "payload_too_large")   # 应用层兜底（nginx 另设 client_max_body_size）
+            return self._deny(400, "payload_too_large", request_id)   # 应用层兜底（nginx 另设 client_max_body_size）
         site = self._match_site(request)
         if site is None or not site.is_active:
-            return self._deny(403, "not_allowed")         # 三种拒绝统一 reason，防 site_id 枚举探测
+            return self._deny(403, "not_allowed", request_id)         # 三种拒绝统一 reason，防 site_id 枚举探测
         # urlsplit 提取 host：split("//")[-1] 会被查询串里的白名单 URL 绕过；
         # "null" Origin（隐私上下文/沙箱 iframe）视为无 host，走 allow_no_origin 分支
         host = (urlsplit(origin).hostname or "").lower() if origin and origin != "null" else ""
         if host and host not in [d.lower() for d in site.domains]:
-            return self._deny(403, "not_allowed")
+            return self._deny(403, "not_allowed", request_id)
         if not host and not site.allow_no_origin:
-            return self._deny(403, "not_allowed")
+            return self._deny(403, "not_allowed", request_id)
         payload = request.data if isinstance(request.data, dict) else {}
         # 原始批次快照：须在 occurred_at 字符串转 datetime 之前深拷贝（datetime 不可 JSON 序列化），
         # 且独立于 ingest/validate 的就地清洗（path 截断等）——存的是"到达时"的原文（脱敏后）。
@@ -65,17 +68,20 @@ class CollectBatchView(APIView):
                 try:
                     ev["occurred_at"] = _iso(ev["occurred_at"])
                 except (ValueError, TypeError):
-                    return self._deny(400, "invalid_occurred_at")
+                    return self._deny(400, "invalid_occurred_at", request_id)
         if not payload.get("visitor_id") or not payload.get("session_id"):
-            return self._deny(400, "invalid_payload")
+            return self._deny(400, "invalid_payload", request_id)
         ip = self._client_ip(request)
         r = ingest_batch(site, payload, ip, request.headers.get("User-Agent", ""))
         try:
-            self._save_raw_batch(site, payload, snapshot, r, ip)
+            self._save_raw_batch(site, payload, snapshot, r, ip, request_id)
         except Exception:   # best-effort：原始表任何失败（含孤代理 encode 崩溃）不影响主采集响应
             pass
+        # 响应契约（spec §2.3）：rejected 明细回传，HTTP 200 不再隐含全量成功
         return Response({"code": 200,
-            "data": {"accepted": r["accepted"], "duplicated": r["duplicated"]},
+            "data": {"accepted": r["accepted"], "duplicated": r["duplicated"],
+                     "rejected": r["rejected"], "request_id": request_id,
+                     "server_time": dj_timezone.now().isoformat()},
             "message": "ok"})
 
     def options(self, request):   # 预检兜底（SDK text/plain 简单请求通常不触发）；ACAO 由 finalize_response 统一补
@@ -103,12 +109,13 @@ class CollectBatchView(APIView):
                              "message": response.data["detail"]}
         return response
 
-    def _save_raw_batch(self, site, payload, snapshot, r, ip):
+    def _save_raw_batch(self, site, payload, snapshot, r, ip, request_id=""):
         """原始批次落库（每次接收一行，含被拒批次）。失败由调用方静默吞掉。
 
         rejected（accepted=0 且有拒绝）整批截断 4KB——超批 51 事件的原文必然超限；
         partial（部分拒绝）不截：正常批次本就受 256KB 请求体上限约束，TextField 可容。
         visitor 允许 None：被拒批次（如超批）不会创建访客记录。
+        request_id 与响应同值（对账："这批原始数据对应哪个响应"）。
         """
         body = json.dumps(snapshot, ensure_ascii=False)
         status = ("rejected" if not r["accepted"] and r["rejected"]
@@ -121,7 +128,7 @@ class CollectBatchView(APIView):
             batch_id=str(payload.get("batch_id", ""))[:64],
             body=body, event_count=r["accepted"],
             rejected_count=len(r["rejected"]) if isinstance(r["rejected"], list) else r["rejected"],
-            status=status, ip=ip)
+            status=status, ip=ip, request_id=str(request_id or "")[:64])
 
     def _match_site(self, request):
         sid = (request.data or {}).get("site_id") if isinstance(request.data, dict) else None
@@ -136,5 +143,7 @@ class CollectBatchView(APIView):
             ip = ip[7:]
         return ip
 
-    def _deny(self, code, reason):
-        return Response({"code": code, "data": None, "message": reason}, status=code)   # ACAO/信封由 finalize_response 统一补
+    def _deny(self, code, reason, request_id=""):
+        # data 带 request_id（对账）；ACAO/信封由 finalize_response 统一补
+        return Response({"code": code, "data": {"request_id": request_id},
+                         "message": reason}, status=code)

server/tracker/services/ingest.py

@@ -4,16 +4,26 @@
   ① event 绝不重复入库：uq_site_event + bulk_create(ignore_conflicts=True) + 入库前 existed 预查
      + 同批 event_id 去重（seen_eids）；
   ② 重放命中的事件不参与会话聚合（new_events 只含未入库事件）；
-  ③ visit_count/active_days/event_count/duration_ms/stay_ms 全走 F 表达式，
-     filter().update() 避免 read-modify-write 全量 save() 用旧值覆盖并发赢家；
+  ③ visit_count/active_days/event_count/duration_ms 走 F 表达式，filter().update() 避免
+     read-modify-write 全量 save() 用旧值覆盖并发赢家；stay_ms 是唯一例外——schema v2 起
+     改为落库后按 Event 表重算（v1 active_ms 加法 + v2 page_id 区间并集，见
+     _recompute_session_stay；15s 周期切分的相邻区间须并集拼回，增量累加会重复计），
+     写入仍是单行 filter().update()；
   ④ 会话 ended_at 用 Greatest 只前进不回退，乱序批次不改早结束时间；
   ⑤ 同批切分出的会话段按"原始 sid"缓存复用，后续事件不再切分。
 
+schema v2（阶段A）：批级 schema_version==2 时事件自带归属四元组（site/visitor/session/
+page_id），会话按事件级 session_id 解析（v1 信封不再是唯一依据）；v2 不做 30min 服务端
+兜底切分（会话切换由 SDK 显式 session_resume 驱动，避免双切）；session_resume 创建的
+新会话经 resume_of 链接前会话。v1 路径完全保留。
+
 已接受的残余风险（M1 不修）：existed 预查与 bulk_create 之间存在毫秒级 TOCTOU 窗口——
 并发请求在预查后、写入前提交同一 event_id 时，本请求的 ignore_conflicts 静默吞掉冲突，
 但该事件已被计入 new_events 参与聚合，聚合可能双计。窗口极窄且仅影响并发重放场景。
+（v2 附注：stay_ms 重算对"并发同会话双批"同样存在同类窗口——后提交者的 SELECT 可能
+看不到前者未提交事件而少算，与 TOCTOU 同级、只影响 stay_ms 单字段。）
 """
-from datetime import timedelta
+from datetime import datetime, timedelta, timezone as dt_tz
 
 from django.db import transaction, IntegrityError
 from django.db.models import F, Q
@@ -31,6 +41,35 @@ _CLASS_FIELDS = ("channel", "platform", "variant", "search_term",
                  "evidence_kind", "resolution", "detail")
 
 
+def _occ_dt(v):
+    """v2 事件 occurred_at 允许 ISO 字符串（SDK 线格式；collect 视图已预转 datetime，
+    直连调用方可能传原文）。解析失败抛 ValueError/TypeError——调用方放行原值，
+    由 validate_event 统一拒 invalid_occurred_at。naive 视为 UTC（与 collect._iso 同口径）。"""
+    if isinstance(v, str):
+        d = datetime.fromisoformat(v.replace("Z", "+00:00"))
+        return d if d.tzinfo else d.replace(tzinfo=dt_tz.utc)
+    return v
+
+
+def _ms(v):
+    """v2 区间字段宽松 int 化：bool/非数值 → None（脏数据防御——recompute 会过滤 None 区间，
+    该事件不参与并集但不拒批）。"""
+    if isinstance(v, bool) or not isinstance(v, (int, float)):
+        return None
+    return int(v)
+
+
+def _time_quality(now, occ) -> str:
+    """v2 时间质量分类：received−occurred 差 d 秒——d>+120s 迟到 "late"、d<-120s 时钟倒偏
+    "skew_backwards"、否则 "ok"。v1 事件留空（late 布尔已有，不动）。"""
+    d = (now - occ).total_seconds()
+    if d > C.CLOCK_SKEW_LATE_SEC:
+        return "late"
+    if d < -C.CLOCK_SKEW_LATE_SEC:
+        return "skew_backwards"
+    return "ok"
+
+
 def _clean_utm(d):
     """utm 键白名单（utm_* 前缀）+ 截断——任意字符串不得经 utm 入库渲染。"""
     return {k[:50]: str(v)[:100] for k, v in (d or {}).items()
@@ -134,17 +173,26 @@ def _touch_visitor(site, vid, ctx, ip):
     return v, created   # created 供首触来源分类（ingest_batch 消费）
 
 
-def _get_session(site, visitor, sid, ctx, occurred_at, ua="", referrer_url=""):
+def _get_session(site, visitor, sid, ctx, occurred_at, ua="", referrer_url="",
+                 skip_idle_split=False):
     """会话权威在 SDK；服务端兜底：同 session_id 族（s1、s1#2…）距最后事件>30min 切分后缀段。
 
     空闲比较必须取整个会话族的最大 ended_at：若只查根段（session_id=sid 精确匹配），
     切分一次后根段 ended_at 陈旧，后续小间隔批次会反复误判超时、每批新开段（visit_count 灌倍）。
+
+    skip_idle_split=True（schema v2）：SDK 显式管理会话（session_resume 驱动切换），服务端
+    兜底不介入（避免双切）；"复用现有段"能力保留——直接按 sid 精确查/建（uq_site_session
+    保证至多一行），不复用 sid#N 族。
     """
     idle = timedelta(milliseconds=C.SESSION_IDLE_MS)
-    fam = list(VisitSession.objects.filter(
-        Q(session_id=sid) | Q(session_id__startswith=sid + "#"), site=site))
-    last = max(fam, key=lambda s: s.ended_at) if fam else None
-    if last and occurred_at - last.ended_at > idle:
+    if skip_idle_split:
+        fam = []
+        last = VisitSession.objects.filter(site=site, session_id=sid).first()
+    else:
+        fam = list(VisitSession.objects.filter(
+            Q(session_id=sid) | Q(session_id__startswith=sid + "#"), site=site))
+        last = max(fam, key=lambda s: s.ended_at) if fam else None
+    if last and not skip_idle_split and occurred_at - last.ended_at > idle:
         suffix = 2 + sum(1 for s in fam if s.session_id.startswith(sid + "#"))
         sid, sess = f"{sid}#{suffix}", None
     else:
@@ -173,6 +221,39 @@ def _get_session(site, visitor, sid, ctx, occurred_at, ua="", referrer_url=""):
     return sess
 
 
+def _recompute_session_stay(sess):
+    """stay_ms 唯一聚合路径（invariant ③ 唯一例外）：落库后从 Event 表重算回写。
+
+    v1/v2 混合的单一解释：v1 事件（page_id 空）沿用 active_ms 加法；v2 事件按 page_id
+    分组做 (active_from_ms, active_to_ms) 区间并集后求和；两者相加。重算式而非增量——
+    SDK 每 15s 切一刀的相邻区间须并集拼回（增量累加会重复计）；uq_site_event 存储层
+    去重保证重算不重复。缺区间/倒置区间的 v2 脏事件整条跳过，不参与并集也不拒批。
+    v1-only 会话重算值与旧增量语义恒等（v1_total 即历史 active_ms 全和）。
+    """
+    v1_total, by_page = 0, {}
+    for page_id, af, at, props in Event.objects.filter(
+            session=sess, event_name="page_stay").values_list(
+            "page_id", "active_from_ms", "active_to_ms", "properties"):
+        if page_id:
+            # 倒置区间（to<from）一并跳过：参与求和会得负值，炸 PositiveBigInteger 列
+            if af is not None and at is not None and at >= af:
+                by_page.setdefault(page_id, []).append((af, at))
+        else:
+            v1_total += int((props or {}).get("active_ms", 0) or 0)
+    union_total = 0
+    for ivs in by_page.values():
+        ivs.sort()
+        cur_s, cur_e = ivs[0]
+        for s, e in ivs[1:]:
+            if s <= cur_e:                      # 相邻/重叠区间拼接（15s 周期切一刀的场景）
+                cur_e = max(cur_e, e)
+            else:
+                union_total += cur_e - cur_s
+                cur_s, cur_e = s, e
+        union_total += cur_e - cur_s
+    VisitSession.objects.filter(pk=sess.pk).update(stay_ms=v1_total + union_total)
+
+
 @transaction.atomic
 def ingest_batch(site, payload: dict, ip: str | None = None, ua: str = "") -> dict:
     now = timezone.now()
@@ -198,9 +279,16 @@ def ingest_batch(site, payload: dict, ip: str | None = None, ua: str = "") -> di
                             referrer_url=_referrer_url(ctx.get("referrer_domain"), pv_referrer),
                             utm=_clean_utm(ctx.get("utm")),
                             landing_path=str(ctx.get("path") or ""), ua=ua or "")
+    v2 = payload.get("schema_version") == 2   # 批级信（缺省 = v1 旧 SDK，路径完全不变）
     ready, session_cache, seen_eids = [], {}, set()
     for i, ev in enumerate(events):
-        reason = validate_event(ev)
+        if v2 and isinstance(ev, dict) and isinstance(ev.get("occurred_at"), str):
+            try:    # v2 事件 occurred_at 允许 ISO 字符串（SDK 线格式）；解析失败放行原值，
+                ev["occurred_at"] = _occ_dt(ev["occurred_at"])   # validate_event 统一拒 invalid_occurred_at
+            except (ValueError, TypeError):
+                pass
+        reason = validate_event(ev, v2=v2, batch_site_id=site.site_id,
+                                batch_visitor_id=payload["visitor_id"])
         if reason:
             result["rejected"].append({"index": i, "event_id": ev.get("event_id", ""),
                                        "reason": reason})
@@ -210,13 +298,33 @@ def ingest_batch(site, payload: dict, ip: str | None = None, ua: str = "") -> di
             continue
         seen_eids.add(ev["event_id"])
         occ = ev["occurred_at"]
-        sid = payload["session_id"]
+        sid = ev["session_id"] if v2 else payload["session_id"]   # v2：归属按事件级字段解析
         sess = session_cache.get(sid)
         if not sess:
             sess = _get_session(site, visitor, sid, ctx, occ, ua=ua or "",
-                                referrer_url=pv_referrer)
+                                referrer_url=pv_referrer, skip_idle_split=v2)
             session_cache[sid] = sess      # 按原始 sid 存：切分后 sess.session_id 是 "s1#2"，
                                            # 若按它存，同批后续事件按 "s1" 查会 miss → 每条再切新会话
+        if v2 and ev["event_name"] == "session_resume" and sess.resume_of_id is None:
+            # 续接链：resume_of 指向 prev_session_id 对应会话；prev 无效/不存在软处理（None，不拒）。
+            # resume_of__isnull=True 条件幂等——重放/同批多条 resume 不覆写已设的链。
+            prev = (ev.get("properties") or {}).get("prev_session_id")
+            prev_sess = (VisitSession.objects.filter(site=site, session_id=prev).first()
+                         if isinstance(prev, str) and prev else None)
+            if prev_sess and prev_sess.pk != sess.pk:
+                VisitSession.objects.filter(pk=sess.pk, resume_of__isnull=True).update(
+                    resume_of=prev_sess)
+                sess.resume_of = prev_sess   # 刷新内存态：同批同 sid 后续 resume 不再重查重设
+        extra = {}
+        if v2:
+            props = ev.get("properties") or {}
+            extra = {"page_id": str(ev.get("page_id") or "")[:64],
+                     "active_from_ms": _ms(props.get("active_from")),
+                     "active_to_ms": _ms(props.get("active_to")),
+                     "flush_reason": str(props.get("flush_reason") or "")[:16],
+                     "time_quality": _time_quality(now, occ),
+                     "sdk_version": str(payload.get("sdk_version") or "")[:16],
+                     "schema_version": 2}
         ready.append(Event(
             site=site, visitor=visitor, session=sess,
             event_id=ev["event_id"], event_name=ev["event_name"],
@@ -225,7 +333,7 @@ def ingest_batch(site, payload: dict, ip: str | None = None, ua: str = "") -> di
             properties=ev.get("properties") or {},
             ip=ip, late=abs((now - occ).total_seconds()) > C.CLOCK_SKEW_LATE_SEC,
             country=_geo_field(geo, "country")[:64], province=_geo_field(geo, "province")[:64],
-            city=_geo_field(geo, "city")[:64], is_bot=is_bot))
+            city=_geo_field(geo, "city")[:64], is_bot=is_bot, **extra))
     if ready:
         existed = set(Event.objects.filter(
             site=site, event_id__in=[e.event_id for e in ready]
@@ -238,11 +346,9 @@ def ingest_batch(site, payload: dict, ip: str | None = None, ua: str = "") -> di
             se = [e for e in new_events if e.session.pk == sess.pk]
             new_end = max(x.occurred_at for x in se)
             span_ms = int((new_end - min(x.occurred_at for x in se)).total_seconds() * 1000)
-            stay_add = sum(int(e.properties.get("active_ms", 0) or 0)
-                           for e in se if e.event_name == "page_stay")
             VisitSession.objects.filter(pk=sess.pk).update(
                 ended_at=Greatest(F("ended_at"), new_end),   # 只前进不回退，防乱序批次改早结束时间
                 event_count=F("event_count") + len(se),
-                duration_ms=F("duration_ms") + span_ms,
-                stay_ms=F("stay_ms") + stay_add)             # 有效停留单独累计，供访客"总停留"取数
+                duration_ms=F("duration_ms") + span_ms)
+            _recompute_session_stay(sess)   # stay_ms 走重算（v1 加法 + v2 区间并集），不再 F 增量
     return result

server/tracker/services/validation.py

@@ -77,14 +77,28 @@ def sanitize_payload(obj):
     return obj
 
 
-def validate_event(ev) -> str | None:
+def validate_event(ev, *, v2=False, batch_site_id="", batch_visitor_id="") -> str | None:
     """返回 None 表示通过；否则返回拒绝原因码。类型/白名单/体积/深度全在此守门。
     通过时会就地截断/清洗并写回 ev（path 先 sanitize_url 再截 MAX_PATH_LEN，title 截
-    MAX_TITLE_LEN）——Task 6 的 ingest 依赖此写回契约。"""
+    MAX_TITLE_LEN）——Task 6 的 ingest 依赖此写回契约。
+
+    v2=True（批级 schema_version==2）：额外校验事件级归属四元组——site_id/visitor_id
+    必须与批级一致、session_id 非空 str（归属依据从 v1 信封转移到事件级）；
+    page_id 必须为非空 ≤64 的合法 UTF-8（session_resume 可空——它归属会话而非页面）。
+    不一致 → attribution_mismatch；page_id 非法 → invalid_page_id。"""
     if not isinstance(ev, dict):
         return "event_invalid_shape"
     if ev.get("event_name") not in C.EVENT_NAMES:
         return "event_name_not_allowed"
+    if v2:
+        if (ev.get("site_id") != batch_site_id
+                or ev.get("visitor_id") != batch_visitor_id
+                or not isinstance(ev.get("session_id"), str) or not ev["session_id"]):
+            return "attribution_mismatch"
+        pid = ev.get("page_id")
+        if ev["event_name"] != "session_resume" and (
+                not isinstance(pid, str) or not pid or len(pid) > 64 or not _utf8_ok(pid)):
+            return "invalid_page_id"
     eid = ev.get("event_id")
     if not isinstance(eid, str) or not eid or len(eid) > 64 or not _utf8_ok(eid):
         return "invalid_event_id"

server/tracker/tests/test_collect.py

@@ -24,7 +24,11 @@ def _body(n=1, sid="s1", vid="v1"):
 def test_collect_ok_creates_events(db, client):
     Site.objects.create(name="s", domains=["www.a.com"], site_id="site_x")
     r = _post(client, _body(2))
-    assert r.status_code == 200 and r.json() == {"code": 200, "data": {"accepted": 2, "duplicated": 0}, "message": "ok"}
+    assert r.status_code == 200
+    body = r.json()   # 包含式断言：v2 响应新增 rejected/request_id/server_time 字段
+    assert body["code"] == 200 and body["message"] == "ok"
+    assert body["data"]["accepted"] == 2 and body["data"]["duplicated"] == 0
+    assert body["data"]["rejected"] == [] and body["data"]["request_id"]
     assert r["Access-Control-Allow-Origin"] == "https://www.a.com"
 
 def test_origin_not_whitelisted_rejected(db, client):
@@ -97,3 +101,27 @@ def test_malformed_content_length_400_not_500(db, client):
     Site.objects.create(name="s", domains=["www.a.com"], site_id="site_x")
     r = _post(client, _body(1), extra={"CONTENT_LENGTH": "abc"})
     assert r.status_code == 400 and r.json()["message"] == "payload_too_large"
+
+
+# ---- 响应契约（阶段A Task3）----
+
+def test_response_contract_rejected_details(db, client):
+    """部分拒绝批：rejected 明细（index/event_id/reason）回传——HTTP 200 不再隐含全量成功。"""
+    Site.objects.create(name="s", domains=["www.a.com"], site_id="site_x")
+    b = _body(2)
+    b["events"][0]["event_name"] = "bad_name"
+    r = _post(client, b)
+    assert r.status_code == 200
+    data = r.json()["data"]
+    assert data["accepted"] == 1 and data["duplicated"] == 0
+    assert data["rejected"] == [{"index": 0, "event_id": "e0",
+                                 "reason": "event_name_not_allowed"}]
+
+def test_request_id_in_response_and_raw_batch(db, client):
+    """request_id（uuid4 hex 32）随响应回传，且与 RawBatch.request_id 同值落库（对账）。"""
+    from tracker.models import RawBatch
+    Site.objects.create(name="s", domains=["www.a.com"], site_id="site_x")
+    r = _post(client, _body(1))
+    rid = r.json()["data"]["request_id"]
+    assert isinstance(rid, str) and len(rid) == 32
+    assert RawBatch.objects.get().request_id == rid

server/tracker/tests/test_ingest.py

@@ -217,3 +217,136 @@ def test_device_model_backfilled_on_later_batch(db):
     ingest_batch(site, b3)
     v.refresh_from_db()
     assert v.device_model == "Pixel 7"                  # 已有值不再覆盖
+
+
+# ---- schema v2（阶段A Task3：事件级归属 / session_resume / 区间并集 / 时间质量）----
+
+SITE_ID = "site_v2"
+
+def _now_iso():
+    return timezone.now().isoformat()
+
+def _v2_event(name, page_id, sid, seq=1, **props):
+    return {"event_id": f"e-{name}-{seq}", "event_name": name,
+            "occurred_at": _now_iso(), "path": "/p", "title": "t",
+            "site_id": SITE_ID, "visitor_id": "v1", "session_id": sid,
+            "page_id": page_id, "seq": seq, "properties": props}
+
+def _v2_batch(events, vid="v1", sid="s1"):
+    return {"schema_version": 2, "site_id": SITE_ID, "visitor_id": vid,
+            "session_id": sid, "sdk_version": "0.2.0",
+            "context": {"referrer_domain": "", "path": "/", "device_type": "PC"},
+            "events": events}
+
+# 20. v2 事件级归属：同批多 session 各自落库（v1 信封 session_id 不再是唯一依据）；
+#     事件级 site/visitor 与批级不一致 → attribution_mismatch
+def test_v2_event_level_attribution(db):
+    site = Site.objects.create(name="s", domains=["a.com"], site_id=SITE_ID)
+    evs = [_v2_event("page_view", "p1", "sA", seq=1),
+           _v2_event("page_view", "p2", "sB", seq=2)]
+    r = ingest_batch(site, _v2_batch(evs, sid="sA"))
+    assert (r["accepted"], r["duplicated"], r["rejected"]) == (2, 0, [])
+    assert VisitSession.objects.count() == 2
+    for sid_name, pid in (("sA", "p1"), ("sB", "p2")):
+        sess = VisitSession.objects.get(session_id=sid_name)
+        e = Event.objects.get(session=sess)
+        assert e.page_id == pid and e.schema_version == 2
+    bad_v = _v2_event("page_view", "p3", "sA", seq=3)
+    bad_v["visitor_id"] = "v-other"                     # 事件级 visitor ≠ 批级
+    bad_s = _v2_event("page_view", "p4", "sA", seq=4)
+    bad_s["site_id"] = "site_other"                     # 事件级 site ≠ 批级
+    r2 = ingest_batch(site, _v2_batch(
+        [bad_v, bad_s, _v2_event("page_view", "p5", "sA", seq=5)], sid="sA"))
+    assert r2["accepted"] == 1
+    assert r2["rejected"] == [
+        {"index": 0, "event_id": "e-page_view-3", "reason": "attribution_mismatch"},
+        {"index": 1, "event_id": "e-page_view-4", "reason": "attribution_mismatch"}]
+
+# 21. session_resume：新会话 resume_of 链接 prev_session_id 对应会话；
+#     prev 不存在软处理（resume_of=None 不拒）；v2 不做 30min 兜底切分
+def test_session_resume_creates_linked_session(db):
+    site = Site.objects.create(name="s", domains=["a.com"], site_id=SITE_ID)
+    ingest_batch(site, _v2_batch([_v2_event("page_view", "p1", "s1", seq=1)], sid="s1"))
+    r = ingest_batch(site, _v2_batch([
+        _v2_event("session_resume", "", "s2", seq=2, prev_session_id="s1", reason="idle_timeout"),
+        _v2_event("page_view", "p2", "s2", seq=3)], sid="s2"))
+    assert (r["accepted"], r["rejected"]) == (2, [])
+    s1 = VisitSession.objects.get(session_id="s1")
+    s2 = VisitSession.objects.get(session_id="s2")
+    assert s2.resume_of == s1
+    r2 = ingest_batch(site, _v2_batch([
+        _v2_event("session_resume", "", "s3", seq=4, prev_session_id="ghost", reason="idle_timeout"),
+        _v2_event("page_view", "p3", "s3", seq=5)], sid="s3"))
+    assert (r2["accepted"], r2["rejected"]) == (2, [])   # prev 无效不拒
+    assert VisitSession.objects.get(session_id="s3").resume_of is None
+    assert Visitor.objects.get(visitor_id="v1").visit_count == 3   # s1/s2/s3
+    # 35min 后同 sid 再发：不切 s2#2（v2 会话切换由 SDK 显式 resume 驱动）
+    late_ev = _v2_event("page_view", "p2b", "s2", seq=6)
+    late_ev["occurred_at"] = (timezone.now() + timedelta(minutes=35)).isoformat()
+    r3 = ingest_batch(site, _v2_batch([late_ev], sid="s2"))
+    assert r3["accepted"] == 1
+    assert VisitSession.objects.filter(session_id__startswith="s2").count() == 1
+
+# 22. v2 page_stay 区间并集（非加法）：[0,15][15,30][40,55]s → 0-30 ∪ 40-55 = 45s
+def test_stay_interval_union_not_sum(db):
+    site = Site.objects.create(name="s", domains=["a.com"], site_id=SITE_ID)
+    secs = [(0, 15), (15, 30), (40, 55)]
+    evs = [_v2_event("page_stay", "p1", "s1", seq=i + 1,
+                     active_from=a * 1000, active_to=b * 1000, flush_reason="periodic")
+           for i, (a, b) in enumerate(secs)]
+    r = ingest_batch(site, _v2_batch(evs, sid="s1"))
+    assert (r["accepted"], r["rejected"]) == (3, [])
+    e1 = Event.objects.get(event_id="e-page_stay-1")
+    assert (e1.active_from_ms, e1.active_to_ms) == (0, 15000)
+    assert e1.flush_reason == "periodic" and e1.sdk_version == "0.2.0"
+    assert VisitSession.objects.get(session_id="s1").stay_ms == 45000  # 非 sum=60000
+
+# 23. v1 路径完全不变：stay_ms 仍为 active_ms 加法，v2 新字段留空
+def test_v1_events_unchanged_path(db):
+    site = Site.objects.create(name="s", domains=["a.com"])
+    ingest_batch(site, _batch([_ev(1), _ev(2, "page_stay", properties={"active_ms": 5000, "path": "/p"})]))
+    ingest_batch(site, _batch([_ev(3, "page_stay", properties={"active_ms": 3000, "path": "/p"})]))
+    sess = VisitSession.objects.get()
+    assert sess.stay_ms == 8000 and sess.event_count == 3
+    e = Event.objects.get(event_id="e1")
+    assert e.schema_version is None and e.page_id == "" and e.time_quality == ""
+    assert e.active_from_ms is None and e.active_to_ms is None
+
+# 24. time_quality 分类：正常 ok / >120s 过去 late / 未来小偏移 ok / 未来超阈 skew_backwards
+def test_time_quality_classification(db):
+    site = Site.objects.create(name="s", domains=["a.com"], site_id=SITE_ID)
+    cases = [(1, 0, "ok"), (2, -130, "late"), (3, 5, "ok"), (4, 130, "skew_backwards")]
+    evs = []
+    for seq, off, _ in cases:
+        e = _v2_event("page_view", f"p{seq}", "s1", seq=seq)
+        e["occurred_at"] = (timezone.now() + timedelta(seconds=off)).isoformat()
+        evs.append(e)
+    r = ingest_batch(site, _v2_batch(evs, sid="s1"))
+    assert (r["accepted"], r["rejected"]) == (4, [])
+    for seq, off, want in cases:
+        e = Event.objects.get(event_id=f"e-page_view-{seq}")
+        assert e.time_quality == want, f"seq={seq} offset={off}s"
+        assert e.late == (abs(off) > 120)               # late 布尔语义不动
+
+# 25. 混合解释：同会话 v1 active_ms 加法 + v2 区间并集 相加（单一解释）
+def test_recompute_mixed_v1_v2(db):
+    site = Site.objects.create(name="s", domains=["a.com"], site_id=SITE_ID)
+    ingest_batch(site, _batch(
+        [_ev(1), _ev(2, "page_stay", properties={"active_ms": 10000, "path": "/p"})], sid="sm"))
+    evs = [_v2_event("page_stay", "px", "sm", seq=1, active_from=0, active_to=10000, flush_reason="periodic"),
+           _v2_event("page_stay", "py", "sm", seq=2, active_from=0, active_to=5000, flush_reason="periodic")]
+    r = ingest_batch(site, _v2_batch(evs, sid="sm"))
+    assert (r["accepted"], r["rejected"]) == (2, [])
+    assert VisitSession.objects.get(session_id="sm").stay_ms == 10000 + 15000  # v1 SUM + v2 并集
+
+# 26. 脏区间防御：非数值/倒置区间不参与并集、不拒批（防 stay_ms 负值炸 PositiveBigInteger）
+def test_dirty_intervals_soft_skipped(db):
+    site = Site.objects.create(name="s", domains=["a.com"], site_id=SITE_ID)
+    evs = [_v2_event("page_stay", "p1", "s1", seq=1, active_from=0, active_to=10000, flush_reason="periodic"),
+           _v2_event("page_stay", "p2", "s1", seq=2, active_from="x", active_to=9000, flush_reason="periodic"),
+           _v2_event("page_stay", "p3", "s1", seq=3, active_from=20000, active_to=10000, flush_reason="periodic")]
+    r = ingest_batch(site, _v2_batch(evs, sid="s1"))
+    assert (r["accepted"], r["rejected"]) == (3, [])   # 脏数据软处理不拒批
+    e2 = Event.objects.get(event_id="e-page_stay-2")
+    assert e2.active_from_ms is None                   # 非数值 int 化为 None
+    assert VisitSession.objects.get(session_id="s1").stay_ms == 10000  # 仅 p1 有效区间