沙箱会话生命周期与记忆检索

## 每跑一次图就多一台常驻 VM，没有任何自动回收路径

_reap_idle 只在 run() 里调，也就是说没人再跑代码时闲置 VM 永远收不掉，
microvm_idle_seconds 形同虚设；而 thread_id 每个 run 都是新的，runner 的
finally 又从不调 cleanup，于是每跑一次带代码节点的图就多一台几百 MB 的常驻
microVM（seatbelt 那边同样积了几十个历史会话目录）。

两处一起补：run 到终态就 cleanup(run_id)（interrupted 除外，人回来还要接着跑），
外加 lifespan 里起一个后台回收任务定期 _reap_idle——回收本来也不该占着请求
路径的时间。实测连跑三次 live_vms 稳定回到 0。

## 超时会连人带文件销毁整台 VM

会话契约是"同 session_id 共享文件系统，可以先写文件下一步再读回来"，而
microVM 超时直接 _destroy，工作区连同用户数据一起没了，返回结果里没有任何
提示。另外三个后端超时只 killpg、工作目录照留——换到 microVM 之后同一张图的
行为就变了。现在超时先在 VM 内 pkill 掉跑飞的进程（入口名带唯一后缀，不会
误伤同会话的并发执行），确认 VM 还活着就保住工作区；收拾不动才销毁。

限额变更那条没法避免重建（VM 内存是启动时定死的），但不能不吭声：同一个 run
里两个代码节点填了不同的 memory_mb 就会触发。ExecResult 新增 session_reset，
代码节点在时间线上明确告警，否则下游只会看到一个无从追查的 FileNotFoundError。

## 新建 scope 写入第一条后永远搜不到

hybrid_rank 的 _norm 在 hi==lo 时把所有分数抹成 0，而候选只有一条时 hi 恒等于
lo，合并分 0.0 会被 min_score 过滤掉。"先写一条、下一轮再 recall"恰恰是 agent
最常见的用法。全都一样分是"都一样好"，不是"都一样差"，改成返回 1.0。

## 记忆去重误判，写入被静默丢弃且返回不相关的旧记忆

remember 拿 recall 的分数判重，但那是 min-max 排名归一化的结果——排第一的那条
在两路信号上都会被归成 1.0，与它和查询像不像毫无关系。于是 scope 里只要已有
两条记忆，新内容随便撞一个字排到第一就被判成重复丢掉，接口还返回 201 并把那条
不相关的旧记忆当成"刚保存的"交回去。改成直接比未归一化的余弦相似度。

验证：三条互不相关的记忆都能写进去、完全相同的仍被去重、单条 scope 可检索。

backend/app/engine/nodes/tools.py

@@ -135,6 +135,12 @@ async def run_code(state: GraphState, ctx: NodeContext) -> dict[str, Any]:
         # 要了硬件隔离却没拿到，必须说出来——否则用户以为自己在 VM 里跑
         ctx.emit(EventType.LOG, level="warn",
                  message=f"节点要求 strict 隔离，但 microVM 未就绪，实际用了 {result.backend}")
+    if result.session_reset:
+        # 会话工作区被清空了。同一会话的下游节点会突然读不到自己写的文件，
+        # 不说一声的话那个 FileNotFoundError 根本无从追查
+        ctx.emit(EventType.LOG, level="warn",
+                 message="沙箱会话已重置，此前在该会话写入的文件已丢失"
+                         "（多为超时重建，或同会话中各节点的内存限额不一致）")
     ctx.emit(
         EventType.SANDBOX_END,
         ok=result.ok,

backend/app/engine/runner.py

@@ -380,6 +380,15 @@ class RunManager:
                 elapsed = int((time.perf_counter() - started) * 1000)
                 await self._finalize(run_id, status, error, final_state, elapsed)
                 self._tasks.pop(run_id, None)
+                # 运行到终态就把沙箱会话收掉。thread_id 每个 run 都是新的，
+                # 不收的话每跑一次带代码节点的图就多一台常驻 microVM（几百 MB），
+                # 而且没有任何自动回收路径会碰它——闲置回收只在"下次有人执行
+                # 代码"时才触发。interrupted 要留着，人回来还得接着跑。
+                if status in ("succeeded", "failed", "cancelled"):
+                    with contextlib.suppress(Exception):
+                        from app.sandbox.manager import sandbox_manager
+
+                        await sandbox_manager.cleanup(run_id)
 
     async def _handle_custom(self, run_id: str, chunk: Any) -> None:
         """节点通过 get_stream_writer 发来的事件。"""

backend/app/main.py

@@ -28,6 +28,9 @@ async def lifespan(_: FastAPI) -> AsyncIterator[None]:
     await init_db()
     await seed_defaults()
     await run_manager.setup()
+    from app.sandbox.manager import sandbox_manager as _sbm
+
+    await _sbm.start_reaper()
     logger.info("AgentLab 就绪 · 数据目录 %s", settings.data_dir)
     try:
         yield

backend/app/memory/embeddings.py

@@ -179,7 +179,11 @@ def hybrid_rank(
     def _norm(xs: list[float]) -> list[float]:
         lo, hi = min(xs), max(xs)
         if hi - lo < 1e-9:
-            return [0.0 for _ in xs]
+            # 全都一样分，说明这一路信号区分不出高下——那是"都一样好"，
+            # 不是"都一样差"。返回 0 会让候选只有一条时它的合并分恒为 0，
+            # 被 min_score 过滤掉，于是新建 scope 写入第一条后永远搜不到，
+            # 而"先写一条、下一轮再 recall"正是 agent 最常见的用法。
+            return [1.0 for _ in xs]
         return [(x - lo) / (hi - lo) for x in xs]
 
     kw_n, vec_n = _norm(kw_scores), _norm(vec_scores)

backend/app/memory/store.py

@@ -7,7 +7,11 @@ from sqlalchemy import delete, func, select
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.db.models import MemoryItem
-from app.memory.embeddings import embed_text, from_blob, hybrid_rank, to_blob
+from app.memory.embeddings import cosine, embed_text, from_blob, hybrid_rank, to_blob
+
+# 判重阈值：余弦相似度高于它才算同一件事。比排名分数严格得多，
+# 因为这里比的是真实语义距离，不是"谁排第一"。
+_DEDUPE_THRESHOLD = 0.92
 
 
 async def remember(
@@ -26,17 +30,33 @@ async def remember(
     反复记下来，不去重的话记忆库几轮就被同义句灌满了。
     """
     content = content.strip()
-    if dedupe:
-        existing = await recall(session, scope=scope, query=content, limit=1, min_score=0.0)
-        if existing and existing[0]["score"] > 0.92:
-            item = await session.get(MemoryItem, existing[0]["id"])
-            if item:
-                item.importance = max(item.importance, importance)
-                item.use_count += 1
-                await session.commit()
-                return item
-
     vec = await embed_text(content)
+
+    if dedupe:
+        # 判重必须看**真实相似度**，不能用 recall 的分数。recall 走 hybrid_rank，
+        # 那里的分数是 min-max 排名归一化的结果：排第一的那条在两路信号上都会
+        # 被归成 1.0，与它和查询像不像毫无关系。于是 scope 里只要已有两条记忆，
+        # 新内容随便撞上一个字排到第一，就会被判成"重复"而丢弃，接口还返回 201
+        # 并把那条不相关的旧记忆当成"刚保存的"交回去——静默丢数据。
+        stmt = select(MemoryItem).where(MemoryItem.scope == scope)
+        if kind:
+            stmt = stmt.where(MemoryItem.kind == kind)
+        candidates = list((await session.execute(stmt)).scalars())
+        best, best_score = None, 0.0
+        for row in candidates:
+            other = from_blob(row.embedding)
+            if other is None:
+                continue
+            score = cosine(vec, other)
+            if score > best_score:
+                best, best_score = row, score
+        if best is not None and best_score > _DEDUPE_THRESHOLD:
+            best.importance = max(best.importance, importance)
+            best.use_count += 1
+            await session.commit()
+            await session.refresh(best)
+            return best
+
     item = MemoryItem(
         scope=scope,
         kind=kind,

backend/app/sandbox/base.py

@@ -43,6 +43,11 @@ class ExecResult(BaseModel):
     backend: str = ""
     files_written: list[str] = Field(default_factory=list)
     error: str | None = None
+    # 这次执行导致会话工作区被重置了（比如 microVM 因超时或限额变更被销毁）。
+    # 会话契约说"同 session_id 共享文件系统，可以先写文件下一步再读回来"，
+    # 所以工作区没了必须说出来——否则下一个节点只会看到 FileNotFoundError，
+    # 完全不知道是谁把文件弄丢的。
+    session_reset: bool = False
 
     def summary(self) -> str:
         if self.timed_out:

backend/app/sandbox/manager.py

@@ -56,6 +56,7 @@ class SandboxManager:
         self._named: dict[str, Sandbox] = {}  # 按名字缓存，供节点级档位使用
         self._lock = asyncio.Lock()
         self._resolved_from: str = ""
+        self._reaper: asyncio.Task[None] | None = None
 
     async def get(self, prefer: str | None = None) -> Sandbox:
         """取后端。prefer 指定档位时给那一档，拿不到就退回默认那台。
@@ -153,6 +154,32 @@ class SandboxManager:
             files=files,
         )
 
+    async def start_reaper(self) -> None:
+        """起一个后台任务定期回收闲置会话。
+
+        以前回收只挂在 run() 里，也就是说"没人再跑代码"时闲置的 microVM
+        永远不会被收掉，microvm_idle_seconds 形同虚设——几百 MB 一台就这么
+        挂着。回收本来也不该占着请求路径的时间。
+        """
+        if self._reaper is not None:
+            return
+
+        async def _loop() -> None:
+            interval = max(30, settings.microvm_idle_seconds // 4)
+            while True:
+                try:
+                    await asyncio.sleep(interval)
+                    for impl in list(self._named.values()) + ([self._backend] if self._backend else []):
+                        reap = getattr(impl, "_reap_idle", None)
+                        if reap is not None:
+                            await reap()
+                except asyncio.CancelledError:
+                    raise
+                except Exception:  # noqa: BLE001 - 回收失败不该让任务停掉
+                    continue
+
+        self._reaper = asyncio.create_task(_loop())
+
     async def health(self) -> dict[str, Any]:
         backend = await self.get()
         info = await backend.health()
@@ -178,6 +205,9 @@ class SandboxManager:
             await impl.cleanup(session_id)
 
     async def close(self) -> None:
+        if self._reaper is not None:
+            self._reaper.cancel()
+            self._reaper = None
         for impl in {id(b): b for b in [*self._named.values(), self._backend] if b}.values():
             await impl.close()
 

backend/app/sandbox/microvm_sandbox.py

@@ -45,6 +45,8 @@ class _Lease:
     memory_mb: int
     cpus: int
     last_used: float = field(default_factory=time.monotonic)
+    # 这台是因为限额变更重建出来的，也就是说上一次的会话文件已经没了
+    was_reset: bool = False
 
 
 class MicroVMSandbox(Sandbox):
@@ -160,6 +162,7 @@ class MicroVMSandbox(Sandbox):
 
         cpus = max(1, int(limits.cpus))
         async with self._lock:
+            reset = False
             lease = self._leases.get(session_id)
             if lease is not None:
                 if (lease.network == limits.network
@@ -167,8 +170,11 @@ class MicroVMSandbox(Sandbox):
                         and lease.cpus == cpus):
                     lease.last_used = time.monotonic()
                     return lease
-                # 限额变了：旧 VM 的边界已经不对，销毁重建
+                # 限额变了：VM 的内存/CPU 是启动时定死的，只能销毁重建。
+                # 但重建等于把会话工作区清空——同一个 run 里两个代码节点各填了
+                # 不同的 memory_mb 就会触发，而调用方对此毫无察觉。标记出来。
                 await self._destroy(session_id)
+                reset = True
 
             # network 必须是 Network 实例（传 bool 会被 SDK 拒掉）
             network = ms.Network.allow_all() if limits.network else ms.Network.none()
@@ -189,6 +195,7 @@ class MicroVMSandbox(Sandbox):
 
             lease = _Lease(vm=vm, network=limits.network,
                            memory_mb=limits.memory_mb, cpus=cpus)
+            lease.was_reset = reset
             self._leases[session_id] = lease
             return lease
 
@@ -201,6 +208,29 @@ class MicroVMSandbox(Sandbox):
         except Exception:  # noqa: BLE001 - 已经没了就算
             pass
 
+    async def _kill_leftovers(self, session_id: str, entry: str) -> bool:
+        """超时后在 VM 内把跑飞的进程杀掉，保住会话工作区。
+
+        返回 True 表示 VM 还能用；False 表示没收拾干净，调用方该销毁重建。
+        """
+        lease = self._leases.get(session_id)
+        if lease is None:
+            return False
+        try:
+            # 只杀这次的入口脚本引出来的进程——入口名带唯一后缀，
+            # 不会误伤同会话里别的并发执行
+            await asyncio.wait_for(
+                lease.vm.shell(f"pkill -9 -f {entry} || true", timeout=10),
+                timeout=15,
+            )
+            # 确认 VM 还答应
+            out = await asyncio.wait_for(
+                lease.vm.exec("sh", ["-c", "echo alive"], timeout=10), timeout=15
+            )
+            return "alive" in str(getattr(out, "stdout_text", "") or "")
+        except Exception:  # noqa: BLE001 - 收拾不动就让调用方销毁
+            return False
+
     async def _reap_idle(self) -> None:
         """回收闲置 VM：它们各自占着几百 MB 内存，不能一直挂着。"""
         cutoff = time.monotonic() - settings.microvm_idle_seconds
@@ -281,12 +311,20 @@ class MicroVMSandbox(Sandbox):
                 timeout=limits.timeout,
             )
         except ms.ExecTimeoutError:
-            # 超时的 VM 里可能还有进程在烧 CPU，状态不可信，直接销毁
-            await self._destroy(sid)
+            # 超时的 VM 里多半还有进程在烧 CPU，得收拾掉。但**先别销毁整台 VM**：
+            # 会话契约说同 session 共享文件系统（上一步写的文件下一步要读回来），
+            # 销毁 VM 等于把工作区连同用户数据一起清空，而另外三个后端超时只
+            # killpg、工作目录照留。先试着在 VM 里把残留进程杀干净，
+            # 实在不行才销毁——并且如实告诉调用方会话被重置了。
+            killed = await self._kill_leftovers(sid, entry)
+            if not killed:
+                await self._destroy(sid)
             return ExecResult(
                 ok=False, backend=self.name, timed_out=True, exit_code=124,
                 duration_ms=int((time.perf_counter() - started) * 1000),
-                error=f"执行超过 {limits.timeout}s 被终止",
+                session_reset=not killed,
+                error=f"执行超过 {limits.timeout}s 被终止"
+                      + ("" if killed else "；VM 已重置，该会话之前写的文件已丢失"),
             )
         except Exception as e:  # noqa: BLE001
             return ExecResult(
@@ -337,6 +375,7 @@ class MicroVMSandbox(Sandbox):
             backend=self.name,
             duration_ms=int((time.perf_counter() - started) * 1000),
             files_written=written,
+            session_reset=lease.was_reset,
             error=error,
         )