涉及文件： src/sage/stream/operators.py、src/sage/stream/datastream.py、src/sage/runtime/flownet/runtime/operator_runtime/stateful.py
问题描述：
KeyByOperator 将 packet 的 partition_key 字段更新后路由出去，按文档设计，下游 stateful operator 应当基于这个 key 实现 per-key 的状态隔离。但检查代码可以发现，partition_key 从未被传递或注入到 StatefulProcessRuntime 的状态读写路径中。当前 stateful.py 的状态操作只按 namespace + key 寻址，没有感知 packet 携带的 partition_key，也没有在 worker 维度做隔离。这意味着：
- 对"user_A"和"user_B"的 packet 调用
process_with_state，得到的是完全相同的状态 namespace，而不是各自独立的 per-key 状态空间。
keyed_state_example.py 注释声称实现了"Per-user feature aggregation"和"Automatic state persistence and recovery"，但实际上状态是保存在 MapFunction 实例变量里的，重启即丢失、跨 worker 无隔离。KeyByTransformation 在 pipeline_compiler.py 中被分类为 SourceActorWrapper，而不是一个独立的路由阶段，这可能导致 key 路由在某些 pipeline 拓扑下不生效。
需要修复的内容：
- 在
FlowEngineV1（或 operator 的执行上下文）中，将 packet 的 partition_key 注入到 StatefulProcessRuntime 的 namespace 构造逻辑中，使得 per-key 状态在存储层真正隔离。
- 明确 keyed state 的语义：
key_by 后的 process_with_state 是否自动获得 per-key 隔离，还是需要用户在 state namespace 中手动嵌入 key？目前这个语义在文档和代码中均未定义。
- 修复
keyed_state_example.py 中描述与实现不符的问题，或补充实现使其真正支持 per-key 隔离。
任务目标： 让 key_by 操作产生真实的 per-key 状态隔离，而不只是一个修改 packet 字段但对 stateful operator 无效的路由暗示。
完成标准： key_by 后的 process_with_state 能获得真实的 per-key 状态隔离，partition_key 被传递到 StatefulProcessRuntime 的 namespace 寻址中，不同 key 的状态互不干扰，keyed_state_example.py 的行为与注释描述一致。
父 Issue: #1484
涉及文件：
src/sage/stream/operators.py、src/sage/stream/datastream.py、src/sage/runtime/flownet/runtime/operator_runtime/stateful.py问题描述：
KeyByOperator将 packet 的partition_key字段更新后路由出去，按文档设计，下游 stateful operator 应当基于这个 key 实现 per-key 的状态隔离。但检查代码可以发现，partition_key从未被传递或注入到StatefulProcessRuntime的状态读写路径中。当前stateful.py的状态操作只按namespace + key寻址，没有感知 packet 携带的partition_key，也没有在 worker 维度做隔离。这意味着：process_with_state，得到的是完全相同的状态 namespace，而不是各自独立的 per-key 状态空间。keyed_state_example.py注释声称实现了"Per-user feature aggregation"和"Automatic state persistence and recovery"，但实际上状态是保存在MapFunction实例变量里的，重启即丢失、跨 worker 无隔离。KeyByTransformation在pipeline_compiler.py中被分类为SourceActorWrapper，而不是一个独立的路由阶段，这可能导致 key 路由在某些 pipeline 拓扑下不生效。需要修复的内容：
FlowEngineV1（或 operator 的执行上下文）中，将 packet 的partition_key注入到StatefulProcessRuntime的 namespace 构造逻辑中，使得 per-key 状态在存储层真正隔离。key_by后的process_with_state是否自动获得 per-key 隔离，还是需要用户在 state namespace 中手动嵌入 key？目前这个语义在文档和代码中均未定义。keyed_state_example.py中描述与实现不符的问题，或补充实现使其真正支持 per-key 隔离。任务目标： 让
key_by操作产生真实的 per-key 状态隔离，而不只是一个修改 packet 字段但对 stateful operator 无效的路由暗示。完成标准：
key_by后的process_with_state能获得真实的 per-key 状态隔离，partition_key被传递到StatefulProcessRuntime的 namespace 寻址中，不同 key 的状态互不干扰，keyed_state_example.py的行为与注释描述一致。父 Issue: #1484