fix(p2): TEVARN_HOME unify, checkpoint ownership, static version guard

P2 from post-a23c95b audit + suggested improvements:

1. TEVARN_HOME: capability/wave_a/agent_ops tools, skill storage, usage
   ledger, run_events/inbox/snapshots, audit, encryption, process_guard,
   smoke_test, cli, win_boot, intent_telemetry, permissions
2. Checkpoint registry stores session_id/user_id/workspace_root; restore
   API enforces ownership match when bound
3. static_frontend warns if served tree looks like 0.5.4-alpha / version
   mismatch vs product 0.4.2 (no silent stale UI)
4. frontend build:export + write:static-version → out/version.json stamp
5. check_gates: no || true; fail if tests import deleted _next_seq

doSync after_seq already landed in a23c95b. Panoramic audit clean.

Author: WagiklinKasa

backend/agent/control_inbox.py

@@ -68,7 +68,7 @@ def _inbox_dir() -> Path:
         return base
     except Exception:
         try:
-            base = Path.home() / ".tevarn" / "control_inbox"
+            base = get_tevarn_home() / "control_inbox"
             base.mkdir(parents=True, exist_ok=True)
             return base
         except Exception:

backend/agent/file_checkpoint.py

@@ -154,19 +154,33 @@ def register_checkpoint(
     snapshot_path: str,
     target_path: str,
     tool: str = "",
+    session_id: str | None = None,
+    user_id: str | None = None,
+    workspace_id: str | None = None,
 ) -> str:
-    """Register a snapshot and return opaque id (thread + cross-process locked)."""
+    """Register a snapshot and return opaque id (thread + cross-process locked).
+
+    Optional ownership fields bind restore to the creating session/user.
+    """
     root = _project_root()
     cid = uuid.uuid4().hex
 
     def _do() -> None:
         reg = _load_registry(root)
-        reg[cid] = {
+        entry: dict[str, Any] = {
             "snapshot": str(snapshot_path),
             "target": str(target_path),
             "tool": tool,
             "created_at": datetime.now(timezone.utc).isoformat(),
+            "workspace_root": str(root),
         }
+        if session_id:
+            entry["session_id"] = str(session_id)
+        if user_id:
+            entry["user_id"] = str(user_id)
+        if workspace_id:
+            entry["workspace_id"] = str(workspace_id)
+        reg[cid] = entry
         if len(reg) > 400:
             items = sorted(
                 reg.items(),
@@ -248,10 +262,22 @@ def snapshot_path_for_tool(name: str, arguments: dict[str, Any]) -> str | None:
     idx = dest_root / "INDEX.txt"
     with idx.open("a", encoding="utf-8") as f:
         f.write(f"{name}\t{target}\t{dest}\n")
+    _sid = str(
+        (arguments or {}).get("_session_id")
+        or (arguments or {}).get("session_id")
+        or ""
+    ).strip() or None
+    _uid = str(
+        (arguments or {}).get("_user_id")
+        or (arguments or {}).get("user_id")
+        or ""
+    ).strip() or None
     cid = register_checkpoint(
         snapshot_path=str(dest),
         target_path=str(target),
         tool=name,
+        session_id=_sid,
+        user_id=_uid,
     )
     # also write id into index for operators
     with idx.open("a", encoding="utf-8") as f:
@@ -267,12 +293,18 @@ def list_recent_checkpoints(limit: int = 20) -> list[str]:
     return [str(p) for p in dirs[:limit]]
 
 
-def restore_checkpoint_file(snapshot_path_or_id: str) -> dict[str, Any]:
+def restore_checkpoint_file(
+    snapshot_path_or_id: str,
+    *,
+    session_id: str | None = None,
+    user_id: str | None = None,
+) -> dict[str, Any]:
     """Restore by opaque id (preferred) or legacy snapshot path.
 
     Security:
       - registry targets must stay under project root
       - legacy path snapshots must live under .tevarn/checkpoints/
+      - if registry has session_id/user_id, caller must match when provided
     """
     raw = str(snapshot_path_or_id or "").strip()
     if not raw:
@@ -284,6 +316,13 @@ def restore_checkpoint_file(snapshot_path_or_id: str) -> dict[str, Any]:
     # Opaque id path
     entry = lookup_checkpoint(raw)
     if entry and entry.get("backend") == "python" and entry.get("snapshot"):
+        # Ownership gate (shared-deploy safety)
+        reg_sid = str(entry.get("session_id") or "").strip()
+        reg_uid = str(entry.get("user_id") or "").strip()
+        if reg_sid and session_id and str(session_id).strip() != reg_sid:
+            return {"ok": False, "error": "checkpoint session mismatch"}
+        if reg_uid and user_id and str(user_id).strip() != reg_uid:
+            return {"ok": False, "error": "checkpoint owner mismatch"}
         snap = Path(str(entry["snapshot"])).expanduser()
         target = Path(str(entry.get("target") or "")).expanduser()
         try:

backend/agent/run_events.py

@@ -33,7 +33,7 @@ def _spill_dir() -> Path:
         return p
     except Exception:
         try:
-            p = Path.home() / ".tevarn" / "run_events"
+            p = get_tevarn_home() / "run_events"
             p.mkdir(parents=True, exist_ok=True)
             return p
         except Exception:

backend/agent/run_snapshot_store.py

@@ -14,6 +14,8 @@
 """
 from __future__ import annotations
 
+from backend.core.config import get_tevarn_home
+
 import json
 import logging
 import os
@@ -55,7 +57,7 @@ def _dir() -> Path:
 
         root = Path(os.environ.get("TEVARN_HOME") or (host_home() / ".tevarn"))
     except Exception:
-        root = Path(os.environ.get("TEVARN_HOME") or Path.home() / ".tevarn")
+        root = Path(os.environ.get("TEVARN_HOME") or get_tevarn_home())
     d = root / "run_snapshots"
     d.mkdir(parents=True, exist_ok=True)
     return d

backend/api/routes/files.py

@@ -719,7 +719,11 @@ def _try_python(key: str) -> dict | None:
             else nullcontext()
         )
         with ctx:
-            result = restore_checkpoint_file(key)
+            result = restore_checkpoint_file(
+                key,
+                session_id=session_id_raw or None,
+                user_id=str(getattr(current_user, "id", "") or "") or None,
+            )
         if result.get("ok"):
             result["backend"] = "python"
             if root_for_restore:

backend/api/routes/smoke_test.py

@@ -1,3 +1,4 @@
+from backend.core.config import get_tevarn_home
 """
 冒烟测试专用路由
 提供同步 HTTP 接口，方便后端直接调用模拟用户对话，
@@ -122,7 +123,7 @@ async def smoke_chat(
 
     # 读取最近日志提取上下文压缩信息
     try:
-        log_dir = os.environ.get("TEVARN_LOG_DIR", "") or str(Path.home() / ".tevarn" / "logs")
+        log_dir = os.environ.get("TEVARN_LOG_DIR", "") or str(get_tevarn_home() / "logs")
         log_path = str(Path(log_dir) / "tevarn.log")
         with open(log_path, "r", encoding="utf-8", errors="ignore") as f:
             lines = f.readlines()[-200:]  # 只看最近 200 行

backend/cli.py

@@ -15,6 +15,8 @@
 
 from __future__ import annotations
 
+from backend.core.config import get_tevarn_home
+
 import argparse
 import json
 import os
@@ -42,7 +44,7 @@ def _base_url(args: argparse.Namespace) -> str:
 
 
 def _token_path() -> Path:
-    home = Path.home() / ".tevarn"
+    home = get_tevarn_home()
     home.mkdir(parents=True, exist_ok=True)
     return home / "cli_token"
 

backend/core/encryption.py

@@ -30,7 +30,7 @@ def _secrets_file_path() -> Path:
         from backend.core.config import get_tevarn_home
         return get_tevarn_home() / "secrets.json"
     except Exception:
-        return Path.home() / ".tevarn" / "secrets.json"
+        return get_tevarn_home() / "secrets.json"
 
 
 def _derive_key_from_jwt_secret() -> bytes:

backend/core/process_guard.py

@@ -12,6 +12,8 @@
 """
 from __future__ import annotations
 
+from backend.core.config import get_tevarn_home
+
 import atexit
 import json
 import logging
@@ -32,7 +34,7 @@
 
 def breadcrumb_path() -> Path:
     """Prefer ~/.tevarn/logs (user-writable); fall back next to package."""
-    home = os.environ.get("TEVARN_HOME") or str(Path.home() / ".tevarn")
+    home = os.environ.get("TEVARN_HOME") or str(get_tevarn_home())
     p = Path(home) / "logs" / "process_breadcrumb.jsonl"
     try:
         p.parent.mkdir(parents=True, exist_ok=True)

backend/kernel/audit_store.py

@@ -20,7 +20,7 @@ def _default_audit_path() -> str:
         from backend.core.config import get_tevarn_home
         return str(get_tevarn_home() / "kernel_events.jsonl")
     except Exception:
-        return os.path.expanduser("~/.tevarn/kernel_events.jsonl")
+        return str(get_tevarn_home() / "kernel_events.jsonl")
 
 
 _DEFAULT_PATH = _default_audit_path()