Star9001
327
代码
代码
Issues63
Pull Requests21
项目讨论
Wiki
分析
Star9001
327
已合并
feat(session-manager): session.create 临时 key 原子改绑(LUA_REBIND + /api/session/rebind) #548
王明琦创建于 9月20日
feat(session-manager): session.create 临时 key 原子改绑(LUA_REBIND + /api/session/rebind) #548
已合并
王明琦创建于 9月20日
从wangmingqi/agent-runtimedevelop合入到openJiuwen/agent-runtimedevelop
共 12 个文件变更+569-7
Mapplications/agent_runtime/docs/design/session-manager-design.md+48-1文件内容审核中，请稍后刷新重试
Aapplications/agent_runtime/docs/feature/2026-09-session-create-rebind.md+86-0文件内容审核中，请稍后刷新重试
Mapplications/agent_runtime/docs/feature/README.md+1-0文件内容审核中，请稍后刷新重试
Mapplications/agent_runtime/docs/spec/session-manager.md+6-4文件内容审核中，请稍后刷新重试
Mapplications/agent_runtime/scripts/verify_redis_cluster.py+16-0
@@ -150,6 +150,22 @@ async def verify(url: str, wipe: bool) -> int:
150 check("SM: evict 四处同删", evicted is not None150 check("SM: evict 四处同删", evicted is not None
151 and evicted.get("scope_id") == "v-scope")151 and evicted.get("scope_id") == "v-scope")
152 await sm.evict("v-s2")152 await sm.evict("v-s2")
153+ # [3c] LUA_REBIND(2026-09-session-create-rebind)：临时 key 槽位原子搬给
154+ # 真实 id——四处多键同脚本访问，cluster 下依赖 hash tag 同槽 + eval 路由锚
155+ await client.hset(sm.k.pod_info("v-scope", "v-pod-1"), mapping={
156+ "sse_url": "http://127.0.0.1:8086", "deploy_ver": "v1"})
157+ await sm.route_place(
158+ session_id="v-temp", scope_id="v-scope", expiry_ts=now + 60,
159+ session_ttl=60, scope_concurrency=10, pod_concurrency=10,
160+ max_pods=2, now=now,
161+ )
162+ rb = await sm.rebind("v-temp", "v-real", now=now + 1)
163+ members = await client.smembers(sm.k.scope_sessions("v-scope"))
164+ check("SM: rebind 临时 key 原子搬移", rb.get("action") == "rebound"
165+ and members == {b"v-real"}, f"{rb.get('action')}/{sorted(members)}")
166+ touched_real, _ = await sm.touch("v-real", now + 2)
167+ check("SM: rebind 后真实 id touch 保活", touched_real is True)
168+ await sm.evict("v-real")
153 169 
154 # [4] RM 状态层170 # [4] RM 状态层
155 rm = ResourceState(client)171 rm = ResourceState(client)
Mapplications/agent_runtime/src/agent_runtime/session_manager/handlers.py+29-2文件内容审核中，请稍后刷新重试
Mapplications/agent_runtime/src/agent_runtime/session_manager/lua_scripts.py+96-0
@@ -11,10 +11,22 @@
11脚本清单（语义见 SM 设计 §5.1，逐条对齐）：11脚本清单（语义见 SM 设计 §5.1，逐条对齐）：
12- LUA_ROUTE_PLACE route 原子核心：亲和续期 / rebind 重解 / 闸门 / first-fit / 提交12- LUA_ROUTE_PLACE route 原子核心：亲和续期 / rebind 重解 / 闸门 / first-fit / 提交
13- LUA_EVICT session 移除唯一原语（四处同删）13- LUA_EVICT session 移除唯一原语（四处同删）
14+- LUA_REBIND 临时路由 key 原子改绑（session.create 真实 id 回填，见下）
14- LUA_TOUCH 保活续期（惰性 evict 兜底；ttl 就地读 session HASH）15- LUA_TOUCH 保活续期（惰性 evict 兜底；ttl 就地读 session HASH）
15- LUA_SWEEP_IDLE_NOTIFY 空 Pod pass 原子核心：SCARD==0 判定 + NX 去重 + ZREM 退出候选16- LUA_SWEEP_IDLE_NOTIFY 空 Pod pass 原子核心：SCARD==0 判定 + NX 去重 + ZREM 退出候选
16- LUA_REGISTER_POD acquire 成功后登记新 Pod（三处注册同写 + 接入序）17- LUA_REGISTER_POD acquire 成功后登记新 Pod（三处注册同写 + 接入序）
17- LUA_CLEANUP_POD notify_pod_dead 清该 (scope,pod) 的全部注册18- LUA_CLEANUP_POD notify_pod_dead 清该 (scope,pod) 的全部注册
19+ 
20+LUA_REBIND 背景：gateway 的 session.create 以临时 key（sess_*/webhttp_*/身份组合串）
21+route 占槽，AgentServer 返回真实 session id 后，gateway 在把 create 响应交还调用方
22+之前调用本脚本把槽位原子搬给真实 id——修复双计数（临时 key 残留至 session_ttl
23+过期）与亲和断裂（真实 id 全新 placement 可能落在别的 Pod，而会话元数据已写在
24+create 实际服务的 Pod 上）。返回 action：
25+- noop from 不存在（已过期被 sweeper 收走 / 已搬过）——幂等
26+- rubble 残骸自卫（同 LUA_EVICT）
27+- evicted to 为空（create 失败驱逐）或绑定 Pod 注册已消失——仅清 from
28+- overtaken to 已有自己的绑定（首条 chat.send 抢先放置）——仅清 from，保留 to
29+- rebound 搬移成功：四处不变量整体从 from 迁至 to，TTL 刷新
18"""30"""
19 31 
20from __future__ import annotations32from __future__ import annotations
@@ -137,6 +149,90 @@ local remaining = redis.call('SCARD', pfx .. 'pod:' .. scope .. ':' .. pod .. ':
137return {'evicted', scope, pod, tostring(remaining)}149return {'evicted', scope, pod, tostring(remaining)}
138"""150"""
139 151 
152+# Argv: prefix, from_session_id, to_session_id（空串=驱逐）, now, default_session_ttl
153+LUA_REBIND = r"""
154+local pfx = ARGV[1]
155+local from = ARGV[2]
156+local to = ARGV[3]
157+local now = tonumber(ARGV[4])
158+local def_ttl = tonumber(ARGV[5])
159+ 
160+local fkey = pfx .. 'session:' .. from
161+ 
162+-- 读 from 绑定
163+local flat = redis.call('HGETALL', fkey)
164+if #flat == 0 then
165+ return {'noop', '', '', '0'} -- 已被 sweeper 收走 / 已搬过，幂等
166+end
167+local m = {}
168+for i = 1, #flat, 2 do m[flat[i]] = flat[i + 1] end
169+ 
170+-- 残骸自卫（同 LUA_EVICT）：缺 scope_id/pod_id 的半成品哈希 → 自清两处，
171+-- 不上抛（上抛会让 create 响应路径对同一坏键永久失败）。
172+if m['scope_id'] == nil or m['pod_id'] == nil then
173+ redis.call('ZREM', pfx .. 'session_expiry', from)
174+ redis.call('DEL', fkey)
175+ return {'rubble', '', '', '0'}
176+end
177+ 
178+local scope, pod = m['scope_id'], m['pod_id']
179+ 
180+-- to 为空 = 驱逐（create 失败，立即释放槽位，不等 TTL）
181+-- to == from = 无可搬，幂等
182+if to == '' or to == from then
183+ if to == '' then
184+ redis.call('SREM', pfx .. 'scope:' .. scope .. ':sessions', from)
185+ redis.call('SREM', pfx .. 'pod:' .. scope .. ':' .. pod .. ':sessions', from)
186+ redis.call('ZREM', pfx .. 'session_expiry', from)
187+ redis.call('DEL', fkey)
188+ local remaining = redis.call('SCARD', pfx .. 'pod:' .. scope .. ':' .. pod .. ':sessions')
189+ return {'evicted', scope, pod, tostring(remaining)}
190+ end
191+ return {'noop', scope, pod, '0'}
192+end
193+ 
194+local tkey = pfx .. 'session:' .. to
195+ 
196+-- to 已有自己的绑定（首条 chat.send 抢先 route 过）：保留 to，仅清 from。
197+-- 绑定可能指向别的 Pod——那是已生效的亲和，不得覆盖。
198+if redis.call('EXISTS', tkey) == 1 then
199+ redis.call('SREM', pfx .. 'scope:' .. scope .. ':sessions', from)
200+ redis.call('SREM', pfx .. 'pod:' .. scope .. ':' .. pod .. ':sessions', from)
201+ redis.call('ZREM', pfx .. 'session_expiry', from)
202+ redis.call('DEL', fkey)
203+ local remaining = redis.call('SCARD', pfx .. 'pod:' .. scope .. ':' .. pod .. ':sessions')
204+ return {'overtaken', scope, pod, tostring(remaining)}
205+end
206+ 
207+-- 绑定 Pod 注册已消失（notify_pod_dead 清理窗口）：亲和信息失效，仅清 from，
208+-- 真实 id 由后续 chat.send 的 route 全新放置。
209+if redis.call('EXISTS', pfx .. 'pod:' .. scope .. ':' .. pod .. ':info') == 0 then
210+ redis.call('SREM', pfx .. 'scope:' .. scope .. ':sessions', from)
211+ redis.call('SREM', pfx .. 'pod:' .. scope .. ':' .. pod .. ':sessions', from)
212+ redis.call('ZREM', pfx .. 'session_expiry', from)
213+ redis.call('DEL', fkey)
214+ return {'evicted', scope, pod, '0'}
215+end
216+ 
217+-- 搬移：四处不变量整体 from → to，TTL 刷新（含 from 已过期未扫的复活场景——
218+-- 慢 create（冷启动）下亲和仍应成立；scope/pod 槽位计数不变（一删一加））。
219+local ttl = tonumber(m['session_ttl']) or def_ttl
220+local expiry = now + ttl
221+redis.call('SADD', pfx .. 'scope:' .. scope .. ':sessions', to)
222+redis.call('SREM', pfx .. 'scope:' .. scope .. ':sessions', from)
223+redis.call('SADD', pfx .. 'pod:' .. scope .. ':' .. pod .. ':sessions', to)
224+redis.call('SREM', pfx .. 'pod:' .. scope .. ':' .. pod .. ':sessions', from)
225+redis.call('HSET', tkey, 'scope_id', scope, 'pod_id', pod,
226+ 'expiry', tostring(expiry), 'session_ttl', tostring(ttl))
227+redis.call('DEL', fkey)
228+redis.call('ZREM', pfx .. 'session_expiry', from)
229+redis.call('ZADD', pfx .. 'session_expiry', expiry, to)
230+-- 复用 Pod 时清 idle_notified（同 ROUTE_PLACE 提交步；空标记失效）
231+redis.call('DEL', pfx .. 'pod:' .. scope .. ':' .. pod .. ':idle_notified')
232+local remaining = redis.call('SCARD', pfx .. 'pod:' .. scope .. ':' .. pod .. ':sessions')
233+return {'rebound', scope, pod, tostring(remaining)}
234+"""
235+ 
140# Argv: prefix, session_id, now, default_session_ttl236# Argv: prefix, session_id, now, default_session_ttl
141LUA_TOUCH = r"""237LUA_TOUCH = r"""
142local pfx = ARGV[1]238local pfx = ARGV[1]
Mapplications/agent_runtime/src/agent_runtime/session_manager/orchestrator.py+39-0文件内容审核中，请稍后刷新重试
Mapplications/agent_runtime/src/agent_runtime/session_manager/state.py+28-0
@@ -190,6 +190,34 @@ class SessionState:
190 return None190 return None
191 return {"scope_id": ret[1], "pod_id": ret[2], "remaining": ret[3]}191 return {"scope_id": ret[1], "pod_id": ret[2], "remaining": ret[3]}
192 192 
193+ async def rebind(
194+ self, from_session_id: str, to_session_id: str, now: int, default_ttl: int = 60,
195+ ) -> dict[str, str]:
196+ """临时路由 key 原子改绑（session.create 真实 id 回填，幂等）。
197+ 
198+ ``to_session_id`` 为空串 = 驱逐（create 失败立即释放槽位）。返回
199+ ``{'action','scope_id','pod_id','remaining'}``，action 见 lua_scripts
200+ ``LUA_REBIND`` 注释（noop/rubble/evicted/overtaken/rebound）。
201+ """
202+ ret = await self.eval(
203+ lua.LUA_REBIND, from_session_id, to_session_id, now, default_ttl,
204+ )
205+ if not ret:
206+ # eval 层空返回兜底（真异常已 WARNING 留痕）——对齐 route_place 的
207+ # fail-closed 姿态：视作 from 不存在，调用方无副作用可做
208+ return {"action": "noop", "scope_id": "", "pod_id": "", "remaining": "0"}
209+ if ret[0] == "rubble":
210+ logger.warning(
211+ "rebind rubble session (hash missing scope/pod): from=%s",
212+ from_session_id,
213+ )
214+ return {
215+ "action": ret[0],
216+ "scope_id": ret[1] if len(ret) > 1 else "",
217+ "pod_id": ret[2] if len(ret) > 2 else "",
218+ "remaining": ret[3] if len(ret) > 3 else "0",
219+ }
220+ 
193 async def touch(self, session_id: str, now: int, default_ttl: int = 60) -> tuple[bool, str]:221 async def touch(self, session_id: str, now: int, default_ttl: int = 60) -> tuple[bool, str]:
194 """保活。返回 (touched, pod_id)。"""222 """保活。返回 (touched, pod_id)。"""
195 ret = await self.eval(lua.LUA_TOUCH, session_id, now, default_ttl)223 ret = await self.eval(lua.LUA_TOUCH, session_id, now, default_ttl)
Mapplications/agent_runtime/tests/integration/test_http_smoke.py+22-0文件内容审核中，请稍后刷新重试
Mapplications/agent_runtime/tests/integration/test_route_flow.py+72-0文件内容审核中，请稍后刷新重试
Mapplications/agent_runtime/tests/session_manager/test_sm_state.py+126-0
@@ -365,3 +365,129 @@ async def test_route_place_rubble_hash_self_heals_not_crashes(sm_state):
365 info = await sm_state.redis.hgetall(sm_state.k.session("s-rubble-rp"))365 info = await sm_state.redis.hgetall(sm_state.k.session("s-rubble-rp"))
366 assert info[b"scope_id"] == SCOPE.encode() and info[b"pod_id"] == b"pod_rp"366 assert info[b"scope_id"] == SCOPE.encode() and info[b"pod_id"] == b"pod_rp"
367 assert b"expiry" in info and b"session_ttl" in info367 assert b"expiry" in info and b"session_ttl" in info
368+ 
369+ 
370+# ---------------------------------------------------------------- LUA_REBIND：临时 key 原子改绑
371+ 
372+ 
373+@requires_lua
374+async def test_rebind_moves_four_places_and_refreshes_ttl(sm_state, placed):
375+ """rebound 主路径：四处不变量整体 from→to，TTL 按新时刻刷新。"""
376+ await placed() # sess_1 已放置 pod_1（expiry=NOW+60, ttl=60）
377+ result = await sm_state.rebind("sess_1", "web_real_1", now=NOW + 5)
378+ assert result["action"] == "rebound"
379+ assert result["scope_id"] == SCOPE and result["pod_id"] == "pod_1"
380+ # 集合成员换名（计数不变）
381+ assert await sm_state.redis.scard(sm_state.k.scope_sessions(SCOPE)) == 1
382+ assert await sm_state.redis.smembers(sm_state.k.scope_sessions(SCOPE)) == {b"web_real_1"}
383+ assert await sm_state.redis.smembers(sm_state.k.pod_sessions(SCOPE, "pod_1")) == {b"web_real_1"}
384+ # 哈希搬移 + expiry 刷新（NOW+5 + 60，读 from 哈希的 session_ttl）
385+ binding = await sm_state.redis.hgetall(sm_state.k.session("web_real_1"))
386+ assert binding[b"scope_id"] == SCOPE.encode()
387+ assert binding[b"pod_id"] == b"pod_1"
388+ assert int(binding[b"expiry"]) == NOW + 5 + 60
389+ assert binding[b"session_ttl"] == b"60"
390+ assert await sm_state.redis.exists(sm_state.k.session("sess_1")) == 0
391+ # ZSET 成员换名
392+ assert await sm_state.redis.zscore(sm_state.k.session_expiry(), "sess_1") is None
393+ assert await sm_state.redis.zscore(sm_state.k.session_expiry(), "web_real_1") == NOW + 65
394+ 
395+ 
396+@requires_lua
397+async def test_rebind_noop_when_from_missing(sm_state):
398+ """from 不存在（已被 sweeper 收走/已搬过）→ 幂等 noop。"""
399+ result = await sm_state.rebind("ghost", "web_real_1", now=NOW)
400+ assert result["action"] == "noop"
401+ assert await sm_state.redis.exists(sm_state.k.session("web_real_1")) == 0
402+ 
403+ 
404+@requires_lua
405+async def test_rebind_same_id_is_noop(sm_state, placed):
406+ await placed()
407+ result = await sm_state.rebind("sess_1", "sess_1", now=NOW + 5)
408+ assert result["action"] == "noop"
409+ assert await sm_state.redis.scard(sm_state.k.scope_sessions(SCOPE)) == 1
410+ 
411+ 
412+@requires_lua
413+async def test_rebind_empty_to_evicts(sm_state, placed):
414+ """to 为空 = 驱逐（create 失败立即释放槽位，不等 session_ttl）。"""
415+ await placed()
416+ result = await sm_state.rebind("sess_1", "", now=NOW + 5)
417+ assert result["action"] == "evicted"
418+ assert result["scope_id"] == SCOPE and result["pod_id"] == "pod_1"
419+ assert await sm_state.redis.scard(sm_state.k.scope_sessions(SCOPE)) == 0
420+ assert await sm_state.redis.scard(sm_state.k.pod_sessions(SCOPE, "pod_1")) == 0
421+ assert await sm_state.redis.exists(sm_state.k.session("sess_1")) == 0
422+ assert await sm_state.redis.zscore(sm_state.k.session_expiry(), "sess_1") is None
423+ 
424+ 
425+@requires_lua
426+async def test_rebind_overtaken_keeps_existing_to_binding(sm_state, placed):
427+ """to 已有自己的绑定（首条 chat.send 抢先 route）→ 仅清 from，保留 to。"""
428+ await placed() # sess_1 → pod_1（1/2）
429+ action, pod = await sm_state.route_place(
430+ "web_real_1", SCOPE, NOW + 60, 60, 3, 2, 2, NOW
431+ )
432+ assert action == "placed" # first-fit 有位，web_real_1 也落 pod_1（2/2）
433+ result = await sm_state.rebind("sess_1", "web_real_1", now=NOW + 5)
434+ assert result["action"] == "overtaken"
435+ # from 四处清除
436+ assert await sm_state.redis.exists(sm_state.k.session("sess_1")) == 0
437+ assert await sm_state.redis.zscore(sm_state.k.session_expiry(), "sess_1") is None
438+ # to 绑定原样保留（expiry 不被改绑刷新）
439+ binding = await sm_state.redis.hgetall(sm_state.k.session("web_real_1"))
440+ assert int(binding[b"expiry"]) == NOW + 60
441+ # 计数不增：scope 仍 2 个成员（web_real_1 + 无），实为 1（sess_1 被清）
442+ assert await sm_state.redis.scard(sm_state.k.scope_sessions(SCOPE)) == 1
443+ 
444+ 
445+@requires_lua
446+async def test_rebind_dead_pod_falls_back_to_evict(sm_state, placed):
447+ """绑定 Pod 注册已消失（notify_pod_dead 窗口）→ 仅清 from，不造死绑定。"""
448+ await placed()
449+ await sm_state.redis.delete(sm_state.k.pod_info(SCOPE, "pod_1"))
450+ result = await sm_state.rebind("sess_1", "web_real_1", now=NOW + 5)
451+ assert result["action"] == "evicted"
452+ assert await sm_state.redis.exists(sm_state.k.session("web_real_1")) == 0
453+ assert await sm_state.redis.scard(sm_state.k.scope_sessions(SCOPE)) == 0
454+ 
455+ 
456+@requires_lua
457+async def test_rebind_rubble_hash_self_heals(sm_state, caplog):
458+ """残骸自卫（同 EVICT/TOUCH）：半成品哈希自清两处，不上抛。"""
459+ import logging as _logging
460+ await sm_state.redis.hset(sm_state.k.session("s-rubble-rb"), "scope_id", SCOPE)
461+ await sm_state.redis.zadd(sm_state.k.session_expiry(), {"s-rubble-rb": 1})
462+ with caplog.at_level(_logging.WARNING, logger="agent_runtime.session_manager"):
463+ result = await sm_state.rebind("s-rubble-rb", "web_real_1", now=NOW)
464+ assert result["action"] == "rubble"
465+ assert await sm_state.redis.exists(sm_state.k.session("s-rubble-rb")) == 0
466+ assert await sm_state.redis.zscore(sm_state.k.session_expiry(), "s-rubble-rb") is None
467+ assert any("rubble" in r.getMessage() for r in caplog.records)
468+ 
469+ 
470+@requires_lua
471+async def test_rebind_resurrects_expired_unswept_binding(sm_state, placed):
472+ """from 已过期未被 sweeper 收走 → 仍搬移（慢 create 下亲和成立），
473+ to 获得全新 TTL；槽位计数不变。"""
474+ await placed()
475+ result = await sm_state.rebind("sess_1", "web_real_1", now=NOW + 3600)
476+ assert result["action"] == "rebound"
477+ binding = await sm_state.redis.hgetall(sm_state.k.session("web_real_1"))
478+ assert int(binding[b"expiry"]) == NOW + 3600 + 60
479+ assert await sm_state.redis.scard(sm_state.k.scope_sessions(SCOPE)) == 1
480+ 
481+ 
482+@requires_lua
483+async def test_rebind_reuses_pod_ttl_from_binding(sm_state):
484+ """session_ttl 就地读 from 哈希（非默认值）：TTL 用绑定自身的 120。"""
485+ await sm_state.register_pod(SCOPE, "pod_t", "http://10.0.0.3:8080/sse", "ver1")
486+ await sm_state.route_place(
487+ "sess_t", SCOPE, NOW + 120, 120, 3, 2, 2, NOW
488+ )
489+ result = await sm_state.rebind("sess_t", "web_real_t", now=NOW + 5, default_ttl=60)
490+ assert result["action"] == "rebound"
491+ binding = await sm_state.redis.hgetall(sm_state.k.session("web_real_t"))
492+ assert int(binding[b"expiry"]) == NOW + 5 + 120
493+ assert binding[b"session_ttl"] == b"120"