fangke/docs/superpowers/plans/2026-09-18-collection-foundation-phase-a.md

采集底座修复（阶段A）实现计划

For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking.

Goal: 修复采集底座——事件级归属快照（消灭孤立停留）、区间制停留（单调时钟/15s 结算）、版本可诊断、URL 清洗三轨制，SDK 0.1.0→0.2.0 + 服务端 schema v2 兼容 v1。

Architecture: SDK 引入 PageSession 上下文（事件入队时冻结 site/visitor/session/page 四元组，flush 用快照拆子批）；停留改 performance.now 区间制；服务端按事件级归属落库、stay_ms 重算式回写（区间并集）、清洗升级三轨制（参数名黑名单 + 归因白名单 + 值判别），TS/Python 共享单源测试向量。

Tech Stack: sdk/ = TypeScript + vitest + esbuild（IIFE，~3KB gzip）；server/ = Django 5 + DRF + MySQL（测试 sqlite），pytest。

Spec: docs/superpowers/specs/2026-09-18-collection-foundation-phase-a-design.md（含全部决策与边界）

执行顺序约束： T1-T4（server）必须先于 T11 部署完成；T5-T10（SDK）依赖 T2 的契约常量对齐；T11-T12 收尾部署。

测试命令： server 在 server/ 下 python -m pytest tracker/tests/ -x -q（142 个基线，随任务递增）；sdk 在 sdk/ 下 npx vitest run（58 个基线）。

红线（沿袭 spec §4.5）： SDK 任何回调异常不得冒泡到宿主页；服务端 ingest 失败不得丢整批（savepoint 隔离既有模式）。

---

Task 1: 清洗三轨制 + 共享测试向量（server）

Files:

• Create: server/tracker/tests/fixtures/sanitize_vectors.json

• Modify: server/tracker/constants.py（新增三轨制常量与规则版本）

• Modify: server/tracker/services/validation.py:19-24（sanitize_url 重写）

• Test: server/tracker/tests/test_validation.py（新增向量驱动的测试）

• Step 1: 写共享测试向量 fixtures

server/tracker/tests/fixtures/sanitize_vectors.json（单源文件，SDK T9 复用同一份）：

{
  "rule_version": "v2",
  "vectors": [
    {"in": "/product?id=456", "out": "/product?id=456"},
    {"in": "/product?id=456&email=a@b.com", "out": "/product?id=456"},
    {"in": "/page?phone=13800138000", "out": "/page"},
    {"in": "/lp?utm_source=meta&utm_campaign=x", "out": "/lp?utm_source=meta&utm_campaign=x"},
    {"in": "/lp?gclid=EAIaIQ&fbclid=IwAR2x", "out": "/lp?gclid=EAIaIQ&fbclid=IwAR2x"},
    {"in": "/p?token=abcdef0123456789abcdef0123456789x", "out": "/p"},
    {"in": "/p?ref=abcdefghijklmnopqrstuvwxyz012345", "out": "/p"},
    {"in": "/p?slug=abcdefghijklmnopqrstuvwxyz0123456", "out": "/p"},
    {"in": "/p?cat=sedans&sort=price", "out": "/p?cat=sedans&sort=price"},
    {"in": "/p?q=%E5%BC%A0%E4%B8%89%E7%9A%84%E9%82%AE%E7%AE%B1a%40b.com", "out": "/p"},
    {"in": "/a/b?name=zhangsan", "out": "/a/b"},
    {"in": "/view?v=abc123DEF-_456", "out": "/view?v=abc123DEF-_456"},
    {"in": "/long?v=aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", "out": "/long"},
    {"in": "https://user:pass@example.com/p?x=1&email=c@d.com#frag", "out": "https://example.com/p?x=1"}
  ]
}

（执行者注意：vectors 覆盖四类——业务参数保留、黑名单参数删、值判别删（@/长随机/超长）、userinfo+fragment 清理。x=1 值"1"是短数字→保留。slug 安全线定 32：ref=abcdefghijklmnopqrstuvwxyz012345（32 连续字母数字）按随机 token 删——真实业务 slug 含 ._- 分隔或更短，32+ 连续字母数字视为 token（D08 威胁模型），33-64 band 由新增的 slug= 向量钉死语义。）

• Step 2: 写失败测试（向量驱动）

server/tracker/tests/test_validation.py 追加：

import json, pathlib
from tracker import constants as C
VECTORS = json.loads((pathlib.Path(__file__).parent / "fixtures" / "sanitize_vectors.json").read_text(encoding="utf-8"))

def test_sanitize_url_vectors():
    for v in VECTORS["vectors"]:
        assert sanitize_url(v["in"]) == v["out"], v["in"]

def test_sanitize_rule_version_exported():
    assert C.SANITIZE_RULE_VERSION == VECTORS["rule_version"]

• Step 3: 跑测试确认失败

Run: cd server && python -m pytest tracker/tests/test_validation.py -x -q Expected: FAIL（SANITIZE_RULE_VERSION 不存在 / userinfo 未剥离）

• Step 4: 实现——constants.py 追加

SANITIZE_RULE_VERSION = "v2"
# 三轨制（spec §2.4，slug 安全线以本计划为准=32）：①参数名黑名单整键删；
# ②归因白名单提取保留；③其余按值判别：纯数字≤10位或 slug≤32 保留，其余删。
URL_PARAM_DROP_PATTERNS = ("email", "phone", "mobile", "tel", "name", "user",
                           "token", "key", "password", "verif", "passwd", "idcard")
ATTRIBUTION_PARAM_WHITELIST = {"gclid", "gbraid", "wbraid", "fbclid", "msclkid",
                               "yclid", "gad_source", "gad_campaignid", "dclid"}
_SAFE_VALUE = re.compile(r"^(?:\d{1,10}|[A-Za-z0-9][A-Za-z0-9._-]{0,30})$")  # slug≤32：32+连续字母数字视为token删

（constants.py 顶部需 import re。）

• Step 5: 实现——validation.py 的 sanitize_url 重写

def sanitize_url(url: str) -> str:
    """三轨制清洗（spec §2.4）：黑名单参数名删 + 归因白名单提 + 其余按值判别。
    另：userinfo 剥离、fragment 删除。规则版本 SANITIZE_RULE_VERSION。"""
    p = urlparse(url)
    kept = []
    for k, v in parse_qsl(p.query, keep_blank_values=True):
        if _URL_DROP_KEY.fullmatch(k):
            continue
        if k.lower() in C.ATTRIBUTION_PARAM_WHITELIST or k.lower().startswith("utm_"):
            kept.append((k, v)); continue
        if C._SAFE_VALUE.match(v or ""):
            kept.append((k, v))
    netloc = p.hostname or ""
    if p.port: netloc = f"{netloc}:{p.port}"
    return p._replace(netloc=netloc, query=urlencode(kept), fragment="").geturl()

（注意 _URL_DROP_KEY 正则要重建以覆盖新黑名单——它在模块头 join constants。）

• Step 6: 跑测试通过 + 全量回归

Run: python -m pytest tracker/tests/ -q Expected: 全 PASS（既有 sanitize 测试如断言旧行为需同步更新——保持语义：黑名单参数仍删）

• Step 7: Commit

git add server/tracker/constants.py server/tracker/services/validation.py server/tracker/tests/
git commit -m "feat(server): URL清洗三轨制+共享测试向量（D08）"

---

Task 2: 契约扩展 + 模型迁移

Files:

• Modify: server/tracker/constants.py:1,13-17（白名单）

• Modify: server/tracker/models.py（Event/VisitSession/RawBatch 新字段）

• Create: migration（python manage.py makemigrations tracker）

• Test: server/tracker/tests/test_models.py

• Step 1: constants.py 契约更新

EVENT_NAMES = {"page_view", "page_stay", "sdk_loaded", "session_resume"}
EVENT_PROP_WHITELIST = {
    "page_view": {"referrer", "referrer_domain", "utm", "nav_type", "prerendering"},
    "page_stay": {"active_ms", "path", "active_from", "active_to", "flush_reason", "measurement_version"},
    "sdk_loaded": {"sdk_version", "build_id", "schema_version", "config_version"},
    "session_resume": {"prev_session_id", "reason"},
}
SCHEMA_VERSION = 2  # 批级信缺省 = v1（旧SDK）

• Step 2: 写失败测试（test_models.py 追加字段存在性+约束）

def test_event_has_phase_a_fields():
    e = Event(**_base_event_kwargs(), page_id="p1", active_from_ms=0,
              active_to_ms=15000, flush_reason="periodic",
              time_quality="ok", sdk_version="0.2.0", schema_version=2)
    e.full_clean()  # 验证字段与长度约束

def test_visit_session_resume_of_self_fk():
    assert VisitSession._meta.get_field("resume_of").related_model is VisitSession

• Step 3: 跑失败 → makemigrations 前先加模型字段

models.py：Event 追加（全部 nullable，旧行=legacy）：

page_id = models.CharField(max_length=64, blank=True, default="", db_index=True)
active_from_ms = models.BigIntegerField(null=True, blank=True)
active_to_ms = models.BigIntegerField(null=True, blank=True)
flush_reason = models.CharField(max_length=16, blank=True, default="")
time_quality = models.CharField(max_length=16, blank=True, default="")
sdk_version = models.CharField(max_length=16, blank=True, default="")
schema_version = models.PositiveSmallIntegerField(null=True, blank=True)

VisitSession 追加 resume_of = models.ForeignKey("self", null=True, blank=True, on_delete=models.SET_NULL, related_name="resumes")；RawBatch 追加 request_id = models.CharField(max_length=64, blank=True, default="")。

• Step 4: python manage.py makemigrations tracker && python manage.py migrate（测试库自动建）

• Step 5: 全量回归 + Commit

git add server/tracker/constants.py server/tracker/models.py server/tracker/migrations/
git commit -m "feat(server): schema v2契约——session_resume白名单/Event归属与区间字段/会话resume_of/request_id"

---

Task 3: ingest v2 路径 + 响应契约

Files:

• Modify: server/tracker/services/ingest.py（核心改造）

• Modify: server/tracker/services/validation.py:67+（validate_event 扩展 v2 字段校验）

• Modify: server/tracker/api/collect.py:44-79（request_id + 响应明细）

• Test: server/tracker/tests/test_ingest.py、test_collect.py

• Step 1: 写失败测试（v2 归属，test_ingest.py 追加）

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 test_v2_event_level_attribution():
    # 同批两个 session_id 的事件 → 各自落库到正确会话（v1 信封 session_id 不再唯一依据）
    ...
def test_session_resume_creates_linked_session():
    # resume 事件 → 新 VisitSession，resume_of 指向 prev；prev 无效 → resume_of=None 不拒
    ...
def test_stay_interval_union_not_sum():
    # 同 page_id 三条 page_stay 区间 [0,15][15,30][40,55]（第三条与1、2重叠边界）
    # → VisitSession.stay_ms == 45000（并集），不是 sum=60000 也不是 sum-active_ms
    ...
def test_v1_events_unchanged_path():
    # 无 page_id 的 v1 批 → 走现有路径，stay_ms 仍为 active_ms 加法（混合解释基线）
    ...
def test_time_quality_classification():
    # occurred_at 未来 5s → "ok"；>120s → "late"；负差 → "skew_backwards"
    ...

• Step 2: 跑失败

• Step 3: 实现 ingest_batch 改造（关键逻辑，伪码展开级）

ingest_batch(site, payload, ip, ua):
    v2 = payload.get("schema_version") == 2
    ... 现有 visitor touch / 首触逻辑不变 ...
    for ev in events:
        reason = validate_event(ev, v2=v2)      # v2 时校验事件级四元组与 page_id uuid
        ... 拒绝/同批去重逻辑不变 ...
        # 归属解析：
        if v2: sid, vid = ev["session_id"], ev["visitor_id"]
               （site 匹配已校验；visitor 事件级 == 批级，不一致 → reject "attribution_mismatch"）
               if ev.event_name == "session_resume":
                   prev = props.prev_session_id
                   sess = _get_or_create_session(...新 sid...)
                   sess.resume_of = VisitSession.filter(prev).first() or None（软处理+计数）
               else: sess = _get_session(...)（同现有）
        else:  sid = payload["session_id"]（现有路径不动）
        Event(..., page_id=ev.get("page_id",""), active_from_ms=..., active_to_ms=...,
              flush_reason=..., time_quality=分类(now-occ),
              sdk_version=payload.get("sdk_version","")[:16], schema_version=2 if v2 else None)
    # stay 聚合（v2）：对 ready 中每个受影响 session：
    #   intervals = [(e.active_from_ms, e.active_to_ms) for e in new_events
    #                if e.event_name=="page_stay" and e.page_id]     # v2 区间
    #   v1_add = sum(active_ms for v1 page_stay in session 内 DB 已有？No——)
    #   简化重算：stay_total = Session 内全部 page_stay 的 v1 加法部分（DB 已有 stay_ms_v1
    #   无法区分）→ 方案：VisitSession.stay_ms = DB.stay_ms 不再增量，改由
    #   _recompute_session_stay(session) 全量重算：v1 事件 SUM(active_ms) + v2 事件区间并集。
    #   （量级安全：单会话事件 <几百；每次入库重算一次受影响会话）
    def _recompute_session_stay(sess):
        stays = Event.objects.filter(session=sess, event_name="page_stay")
        total = sum(p.get("active_ms",0) or 0 for e in stays.filter(page_id="")
                    for p in [e.properties])
        by_page = defaultdict(list)
        for e in stays.exclude(page_id=""):
            by_page[e.page_id].append((e.active_from_ms, e.active_to_ms))
        for ivs in by_page.values():
            total += _union_ms(ivs)      # 排序合并区间
        VisitSession.objects.filter(pk=sess.pk).update(stay_ms=total)

（执行者注意：_union_ms = 经典排序合并；duration_ms/event_count/ended_at 沿用现有 F 增量逻辑不变——只有 stay_ms 改重算。v1 路径的 stay_ms 也统一走重算（v1 事件 SUM 语义与原逻辑等价，代码单一路径）。）

• Step 4: 实现 collect.py 改造

• post() 开头 request_id = uuid4().hex[:32]；_save_raw_batch(..., request_id) 落库；响应改为：

return Response({"code": 200, "data": {"accepted": r["accepted"], "duplicated": r["duplicated"],
    "rejected": r["rejected"], "request_id": request_id, "server_time": timezone.now().isoformat()},
    "message": "ok"})

• collect 级拒绝（invalid_payload 等）响应同样带 request_id。

• Step 5: test_collect.py 追加响应契约测试（rejected 明细回传、request_id 存在）→ 全量回归

• Step 6: Commit

git add server/tracker/
git commit -m "feat(server): ingest v2事件级归属+session_resume+stay区间并集重算+响应明细契约"

---

Task 4: 聚合点核查适配 + server 收尾回归

Files:

• 核查（按需修改）: server/tracker/api/dashboard.py:45-51、server/tracker/api/visitors.py:54,88,106

• Test: server/tracker/tests/test_dashboard.py、test_visitors.py

• Step 1: 核查三处聚合的取数来源

• 若读 VisitSession.stay_ms / Visitor.total_stay_ms（ Visitors total 是 SUM over sessions？）→ 重算式回写后自动正确，只加回归测试（v2 区间数据下 avg_stay_ms 数值正确）。

• 若任何一处直接 SUM(Event.properties.active_ms) → 改为读 VisitSession.stay_ms 或并集逻辑。

• 访客详情时间线（visitors.py）若按事件列表展示 → page_stay 多条增量正常显示（区间展示阶段B优化），仅确认不崩。

• Step 2: 追加回归测试：构造 v1+v2 混合会话数据，断言 dashboard avg_stay / visitors total_stay 数值正确（v1 加法 + v2 并集相加）。

• Step 3: server 全量回归（基线 142 + 新增全部通过）

Run: cd server && python -m pytest tracker/tests/ -q Expected: 0 failed

• Step 4: Commit

git add server/tracker/
git commit -m "test(server): v1/v2混合停留聚合回归+聚合点核查适配"

---

Task 5: SDK 版本注入 + 事件归属骨架

Files:

• Modify: sdk/package.json（version 0.2.0）

• Modify: sdk/build.mjs（define 注入）

• Modify: sdk/vitest.config.ts（define 注入测试值——否则 __BUILD_ID__ 在 dispatch 的静默 catch 里 ReferenceError，init 半途而废且零报错）

• Create: sdk/src/types.ts（归属快照类型 + __BUILD_ID__ ambient 声明）

• Test: sdk/test/index.test.ts 适配

• Step 1: types.ts（新建）

/** 事件归属四元组：入队时冻结，flush 不再重读（spec §1.1） */
export interface Attribution {
  site_id: string;
  visitor_id: string;
  session_id: string;
  page_id: string;
  seq: number;
  purpose?: null;      // V1.1 预留
  epoch?: null;        // V1.1 预留
  grant_ref?: null;    // V1.1 预留
}
export const SDK_VERSION = "0.2.0";        // 与 package.json 同步手维护
export const SCHEMA_VERSION = 2;
export const CONFIG_VERSION = "1";

/** build_id 由 esbuild/vitest define 注入的全局常量（ambient 声明，
    不能用模块级 export declare——esbuild 打包时会被擦除导致引用悬空） */
declare global {
  // eslint-disable-next-line no-var
  var __BUILD_ID__: string | undefined;
}
/** 运行时兜底读取（build 注入缺失时降级 "unknown"，绝不抛） */
export const BUILD_ID: string = globalThis.__BUILD_ID__ ?? "unknown";

• Step 2: build.mjs 注入

import { execSync } from "node:child_process";
const buildId = execSync("git log -1 --format=%h-%cd --date=format:%Y%m%d%H%M").toString().trim();
await build({ entryPoints: ["src/index.ts"], bundle: true, minify: true, format: "iife",
  outfile: "dist/v1.js", target: ["es2020"],
  // define 键必须用点链：esbuild 只做精确表达式匹配，裸键 __BUILD_ID__
  // 不会替换 globalThis.__BUILD_ID__ 成员访问（实测 0.24.2），build_id 将恒为 "unknown"
  define: { "globalThis.__BUILD_ID__": JSON.stringify(buildId) } });

（fallback：git 失败时用 "dev-" + Date.now()，try/catch 包裹——构建环境无 git 不得挂。）

vitest.config.ts 同步注入（键同样用点链）：

export default defineConfig({
  define: { "globalThis.__BUILD_ID__": JSON.stringify("vitest") },
  test: { environment: "happy-dom" },
});

• Step 3: 失败测试 → 实现 sdk_loaded 携带版本组 → 全量跑过 → Commit

git add sdk/package.json sdk/build.mjs sdk/src/types.ts sdk/test/
git commit -m "feat(sdk): 版本注入与事件归属骨架（0.2.0/build_id/schema_version）"

---

Task 6: PageSession 模块（归属冻结核心）

Files:

• Create: sdk/src/page.ts

• Test: sdk/test/page.test.ts（新建，fake timers）

• Step 1: 失败测试（核心用例清单）

// vitest vi.useFakeTimers + vi.setSystemTime 模拟
test("事件入队时冻结归属：flush 延迟且期间会话超时，事件仍带旧 session_id")
test("会话恢复：空闲>30min 后交互 → session_resume 事件（prev_session_id/reason）+ 新 sid")
test("恢复检查点三处：交互时 / visibilitychange visible 时 / flush 前")
test("bfcache：pageshow persisted=true → 沿用原 page_id，seq 继续，不产生新 page_view")
test("SPA pushState path 变化 → 新 page_id + 新 page_view；replaceState/query/hash 变化 → 不切段")
test("seq 页内单调递增；hidden 期间不推进活动判定")

• Step 2: 实现 page.ts（结构）

export class PageSession {
  readonly pageId: string;
  private sessionId: string;          // 入队快照用；resume 时换新
  private seqCounter = 0;
  private lastActivity: number;       // performance.now 基准
  // API:
  attribution(): Attribution          // {site,visitor,session:sessionId,page:pageId,seq:++seqCounter}
  touch(): void                       // 活动刷新 + 空闲超时检查（>30min → resume 流程）
  maybeResume(): void                 // visible/flush 前检查点
  close(): void                       // 结算尾部停留（由 stay.ts 回调协作）
}
// 模块级单例 current + startPageSession()/SPA hook 安装
// history.pushState/replaceState/popstate hook：规范化 path（location.pathname，去 query/hash）
// 变化 → current.close() → 新 PageSession → 发 page_view
// pageshow persisted → current 复活标记（不新建）

（identity.ts 的 getSessionId 改造：PageSession 创建时调用一次取 session 并冻结；touchActivity 语义并入 PageSession.touch。）

• Step 3: 跑过 → Commit feat(sdk): PageSession事件级归属冻结+会话恢复+bfcache+SPA切段

---

Task 7: stay.ts 区间制重做

Files:

• Create: sdk/src/stay.ts

• Modify: sdk/src/events.ts（startPageTracking 拆走，保留 navSignals 供 page_view）

• Test: sdk/test/stay.test.ts（新建）

• Step 1: 失败测试（fake timers + fake performance.now）

test("15s 连续活动切 4 条相邻区间，各自 [from,to] 不重叠")
test("hidden 结算当前区间；pagehide 紧随不重发（共享结算状态）")
test("hidden 期间不累计；恢复 visible 后活动从恢复点开新区间")
test("单调时钟：vi.spyOn(performance,'now') 推进，系统墙钟回拨不影响区间长度")
test("flush_reason 正确：periodic/hidden/pagehide/route_change")
test("入队事件携带 active_from/active_to/active_ms/page_id/measurement_version=2")

• Step 2: 实现（核心算法）

区间状态: intervalStart | null（活动有效时非 null）
tick(1s): 若 可见 && now-lastAct<30s: 若 !intervalStart → intervalStart=now-1000
          （宽限对齐现有语义：沿用"可见+30s 交互"判定，measurement_version 标记）
cut(reason): 若 intervalStart: emit page_stay{active_from: epochMs(intervalStart),
           active_to: epochMs(now), active_ms: now-intervalStart, flush_reason: reason,
           measurement_version: "2", ...pageSession.attribution()}; intervalStart=null
setInterval(stayIntervalMs=15000): cut("periodic")
visibilitychange hidden / pagehide: cut(reason) 幂等（intervalStart==null 则 no-op）
epochMs(perfNow): 页面加载墙钟 Date.now() - performance.timeOrigin + perfNow（展示时间）

• Step 3: 跑过 → Commit feat(sdk): 停留改单调时钟区间制——15s切分/共享结算/幂等

---

Task 8: transport 重构（快照信封/拆子批/预算/响应）

Files:

• Modify: sdk/src/transport.ts

• Test: sdk/test/transport.test.ts（扩展）

• Step 1: 失败测试

test("flush 从队首事件快照取归属，不调用 getSessionId（vi.mock identity 断言零调用）")
test("混合归属队列拆子批：两组 (session,page) → 两次独立请求体各含正确信封")
test("子批独立重试：组A 500 失败重试，组B 200 不重试")
test("beacon 超 60KB → 降级 fetch keepalive；再失败丢最旧计 dropped")
test("响应解析：rejected 明细累计进 counts；server_time 记录供 offset 估计")
test("v1 兼容形态不再产生（新 SDK 恒 schema_version:2 信封 + 事件级归属）")

• Step 2: 实现要点

• flush()：取 q 分组 groupEvents(events)（Map key=${visitor}:${session}:${page}，组内保持顺序）→ 每组构造 body {batch_id, site_id, schema_version:2, sdk_version, build_id, visitor_id, session_id, sent_at, context, events}（信封归属=组首事件快照）→ send()/beacon。

• beacon 路径：new Blob([body]).size > 60_000 → 改 fetch keepalive。

• send 成功：await res.json().catch(()=>null) → counts.rejected += data.rejected.length 等；记录 lastServerTime。

• Step 3: 跑过 → Commit feat(sdk): 传输快照信封+混合拆批+beacon预算+响应计数

---

Task 9: SDK 侧 sanitizeUrl 三轨制对齐（共享向量）

Files:

• Modify: sdk/src/context.ts:1-58（URL_PARAM_DROP 表 + sanitizeUrl 重写）

• Modify: sdk/tsconfig.json（resolveJsonModule: true——跨树 JSON import 的 typecheck 支持）

• Test: sdk/test/context.test.ts（追加向量测试）

• Step 1: 失败测试（读 server 的同一份 fixtures）

import vectors from "../../server/tracker/tests/fixtures/sanitize_vectors.json";
test.each(vectors.vectors)("sanitizeUrl vector", (v) => {
  expect(sanitizeUrl(v.in)).toBe(v.out);
});

（in 是保留字不能作解构绑定名——用整对象参数 v。vitest/vite 原生支持 JSON import 且无 root 限制（已核 vitest.config.ts），跨出 sdk/ 的相对路径可解析。）

• Step 2: TS 实现与服务端规则逐条对齐（黑名单 fullmatch、归因白名单+utm_ 前缀、_SAFE_VALUE 同正则、userinfo/fragment 清理——TS 用 new URL() 处理 userinfo，注意相对路径 input 时 URL 不可用 → 沿用字符串切分法与 Python urlparse 行为对齐）

• Step 3: 跑过（含 server 侧同步再跑一遍防漂移）→ Commit feat(sdk): 入队前清洗三轨制对齐服务端（共享向量）

---

Task 10: index.ts 接线 + SDK 全量回归 + 构建

Files:

• Modify: sdk/src/index.ts（PageSession/stay 接线，sdk_loaded 版本组）

• Step 1: 接线：init → startPageSession()（创建 PageSession、安装 SPA hook、pageshow 监听）→ startStayTracking(transport, pageSession) → page_view 携带 attribution → sdk_loaded {sdk_version, build_id, schema_version, config_version}。

• Step 2: 全量测试：cd sdk && npx vitest run（基线 58 + 新增全过；旧断言按新结构更新——归属字段进事件体、信封拆组）

• Step 3: 构建：npm run build（build.mjs）→ dist/v1.js 生成 + gzip 体积记录（目标 <4KB gzip）

• Step 4: Commit feat(sdk): 0.2.0接线——归属快照/区间停留/清洗对齐/响应契约全链路

---

Task 11: server 部署 + v1 线上回归

• Step 1: 本地全量测试最后一遍 + 打包：cd server && python -m pytest tracker/tests/ -q
• Step 2: 部署（标准流程，tar server → scp 经跳板 → 解压 /opt/cms-cj 原地覆盖 → cd deploy && docker compose up -d --build web（自动 migrate）→ 容器状态 + API 探活）
• Step 3: v1 线上回归：hesucar 现网（旧 SDK 缓存）继续上报 24h 观察——RawBatch 正常入库、无 5xx、accepted 率不降（用生产 admin 或 Direct SQL 抽查）
• Step 4: 汇报：容器状态、migrate 输出、回归结果

Task 12: SDK 部署 + 生产验证 + 验收台账

• Step 1: 部署 sdk/dist：tar → scp 经跳板 → 解压覆盖 /opt/cms-cj/sdk/dist（nginx 挂载点，见 memory：目标是 admin/dist 同级真实挂载路径 sdk/dist——部署前 docker inspect tracker-nginx 复核 Mounts）
• Step 2: v2 上线验证：Playwright 访问 hesucar 首页 → 断言 v1.js 新 build_id 加载 → 等待/触发 flush → 生产 DB/admin 查：事件带 page_id、session_resume 出现于空闲后、page_stay 区间多条、rejected 明细回传
• Step 3: 48h 观察指标（D02 复现指标）：孤立 page_stay 会话（同 session 只有 page_stay 无 page_view）占比 → 应为 0；事件量增幅记录；RawBatch status 分布
• Step 4: 写台账 docs/验收记录-阶段A.md：OT01-OT04/OT06/OT17 用例 ↔ 自动化测试映射 + 生产观察数据 + 构建版本
• Step 5: Commit 台账 + 汇报：部署版本（build_id）、验证结果、遗留项（阶段B 清单）

---

风险提示（执行者必读）

1. 勿动归属之外的既有不变量：ingest.py 头注释五条并发不变量中 ①②④⑤ 全保留，③ 的 F 表达式模式保留——唯 stay_ms 字段按风险项 2 改重算（其余字段 duration_ms/event_count/ended_at 仍 F 增量）。
2. stay_ms 重算是唯一聚合改动：duration_ms/event_count/ended_at 维持现有 F 增量；只有 stay_ms 从增量改重算。v1+v2 混合解释：v1 SUM + v2 并集相加。recompute 时对 page_id 非空但 active_from_ms/active_to_ms 为 null 的脏数据 filter 防御（validate_event v2 分支强制区间完整，双保险）。
3. SDK 红线：任何回调异常静默（spec §4.5）；PageSession/SPA hook 全部 try/catch 包裹。BUILD_ID 经 globalThis.__BUILD_ID__ ?? "unknown" 兜底——define 缺失（vitest/异常构建环境）不得炸。
4. 测试向量是契约：T1 的 fixtures 文件 T9 复用，两侧任何一侧改规则必须同步改向量并两侧重跑。
5. 生产部署顺序：server 先（T11）验证 v1 无回归后才发 SDK（T12）；SDK 回滚只需还原 sdk/dist，服务端不动。
6. 测试 helper 均为新建：_base_event_kwargs/_now_iso/SITE_ID 等在 server tests 中不存在——执行者创建它们而非寻找。Task 1 Step 6 点名更新既有测试 test_pageview_referrer_truncated_after_reencode（CJK q 值不再匹配 _SAFE_VALUE，改用超长 utm 值驱动同一路径）。