SQLite 运行时状态和 agent graph 存储

async fn read_thread_state_audit_rows(path: &Path) -> Result<Vec<ThreadStateAuditRow>>

作用：从指定的状态数据库文件里读取线程的简要信息，并且保证只读取、不创建、不修改。有人做诊断、排查问题或生成审计报告时会用它。

数据流：输入是一条数据库文件路径。函数先按这条路径准备 SQLite 连接选项：文件不存在就报错，不自动创建；用只读方式打开；关闭 SQL 日志。然后它建立一个最多只有 1 个连接的连接池，执行查询，从 threads 表取出 id、rollout_path、archived、source、model_provider 这些列。查完后关闭连接池。最后它把每一行数据库结果变成 ThreadStateAuditRow：把 rollout_path 从文字变成路径，把 archived 从数字 0/非0 变成真假值。输出是一组线程审计记录；如果打开数据库或读取字段失败，就返回错误。

调用关系：这是这个文件对外提供的主要入口。调用者在需要安全检查状态数据库时调用它；它自己把底层工作交给 sqlx 这个数据库库来完成，比如创建连接选项、建立连接池和执行 SQL 查询。它不调用项目里的修复或迁移流程，因此不会改变数据库现场。

调用图：外部调用 3 个（new, new, query）。

fn new(
        id: ThreadId,
        rollout_path: PathBuf,
        created_at: DateTime<Utc>,
        source: SessionSource,
    ) -> Self

作用：创建一个线程资料卡的“半成品”。调用者只需要先给出最关键的几项，比如线程 ID、记录文件路径、创建时间和来源，其它字段先用安全的默认值占位。

数据流：进去的是线程 ID、rollout 文件路径、创建时间和会话来源 → 它把这些放进 ThreadMetadataBuilder，并把工作目录、模型供应商、CLI 版本等暂时留空，把沙箱策略设成只读，把审批模式设成按需询问 → 出来的是一个可以继续补资料的 builder，不会直接生成最终资料卡。

调用关系：很多准备测试数据、种子数据、写入线程元数据的流程会先调用它。它内部会用外部的 PathBuf::new 之类创建空路径，并用 SandboxPolicy::new_read_only_policy 给新线程一个保守的默认安全策略。

调用图：被 35 处调用（test_thread_metadata, seed_stage1_output, thread_list_parent_filter_reads_direct_children_from_state_db, seed_recent_thread, upsert_thread_metadata, seed_thread_metadata, seed_stage1_candidate, seed_stage1_output, builder_from_items, builder_from_session_meta (+15 more)）；外部调用 2 个（new, new_read_only_policy）。

fn build(&self, default_provider: &str) -> ThreadMetadata

作用：把半成品 builder 变成完整的 ThreadMetadata。它会补齐缺失值，并把一些枚举值转换成适合保存和展示的字符串。

数据流：进去的是 builder 自己保存的字段，以及一个默认模型供应商名称 → 它把来源、沙箱策略、审批模式转成字符串，把时间统一成毫秒精度，缺失的更新时间用创建时间代替，缺失的模型供应商用默认值代替，还会尝试从来源里补 agent 路径 → 出来的是一张完整的线程资料卡 ThreadMetadata。

调用关系：它通常接在 ThreadMetadataBuilder::new 和字段补充之后使用。它把格式转换交给 enum_to_string，把时间规整交给 canonicalize_datetime，最后组装出真正会进入状态库或列表流程的数据。

调用图：调用 2 个内部函数（enum_to_string, canonicalize_datetime）；外部调用 2 个（clone, new）。

fn prefer_existing_git_info(&mut self, existing: &Self)

作用：合并新旧资料卡时，尽量保留旧资料卡里已经知道的 Git 信息。这样重新从日志生成资料时，不会把已有的提交号、分支名、远程仓库地址弄丢。

数据流：进去的是当前这张可修改的资料卡 self，以及一张已有资料卡 existing → 它检查 existing 里的 git_sha、git_branch、git_origin_url，只要旧值存在，就复制到当前资料卡 → 出来没有新返回值，但当前资料卡的 Git 字段可能被旧值覆盖。

调用关系：它用于“根据 rollout 文件重新整理元数据”这类合并场景。它不调用别的项目函数，只做简单判断和复制，目的是保护数据库中比日志更完整的信息。

fn prefer_existing_explicit_title(&mut self, existing: &Self)

作用：合并新旧资料卡时，尽量保留用户真正起过的标题。它会避免把用户标题误换成自动生成的第一条消息。

数据流：进去的是当前资料卡 self 和已有资料卡 existing → 它先判断旧标题是不是空，或者旧标题其实只是第一条用户消息；如果旧标题像是用户手动改过的，再看当前标题是不是空或自动标题；符合条件就把旧标题复制过来 → 出来没有返回值，但 self.title 可能被保留下来的旧标题替换。

调用关系：它常用于元数据重建或同步时。它不依赖外部函数，重点是在“自动标题”和“用户明确标题”之间做一个保护用户意图的选择。

fn diff_fields(&self, other: &Self) -> Vec<&'static str>

作用：比较两张线程资料卡，告诉调用者哪些字段不一样。它适合用来判断数据库是否需要更新，或者调试为什么两份元数据不同。

数据流：进去的是 self 和另一张 ThreadMetadata → 它逐个比较 ID、路径、时间、来源、模型、标题、Git 信息等字段 → 出来的是一个字段名列表，只包含发生变化的字段；它不改动两张资料卡。

调用关系：它是元数据对账的小工具。内部只是创建一个列表并逐项加入差异字段，不把工作交给复杂子流程。

调用图：外部调用 1 个（new）。

fn canonicalize_datetime(dt: DateTime<Utc>) -> DateTime<Utc>

作用：把时间统一到项目实际保存的精度，避免同一个时间因为纳秒等细节不同而被误认为变化了。

数据流：进去的是一个 UTC 时间 → 它先用 datetime_to_epoch_millis 转成 Unix 毫秒时间戳，再用 epoch_millis_to_datetime 转回 DateTime → 出来的是被规整后的时间；如果转换失败，就保留原时间。

调用关系：ThreadMetadataBuilder::build 会调用它来规整 created_at、updated_at、archived_at。它自己把两步转换分别交给 datetime_to_epoch_millis 和 epoch_millis_to_datetime。

调用图：调用 2 个内部函数（datetime_to_epoch_millis, epoch_millis_to_datetime）；被 1 处调用（build）。

fn try_from_row(row: &SqliteRow) -> Result<Self>

作用：把 SQLite 查询结果中的一行，读成 ThreadRow 这个临时结构。SQLite 是本地数据库，这一步相当于把数据库表格的一行搬进程序内存。

数据流：进去的是 SqliteRow，也就是数据库返回的一行 → 它按列名读取 id、rollout_path、created_at、model、title、Git 信息等字段 → 成功时出来一个 ThreadRow；如果某列不存在或类型不对，就返回错误。

调用关系：它位于数据库读取之后、变成正式 ThreadMetadata 之前。它把每个字段的读取交给 sqlx 的 try_get，自己负责把整行收集齐。

调用图：外部调用 1 个（try_get）。

fn try_from(row: ThreadRow) -> std::result::Result<Self, Self::Error>

作用：把数据库口味的 ThreadRow 转成程序真正使用的 ThreadMetadata。它会做校验和格式转换，防止坏 ID、坏时间戳直接混进系统。

数据流：进去的是 ThreadRow，里面很多值还是字符串或整数时间戳 → 它把字符串 ID 转成 ThreadId，把路径字符串转成 PathBuf，把毫秒时间戳转成 UTC 时间，把 thread_source 和 reasoning_effort 尝试解析成更明确的类型，并把空字符串预览、空第一条消息变成 None → 出来是 ThreadMetadata；遇到无效 ID、无效时间等会返回错误。

调用关系：测试函数 thread_row_parses_reasoning_effort 和 thread_row_preserves_model_defined_reasoning_effort_values 会直接验证它。它会调用 ThreadId 的 try_from、epoch_millis_to_datetime、PathBuf::from 等转换函数，是数据库行进入业务模型的关键关口。

调用图：调用 2 个内部函数（try_from, epoch_millis_to_datetime）；被 2 处调用（thread_row_parses_reasoning_effort, thread_row_preserves_model_defined_reasoning_effort_values）；外部调用 1 个（from）。

fn anchor_from_item(item: &ThreadMetadata, sort_key: SortKey) -> Option<Anchor>

作用：从一条线程资料卡里取出分页用的“锚点”。锚点可以理解成书签：下一页从这个时间位置继续往后翻。

数据流：进去的是一条 ThreadMetadata 和排序依据 SortKey → 如果按创建时间排序，就取 created_at；如果按更新时间排序，就取 updated_at → 出来是包含这个时间的 Anchor。

调用关系：它服务于线程列表的 keyset pagination，也就是按上一次最后看到的位置继续查下一页的分页方式。它不调用其它函数，只根据排序规则挑一个时间。

fn datetime_to_epoch_millis(dt: DateTime<Utc>) -> i64

作用：把 UTC 时间转成 Unix 毫秒时间戳。Unix 时间戳就是从 1970 年 1 月 1 日到这个时间经过了多久，这里单位是毫秒。

数据流：进去的是 DateTime<Utc> → 它调用时间库的 timestamp_millis 取出毫秒数 → 出来是 i64 整数，方便存数据库和比较大小。

调用关系：canonicalize_datetime 会调用它作为第一步，把时间先压成数据库常用的数字格式。

调用图：被 1 处调用（canonicalize_datetime）；外部调用 1 个（timestamp_millis）。

fn datetime_to_epoch_seconds(dt: DateTime<Utc>) -> i64

作用：把 UTC 时间转成 Unix 秒时间戳。它适合给仍然按秒保存的字段使用，比如归档时间的某些存法。

数据流：进去的是 DateTime<Utc> → 它调用时间库的 timestamp 取出秒数 → 出来是 i64 整数。

调用关系：这个函数是时间格式转换的小工具。它不在给出的调用图里被其它函数调用，但和 epoch_seconds_to_datetime 是一对相反方向的转换。

调用图：外部调用 1 个（timestamp）。

fn epoch_millis_to_datetime(value: i64) -> Result<DateTime<Utc>>

作用：把 Unix 毫秒时间戳转回 UTC 时间，并兼容旧数据库里按“秒”存的时间。这样老数据和新数据能一起正常排序。

数据流：进去的是一个整数时间戳 → 它先判断这个数如果小得像旧式秒时间戳，就乘以 1000 当毫秒用；然后调用时间库从毫秒构造 DateTime → 出来是 UTC 时间；如果数字根本不是合法时间，就返回错误。

调用关系：ThreadMetadata::try_from 用它把数据库中的 created_at、updated_at 转成时间；canonicalize_datetime 也用它把毫秒数转回标准时间。

调用图：被 2 处调用（try_from, canonicalize_datetime）；外部调用 1 个（from_timestamp_millis）。

fn epoch_seconds_to_datetime(value: i64) -> Result<DateTime<Utc>>

作用：把 Unix 秒时间戳转成 UTC 时间。它用于那些明确按秒保存的字段。

数据流：进去的是一个秒级整数时间戳 → 它调用时间库 from_timestamp 生成 UTC 时间 → 出来是 DateTime；如果时间戳非法，就返回错误。

调用关系：ThreadMetadata::try_from 在处理 archived_at 这类秒级时间时会用到它。它和 datetime_to_epoch_seconds 是反向的一组工具。

调用图：外部调用 1 个（from_timestamp）。

fn thread_row(reasoning_effort: Option<&str>) -> ThreadRow

作用：给测试造一条假的数据库行。测试可以用它快速得到一个 ThreadRow，不用真的连数据库。

数据流：进去的是一个可选的 reasoning_effort 字符串，比如 high 或 future → 它填好固定的 ID、路径、时间、模型、工作目录等字段，并把传入的 reasoning_effort 放进去 → 出来是一条 ThreadRow 测试数据。

调用关系：测试函数 thread_row_parses_reasoning_effort 和 thread_row_preserves_model_defined_reasoning_effort_values 会调用它，然后再交给 ThreadMetadata::try_from 做转换验证。

调用图：外部调用 1 个（new）。

fn expected_thread_metadata(reasoning_effort: Option<ReasoningEffort>) -> ThreadMetadata

作用：给测试造出“应该得到的正确答案”。这样测试能把实际转换结果和预期资料卡逐字段比较。

数据流：进去的是一个可选的 ReasoningEffort 值 → 它创建固定的 ThreadId、路径、UTC 时间、模型、工作目录等字段，并填入传入的推理强度 → 出来是一张预期的 ThreadMetadata。

调用关系：两个 reasoning_effort 相关测试都会用它生成对照答案。它调用 ThreadId::from_string、DateTime::from_timestamp 和 PathBuf::from 之类函数来构造可信的期望值。

调用图：调用 1 个内部函数（from_string）；外部调用 3 个（from_timestamp, from, new）。

fn thread_row_parses_reasoning_effort()

作用：测试普通的 reasoning_effort 字符串能被正确解析。这里验证 high 会变成 ReasoningEffort::High。

数据流：进去没有外部输入 → 它用 tests::thread_row 造一条 reasoning_effort 为 high 的行，调用 ThreadMetadata::try_from 转换，再和 tests::expected_thread_metadata 生成的期望值比较 → 如果结果不一致，测试失败。

调用关系：它直接覆盖 ThreadMetadata::try_from 的解析行为，确保数据库里保存的常见推理强度不会读错。

调用图：调用 1 个内部函数（try_from）；外部调用 2 个（assert_eq!, thread_row）。

fn thread_row_preserves_model_defined_reasoning_effort_values()

作用：测试未来或模型自定义的 reasoning_effort 值不会被丢掉。比如 future 不是固定枚举值，也应该保留下来。

数据流：进去没有外部输入 → 它造一条 reasoning_effort 为 future 的 ThreadRow，转换成 ThreadMetadata，再检查结果是不是 ReasoningEffort::Custom("future") → 如果自定义值被丢弃或报错，测试失败。

调用关系：它保护 ThreadMetadata::try_from 的向前兼容能力。意思是以后模型加了新推理强度，旧程序至少还能把字符串保留下来，而不是直接坏掉。

调用图：调用 1 个内部函数（try_from）；外部调用 2 个（assert_eq!, thread_row）。

async fn get_thread(&self, id: ThreadId) -> anyhow::Result<Option<crate::ThreadMetadata>>

作用：按线程 ID 从数据库里读取一条完整的线程资料。别人需要知道某个会话的标题、路径、模型、Git 信息等详情时会用它。

数据流：进去的是一个线程 ID；它把 ID 转成数据库能查的字符串，查询 threads 表，把查到的一行转换成 ThreadMetadata；出来的是“没有这条记录”或一份完整资料。

调用关系：归档、取消归档和应用 rollout 增量前都会先用它拿旧资料，再在旧资料基础上改动后写回。

调用图：被 3 处调用（apply_rollout_items, mark_archived, mark_unarchived）；外部调用 2 个（to_string, query）。

async fn get_thread_memory_mode(&self, id: ThreadId) -> anyhow::Result<Option<String>>

作用：只读取某个线程的 memory_mode 字段，也就是这个线程的记忆功能模式。需要快速检查记忆开关时不用加载整条线程资料。

数据流：进去的是线程 ID；它查询 threads 表中的 memory_mode；出来的是可能存在的字符串，查不到线程或字段读取失败就返回空。

调用关系：它是一个独立的轻量读取接口，主要给上层在关心记忆模式时直接调用。

调用图：外部调用 2 个（to_string, query）。

async fn set_thread_preview_if_empty(
        &self,
        thread_id: ThreadId,
        preview: &str,
    ) -> anyhow::Result<bool>

作用：只在线程预览为空时填上一段预览文字。这样可以补齐旧数据，又不会覆盖用户已经看到的摘要。

数据流：进去的是线程 ID 和预览文字；它先去掉首尾空白，空文字直接不做事；否则只更新 preview 仍为空字符串的那一行；出来是是否真的改到了数据库。

调用关系：它通常被补数据或迁移逻辑调用，用来安全地填空，不参与完整线程写入流程。

调用图：外部调用 2 个（to_string, query）。

async fn upsert_thread_spawn_edge(
        &self,
        parent_thread_id: ThreadId,
        child_thread_id: ThreadId,
        status: crate::DirectionalThreadSpawnEdgeStatus,
    ) -> anyhow::Resul

作用：保存或替换一条“父线程启动了子线程”的关系，并记录这条关系当前是打开还是关闭。没有它，多代理场景下就不知道哪个子任务属于哪个父任务。

数据流：进去的是父线程 ID、子线程 ID 和关系状态；它写入 thread_spawn_edges 表，如果这个子线程已有父关系就更新；出来没有额外数据，只保证数据库保存成功。

调用关系：列表、查找和删除子树都会依赖这些边；测试和运行时代码都会用它建立父子关系。

调用图：外部调用 3 个（to_string, query, as_ref）。

async fn set_thread_spawn_edge_status(
        &self,
        child_thread_id: ThreadId,
        status: crate::DirectionalThreadSpawnEdgeStatus,
    ) -> anyhow::Result<()>

作用：更新某个子线程入口关系的状态。比如子任务结束后，可以把关系从打开改成关闭。

数据流：进去的是子线程 ID 和新状态；它按 child_thread_id 更新 thread_spawn_edges 表；出来只表示操作是否成功执行，函数本身不返回是否命中。

调用关系：它和查询打开/关闭子线程的函数配合，用来让线程家谱反映任务生命周期。

调用图：外部调用 3 个（to_string, query, as_ref）。

async fn list_thread_spawn_children_with_status(
        &self,
        parent_thread_id: ThreadId,
        status: crate::DirectionalThreadSpawnEdgeStatus,
    ) -> anyhow::Result<Vec<ThreadId>>

作用：列出某个父线程下面，状态匹配的直接子线程。比如只想看还打开的子任务时会用它。

数据流：进去的是父线程 ID 和状态；它把实际查询交给通用的匹配函数；出来是按 ID 排好的子线程 ID 列表。

调用关系：这是带状态过滤的公开入口，内部复用 StateRuntime::list_thread_spawn_children_matching。

调用图：调用 1 个内部函数（list_thread_spawn_children_matching）。

async fn list_thread_spawn_children(
        &self,
        parent_thread_id: ThreadId,
    ) -> anyhow::Result<Vec<ThreadId>>

作用：列出某个父线程的所有直接子线程，不管关系状态是什么。适合需要完整看一层子任务的地方。

数据流：进去的是父线程 ID；它调用通用匹配函数但不传状态过滤；出来是所有直接子线程 ID。

调用关系：这是无状态过滤的公开入口，内部复用 StateRuntime::list_thread_spawn_children_matching。

调用图：调用 1 个内部函数（list_thread_spawn_children_matching）。

async fn list_thread_spawn_descendants_with_status(
        &self,
        root_thread_id: ThreadId,
        status: crate::DirectionalThreadSpawnEdgeStatus,
    ) -> anyhow::Result<Vec<ThreadId>>

作用：列出某个根线程下面所有状态匹配的后代线程，不只是一层孩子。适合追踪一整棵还处于某种状态的子任务树。

数据流：进去的是根线程 ID 和状态；它委托递归查询函数；出来是先近后远、同层按 ID 排序的线程 ID 列表。

调用关系：这是带状态过滤的公开入口，内部复用 StateRuntime::list_thread_spawn_descendants_matching。

调用图：调用 1 个内部函数（list_thread_spawn_descendants_matching）。

async fn list_thread_spawn_descendants(
        &self,
        root_thread_id: ThreadId,
    ) -> anyhow::Result<Vec<ThreadId>>

作用：列出某个根线程启动出来的所有后代线程。就像查家谱，不只看儿女，也看孙辈和更深层。

数据流：进去的是根线程 ID；它调用递归查询但不限制状态；出来是稳定排序后的所有后代线程 ID。

调用关系：删除和排查线程树时会依赖这种完整子树视图，内部复用 StateRuntime::list_thread_spawn_descendants_matching。

调用图：调用 1 个内部函数（list_thread_spawn_descendants_matching）。

async fn find_thread_spawn_child_by_path(
        &self,
        parent_thread_id: ThreadId,
        agent_path: &str,
    ) -> anyhow::Result<Option<ThreadId>>

作用：在某个父线程的直接子线程里，按 agent_path 找对应线程。agent_path 可以理解成代理的固定路径名，用来定位某个代理实例。

数据流：进去的是父线程 ID 和路径字符串；它连接父子关系表和线程表查最多两条；再交给 one_thread_id_from_rows 判断是没有、唯一一个，还是重复冲突；出来是可选线程 ID。

调用关系：它负责直接孩子的查找，最后用 one_thread_id_from_rows 做唯一性检查，防止同一路径对应多个线程时悄悄选错。

调用图：调用 1 个内部函数（one_thread_id_from_rows）；外部调用 2 个（to_string, query）。

async fn find_thread_spawn_descendant_by_path(
        &self,
        root_thread_id: ThreadId,
        agent_path: &str,
    ) -> anyhow::Result<Option<ThreadId>>

作用：在某个根线程的整棵后代树里，按 agent_path 找线程。适合子代理可能藏在多层任务下面的情况。

数据流：进去的是根线程 ID 和路径；它用递归 SQL 先找全部后代，再按路径过滤并最多取两条；出来是没有、唯一 ID，或重复时报错。

调用关系：它和直接孩子查找类似，但范围扩大到整棵树，也复用 one_thread_id_from_rows 做防误选检查。

调用图：调用 1 个内部函数（one_thread_id_from_rows）；外部调用 2 个（to_string, query）。

async fn list_thread_spawn_children_matching(
        &self,
        parent_thread_id: ThreadId,
        status: Option<crate::DirectionalThreadSpawnEdgeStatus>,
    ) -> anyhow::Result<Vec<ThreadId>>

作用：这是列直接子线程的内部通用函数，可选地按状态过滤。公开的两个 children 列表函数都靠它干实际查询。

数据流：进去的是父线程 ID 和可选状态；它动态拼出安全的 SQL 查询，读取 child_thread_id；出来是 ThreadId 列表。

调用关系：StateRuntime::list_thread_spawn_children 和 StateRuntime::list_thread_spawn_children_with_status 都把请求转给它。

调用图：被 2 处调用（list_thread_spawn_children, list_thread_spawn_children_with_status）；外部调用 2 个（new, to_string）。

async fn list_thread_spawn_descendants_matching(
        &self,
        root_thread_id: ThreadId,
        status: Option<crate::DirectionalThreadSpawnEdgeStatus>,
    ) -> anyhow::Result<Vec<ThreadId>

作用：这是列所有后代线程的内部通用函数，可选地按状态过滤。它用数据库递归查询来走完整棵父子树。

数据流：进去的是根线程 ID 和可选状态；它构造 WITH RECURSIVE 查询，一层层找子线程并记录深度；出来是按深度和 ID 排好的后代 ID。

调用关系：StateRuntime::list_thread_spawn_descendants 和 StateRuntime::list_thread_spawn_descendants_with_status 都调用它。

调用图：被 2 处调用（list_thread_spawn_descendants, list_thread_spawn_descendants_with_status）；外部调用 2 个（new, to_string）。

async fn insert_thread_spawn_edge_if_absent(
        &self,
        parent_thread_id: ThreadId,
        child_thread_id: ThreadId,
    ) -> anyhow::Result<()>

作用：如果还没有父子关系，就补上一条默认打开状态的关系。它不会覆盖已有关系。

数据流：进去的是父线程 ID 和子线程 ID；它尝试插入 thread_spawn_edges，冲突时什么也不做；出来只表示数据库操作完成。

调用关系：它由 StateRuntime::insert_thread_spawn_edge_from_source_if_absent 调用，用来把线程来源里隐藏的父子关系落库。

调用图：被 1 处调用（insert_thread_spawn_edge_from_source_if_absent）；外部调用 2 个（to_string, query）。

async fn insert_thread_spawn_edge_from_source_if_absent(
        &self,
        child_thread_id: ThreadId,
        source: &str,
    ) -> anyhow::Result<()>

作用：从线程 source 字段里解析父线程 ID，并在缺失时补父子关系。这样旧数据或外部传来的来源信息也能形成线程家谱。

数据流：进去的是子线程 ID 和 source 字符串；它先解析 source，如果没有父线程就结束；如果有，就调用插入关系函数；出来没有额外数据。

调用关系：插入线程和 upsert 线程后都会调用它，实际插入动作交给 StateRuntime::insert_thread_spawn_edge_if_absent。

调用图：调用 2 个内部函数（insert_thread_spawn_edge_if_absent, thread_spawn_parent_thread_id_from_source_str）；被 2 处调用（insert_thread_if_absent, upsert_thread_with_creation_memory_mode）。

async fn find_rollout_path_by_id(
        &self,
        id: ThreadId,
        archived_only: Option<bool>,
    ) -> anyhow::Result<Option<PathBuf>>

作用：按线程 ID 找它对应的 rollout 文件路径。rollout 文件可以理解成会话原始流水账，数据库记录需要和它对应起来。

数据流：进去的是线程 ID 和是否只查归档记录的选项；它按条件查 rollout_path；出来是可选的 PathBuf 路径。

调用关系：它是数据库到磁盘文件之间的定位工具，上层想从线程记录回到原始文件时会用。

调用图：外部调用 2 个（new, to_string）。

async fn find_thread_by_exact_title(
        &self,
        title: &str,
        allowed_sources: &[String],
        model_providers: Option<&[String]>,
        archived_only: bool,
        cwd: Optio

作用：找标题完全等于某个字符串的最新线程。适合按用户看到的标题快速恢复一个会话。

数据流：进去的是标题、来源限制、模型提供方限制、归档条件和可选工作目录；它复用统一的列选择、过滤和排序拼 SQL；出来是最新的一条线程资料或空。

调用关系：它把查询构造工作交给 push_thread_select_columns、push_thread_filters 和 push_thread_order_and_limit。

调用图：调用 3 个内部函数（push_thread_filters, push_thread_order_and_limit, push_thread_select_columns）；外部调用 1 个（new）。

async fn list_threads(
        &self,
        page_size: usize,
        filters: ThreadFilterOptions<'_>,
    ) -> anyhow::Result<crate::ThreadsPage>

作用：分页列出线程列表。页面上展示最近会话、历史会话时通常走这个入口。

数据流：进去的是页大小和过滤条件；它把父线程限制设为空，交给通用列表函数；出来是一页线程和下一页锚点。

调用关系：它是普通全局列表入口，实际逻辑在 StateRuntime::list_threads_matching。

调用图：调用 1 个内部函数（list_threads_matching）。

async fn list_threads_by_parent(
        &self,
        page_size: usize,
        parent_thread_id: ThreadId,
        filters: ThreadFilterOptions<'_>,
    ) -> anyhow::Result<crate::ThreadsPage>

作用：分页列出某个父线程的直接子线程资料。比只返回 ID 更适合界面展示子任务列表。

数据流：进去的是页大小、父线程 ID 和过滤条件；它把父线程 ID 交给通用列表函数；出来是一页子线程资料和下一页锚点。

调用关系：它是按父线程列表的公开入口，实际逻辑在 StateRuntime::list_threads_matching。

调用图：调用 1 个内部函数（list_threads_matching）。

async fn list_threads_matching(
        &self,
        page_size: usize,
        filters: ThreadFilterOptions<'_>,
        parent_thread_id: Option<ThreadId>,
    ) -> anyhow::Result<crate::ThreadsPag

作用：这是线程分页列表的核心实现。它负责多查一条来判断是否还有下一页。

数据流：进去的是页大小、过滤条件和可选父线程 ID；它构造查询，取 page_size+1 条，转换成元数据；如果多出一条就丢掉并生成 next_anchor；出来是 ThreadsPage。

调用关系：StateRuntime::list_threads 和 StateRuntime::list_threads_by_parent 都调用它，查询文本由 push_list_threads_query 生成。

调用图：调用 1 个内部函数（push_list_threads_query）；被 2 处调用（list_threads, list_threads_by_parent）；外部调用 1 个（new）。

async fn list_thread_ids(
        &self,
        limit: usize,
        anchor: Option<&crate::Anchor>,
        sort_key: crate::SortKey,
        allowed_sources: &[String],
        model_providers: Op

作用：只列线程 ID，不加载完整资料。适合后台批处理只需要知道哪些线程存在的情况。

数据流：进去的是数量限制、翻页锚点、排序字段、来源和模型过滤、归档条件；它拼出只选 id 的 SQL；出来是 ThreadId 列表。

调用关系：它复用 push_thread_filters 和 push_thread_order_and_limit，和完整列表保持同一套过滤排序规则。

调用图：调用 2 个内部函数（push_thread_filters, push_thread_order_and_limit）；外部调用 1 个（new）。

async fn upsert_thread(&self, metadata: &crate::ThreadMetadata) -> anyhow::Result<()>

作用：插入或更新一条线程资料，是最常用的保存入口。upsert 的意思是“有就更新，没有就插入”。

数据流：进去的是 ThreadMetadata；它把记忆模式参数留空，转交给更底层的 upsert 函数；出来只表示保存成功或失败。

调用关系：应用 rollout、归档和取消归档都会调用它；真正写数据库的是 StateRuntime::upsert_thread_with_creation_memory_mode。

调用图：调用 1 个内部函数（upsert_thread_with_creation_memory_mode）；被 3 处调用（apply_rollout_items, mark_archived, mark_unarchived）。

async fn insert_thread_if_absent(
        &self,
        metadata: &crate::ThreadMetadata,
    ) -> anyhow::Result<bool>

作用：只在线程不存在时插入，存在时绝不覆盖。适合用低可信或备用资料补建索引，避免把新资料写旧。

数据流：进去的是线程资料；它分配更新时间、生成预览，执行 INSERT ... DO NOTHING；随后尝试补父子关系；出来是是否真的插入了新行。

调用关系：它调用 allocate_thread_updated_at、metadata_preview 和 StateRuntime::insert_thread_spawn_edge_from_source_if_absent。

调用图：调用 3 个内部函数（allocate_thread_updated_at, insert_thread_spawn_edge_from_source_if_absent, metadata_preview）；外部调用 1 个（query）。

async fn set_thread_memory_mode(
        &self,
        thread_id: ThreadId,
        memory_mode: &str,
    ) -> anyhow::Result<bool>

作用：更新某个线程的记忆模式。比如从 rollout 里读到新的 memory_mode 后，需要把数据库里的值同步过去。

数据流：进去的是线程 ID 和模式字符串；它更新 threads 表；出来是是否有行被改到。

调用关系：StateRuntime::apply_rollout_items 在从新流水项里发现 memory_mode 后会调用它。

调用图：被 1 处调用（apply_rollout_items）；外部调用 2 个（to_string, query）。

async fn update_thread_title(
        &self,
        thread_id: ThreadId,
        title: &str,
    ) -> anyhow::Result<bool>

作用：单独修改线程标题，不碰别的字段。适合用户重命名会话。

数据流：进去的是线程 ID 和新标题；它只更新 title 字段；出来是是否命中并修改了线程。

调用关系：这是独立的小更新入口，不走完整 metadata upsert，避免无关字段被覆盖。

调用图：外部调用 2 个（to_string, query）。

async fn touch_thread_updated_at(
        &self,
        thread_id: ThreadId,
        updated_at: DateTime<Utc>,
    ) -> anyhow::Result<bool>

作用：只刷新线程的更新时间，不改标题、内容预览等资料。适合某些事件只需要让会话排到更近的位置。

数据流：进去的是线程 ID 和候选时间；它先通过 allocate_thread_updated_at 处理成适合排序的时间，再更新秒和毫秒两列；出来是是否改到行。

调用关系：它复用更新时间分配规则，和插入、upsert 保持分页排序一致。

调用图：调用 1 个内部函数（allocate_thread_updated_at）；外部调用 2 个（to_string, query）。

fn allocate_thread_updated_at(
        &self,
        updated_at: DateTime<Utc>,
    ) -> anyhow::Result<DateTime<Utc>>

作用：给线程更新时间分配一个适合排序的毫秒时间。它解决多个热更新同一毫秒发生时，分页游标可能分不清先后的问题。

数据流：进去的是候选 DateTime；它和进程内保存的最高毫秒值比较：新的时间直接用，很旧的历史回填保持原样，太接近的热更新就加 1 毫秒；出来是分配后的 DateTime。

调用关系：insert_thread_if_absent、touch_thread_updated_at 和 upsert_thread_with_creation_memory_mode 都依赖它保证列表排序稳定。

调用图：被 3 处调用（insert_thread_if_absent, touch_thread_updated_at, upsert_thread_with_creation_memory_mode）。

async fn update_thread_git_info(
        &self,
        thread_id: ThreadId,
        git_sha: Option<Option<&str>>,
        git_branch: Option<Option<&str>>,
        git_origin_url: Option<Option<&str

作用：只更新线程的 Git 信息，比如提交号、分支、远端地址。它刻意不碰其他线程资料，防止并发写入时误覆盖。

数据流：进去的是线程 ID，以及三个“是否更新、更新成什么”的可选值；它用 SQL 的 CASE 只改被指定的字段；出来是是否命中线程。

调用关系：这是 Git 信息的专用修补入口，和完整 upsert 分开，避免把标题、token、预览等新数据写回旧值。

调用图：外部调用 2 个（to_string, query）。

async fn upsert_thread_with_creation_memory_mode(
        &self,
        metadata: &crate::ThreadMetadata,
        creation_memory_mode: Option<&str>,
    ) -> anyhow::Result<()>

作用：真正执行线程插入或更新的底层函数，还能在首次创建时指定 memory_mode。它是这个文件保存线程资料的核心。

数据流：进去的是线程资料和可选创建时记忆模式；它分配更新时间、生成预览，执行 INSERT ... ON CONFLICT UPDATE；更新时保留已有非空 Git 字段和已有预览；最后补父子关系；出来没有额外数据。

调用关系：StateRuntime::upsert_thread 和 StateRuntime::apply_rollout_items 都调用它；它又调用 allocate_thread_updated_at、metadata_preview 和插入父子关系的函数。

调用图：调用 3 个内部函数（allocate_thread_updated_at, insert_thread_spawn_edge_from_source_if_absent, metadata_preview）；被 2 处调用（apply_rollout_items, upsert_thread）；外部调用 1 个（query）。

async fn apply_rollout_items(
        &self,
        builder: &ThreadMetadataBuilder,
        items: &[RolloutItem],
        new_thread_memory_mode: Option<&str>,
        updated_at_override: Option<D

作用：把新读到的 rollout 流水项增量应用到数据库里的线程资料。也就是把原始会话日志变成可搜索、可排序的线程索引。

数据流：进去的是元数据构建器、一批流水项、新线程记忆模式和可选更新时间；它先取旧资料，没有就新建，再逐条应用流水项，保留旧 Git 信息，确定更新时间，最后插入或更新；如果流水项里有 memory_mode 也单独写入；出来表示同步完成。

调用关系：它调用 get_thread 读取旧状态，调用 upsert_thread 或 upsert_thread_with_creation_memory_mode 写回，并用 extract_memory_mode 找记忆模式。

调用图：调用 5 个内部函数（get_thread, set_thread_memory_mode, upsert_thread, upsert_thread_with_creation_memory_mode, extract_memory_mode）；外部调用 1 个（is_empty）。

async fn mark_archived(
        &self,
        thread_id: ThreadId,
        rollout_path: &Path,
        archived_at: DateTime<Utc>,
    ) -> anyhow::Result<()>

作用：把线程标记为已归档。归档后的线程会从普通列表中隐藏，进入归档列表。

数据流：进去的是线程 ID、新 rollout 路径和归档时间；它读取旧资料，设置 archived_at，更新路径和文件修改时间，最后 upsert 写回；查不到线程就什么也不做。

调用关系：它先用 get_thread 取资料，再用 upsert_thread 保存；如果读到的 ID 不一致会记录警告。

调用图：调用 2 个内部函数（get_thread, upsert_thread）；外部调用 2 个（to_path_buf, warn!）。

async fn mark_unarchived(
        &self,
        thread_id: ThreadId,
        rollout_path: &Path,
    ) -> anyhow::Result<()>

作用：把线程从归档状态恢复出来。这样它又会出现在普通线程列表里。

数据流：进去的是线程 ID 和恢复后的 rollout 路径；它读取旧资料，把 archived_at 清空，更新路径和修改时间，再写回；查不到就直接结束。

调用关系：它和 mark_archived 是一对反向操作，同样使用 get_thread 和 upsert_thread。

调用图：调用 2 个内部函数（get_thread, upsert_thread）；外部调用 2 个（to_path_buf, warn!）。

async fn delete_thread(&self, thread_id: ThreadId) -> anyhow::Result<u64>

作用：删除单个线程以及相关状态。它是单线程删除的方便入口。

数据流：进去的是一个线程 ID；它包装成数组交给严格删除函数；出来是实际删除的 threads 表行数。

调用关系：它把所有复杂清理都交给 StateRuntime::delete_threads_strict。

调用图：调用 1 个内部函数（delete_threads_strict）。

async fn delete_threads_strict(&self, thread_ids: &[ThreadId]) -> anyhow::Result<u64>

作用：严格删除一批线程，并清理日志、记忆、目标、动态工具、父子关系和任务分配。它特别注意删除顺序，保证失败后还能重试。

数据流：进去的是线程 ID 列表；空列表直接返回 0；它先清外部日志和记忆/目标，再在事务里取消或重置相关 agent job，删动态工具和父子边，最后删 threads 行并提交；出来是删除的线程行数。

调用关系：delete_thread 调用它。它是危险操作的核心，所以先保留足够的关系信息，直到最后才删除主表和 spawn 边。

调用图：被 1 处调用（delete_thread）；外部调用 4 个（now, is_empty, iter, query）。

fn one_thread_id_from_rows(
    rows: Vec<sqlx::sqlite::SqliteRow>,
    agent_path: &str,
) -> anyhow::Result<Option<ThreadId>>

作用：把查询结果检查成“最多一个线程 ID”。如果同一个 agent_path 找到多个线程，它会报错而不是随便选一个。

数据流：进去的是数据库行和路径字符串；它把每行 id 转成 ThreadId，数数量；0 个返回空，1 个返回该 ID，多于 1 个返回错误。

调用关系：find_thread_spawn_child_by_path 和 find_thread_spawn_descendant_by_path 都用它做唯一性把关。

调用图：被 2 处调用（find_thread_spawn_child_by_path, find_thread_spawn_descendant_by_path）；外部调用 1 个（anyhow!）。

fn push_list_threads_query(
    builder: &mut QueryBuilder<Sqlite>,
    filters: ThreadFilterOptions<'_>,
    parent_thread_id: Option<ThreadId>,
    limit: usize,
)

作用：拼出完整的线程列表查询 SQL。它集中处理选择哪些列、怎么过滤、是否只看某个父线程的孩子、以及怎么排序限制数量。

数据流：进去的是 SQL 构造器、过滤条件、可选父线程 ID 和 limit；它依次追加 SELECT、FROM、WHERE、父子限制、ORDER BY 和 LIMIT；出来是被填好的查询构造器。

调用关系：list_threads_matching 用它生成实际查询；索引计划测试也直接调用它确认数据库会走合适索引。

调用图：调用 3 个内部函数（push_thread_filters, push_thread_order_and_limit, push_thread_select_columns）；被 2 处调用（list_threads_matching, list_threads_uses_indexes_matching_cwd_filters）；外部调用 2 个（push, push_bind）。

fn push_thread_select_columns(builder: &mut QueryBuilder<Sqlite>)

作用：统一声明线程列表和详情查询要取哪些列。这样不同查询拿到的字段顺序和名字一致，后面才能稳定转换成 ThreadMetadata。

数据流：进去的是 SQL 构造器；它追加一大段 SELECT 列清单，并把毫秒时间列起名成 created_at、updated_at；出来是更新后的构造器。

调用关系：find_thread_by_exact_title 和 push_list_threads_query 都调用它，避免重复写同一套列名。

调用图：被 2 处调用（find_thread_by_exact_title, push_list_threads_query）；外部调用 1 个（push）。

fn extract_memory_mode(items: &[RolloutItem]) -> Option<String>

作用：从一批 rollout 项里找最近一次出现的 memory_mode。倒着找是因为后面的记录更接近当前状态。

数据流：进去的是流水项切片；它从后往前看，只检查 SessionMeta 类型里的 memory_mode；出来是找到的字符串或空。

调用关系：apply_rollout_items 用它决定是否要再调用 set_thread_memory_mode 更新数据库。

调用图：被 1 处调用（apply_rollout_items）；外部调用 1 个（iter）。

fn thread_spawn_parent_thread_id_from_source_str(source: &str) -> Option<ThreadId>

作用：从 source 字符串里解析父线程 ID。source 可能是 JSON，也可能是普通字符串形式，所以这里做了兼容。

数据流：进去的是 source 文本；它先尝试按 JSON 解析，失败再按字符串值解析成 SessionSource；如果解析成功且里面有父线程 ID，就返回它；否则返回空。

调用关系：insert_thread_spawn_edge_from_source_if_absent 调用它，从线程来源自动补父子关系。

调用图：被 1 处调用（insert_thread_spawn_edge_from_source_if_absent）；外部调用 1 个（from_str）。

fn push_thread_filters(
    builder: &mut QueryBuilder<Sqlite>,
    options: ThreadFilterOptions<'a>,
)

作用：给线程查询追加统一的过滤条件，比如是否归档、来源、模型提供方、工作目录、搜索词和翻页锚点。它让所有列表入口的筛选规则一致。

数据流：进去的是 SQL 构造器和过滤选项；它追加 WHERE 条件，并把外部输入都作为绑定参数放进去；出来是带过滤条件的查询。

调用关系：find_thread_by_exact_title、list_thread_ids、push_list_threads_query 以及启动任务领取逻辑都会用它。

调用图：被 4 处调用（claim_stage1_jobs_for_startup, find_thread_by_exact_title, list_thread_ids, push_list_threads_query）；外部调用 3 个（push, push_bind, separated）。

fn push_thread_order_and_limit(
    builder: &mut QueryBuilder<Sqlite>,
    sort_key: SortKey,
    sort_direction: SortDirection,
    order_by_index: OrderByIndex,
    limit: usize,
)

作用：给线程查询追加排序和数量限制。它也能选择是否让 SQLite 用排序字段索引，以避开某些情况下的慢查询计划。

数据流：进去的是排序字段、升降序、是否允许排序索引和 limit；它追加 ORDER BY 和 LIMIT；出来是更新后的 SQL 构造器。

调用关系：find_thread_by_exact_title、list_thread_ids 和 push_list_threads_query 都调用它，统一排序分页方式。

调用图：被 3 处调用（find_thread_by_exact_title, list_thread_ids, push_list_threads_query）；外部调用 2 个（push, push_bind）。

fn metadata_preview(metadata: &crate::ThreadMetadata) -> &str

作用：从线程资料里挑一段可显示的预览文字。优先用 preview，没有就用第一条用户消息，再没有就是空字符串。

数据流：进去的是 ThreadMetadata；它依次检查 preview 和 first_user_message；出来是一个字符串引用。

调用关系：insert_thread_if_absent 和 upsert_thread_with_creation_memory_mode 写数据库前都用它填 preview 字段。

调用图：被 2 处调用（insert_thread_if_absent, upsert_thread_with_creation_memory_mode）。

async fn upsert_thread_keeps_creation_memory_mode_for_existing_rows()

作用：验证线程创建时写入的 memory_mode，在后续普通 upsert 时不会被改掉。这个测试防止老配置被意外覆盖。

数据流：测试先建临时运行时和线程，带 disabled 模式插入，再改标题执行 upsert；最后读数据库确认 memory_mode 仍是 disabled。

调用关系：测试运行器调用它；它覆盖 upsert_thread_with_creation_memory_mode 和 upsert_thread 的配合行为。

调用图：调用 4 个内部函数（from_string, init, test_thread_metadata, unique_temp_dir）；外部调用 2 个（assert_eq!, query_scalar）。

async fn delete_thread_cleans_associated_state() -> Result<()>

作用：验证删除线程时，相关日志、目标、动态工具、父子关系和任务分配都会被清掉或重置。它保证删除不是只删一张主表。

数据流：测试准备父子线程、日志、目标、动态工具和运行中的 job item；调用 delete_threads_strict；之后逐项检查线程没了、周边状态干净、任务被取消或退回 pending。

调用关系：测试运行器调用它；它覆盖 delete_threads_strict、delete_thread 和辅助断言函数。

调用图：调用 4 个内部函数（from_string, init, test_thread_metadata, unique_temp_dir）；外部调用 8 个（assert!, assert_eq!, json!, query, query_scalar, assert_thread_cleanup_state, seed_thread_cleanup_state, vec!）。

async fn delete_thread_keeps_retry_graph_on_cleanup_failure() -> Result<()>

作用：验证删除过程中如果日志库出错，线程和父子关系不会先被删掉。这样下次还能根据原关系重试清理。

数据流：测试创建线程和子关系后关闭日志库，强制 delete_thread 失败；然后检查线程仍存在，后代关系仍能查到。

调用关系：测试运行器调用它；它保护 delete_threads_strict 的删除顺序设计。

调用图：调用 4 个内部函数（from_string, init, test_thread_metadata, unique_temp_dir）；外部调用 3 个（assert!, assert_eq!, seed_thread_cleanup_state）。

async fn seed_thread_cleanup_state(
        runtime: &StateRuntime,
        thread_id: ThreadId,
        child_thread_id: ThreadId,
    ) -> Result<()>

作用：给删除相关测试准备一批附属状态。它不是产品功能，只是测试夹具。

数据流：进去的是运行时、线程 ID 和子线程 ID；它插入父子关系、线程目标和一条日志；出来是准备好的测试环境。

调用关系：删除测试调用它，减少重复造数据代码。

调用图：调用 1 个内部函数（thread_goals）；外部调用 3 个（to_string, query, upsert_thread_spawn_edge）。

async fn assert_thread_cleanup_state(
        runtime: &StateRuntime,
        thread_id: ThreadId,
    ) -> Result<()>

作用：检查某个线程的附属状态是否已经清理干净。它帮助删除测试把多个断言集中起来。

数据流：进去的是运行时和线程 ID；它查询父子边数量、线程目标和日志；如果还有残留就断言失败。

调用关系：delete_thread_cleans_associated_state 调用它确认清理结果。

调用图：外部调用 7 个（default, assert!, assert_eq!, to_string, query_scalar, query_logs, vec!）。

async fn list_threads_updated_after_returns_oldest_changes_first()

作用：验证按更新时间升序翻页时，锚点之后的旧到新顺序正确。它防止分页漏掉或跳错线程。

数据流：测试插入多个更新时间相同或不同的线程，设置锚点分页查询；检查第一页和第二页的 ID 与 next_anchor。

调用关系：测试运行器调用它；它覆盖 list_threads、ThreadFilterOptions 和更新时间排序。

调用图：调用 4 个内部函数（from_string, init, test_thread_metadata, unique_temp_dir）；外部调用 2 个（from_timestamp, assert_eq!）。

async fn list_threads_filters_by_cwd()

作用：验证线程列表能按工作目录过滤，并且空目录过滤会返回空列表。工作目录就是会话运行时所在的文件夹。

数据流：测试插入三个不同 cwd 的线程，传入两个 cwd 做分页查询，再传空 cwd 列表查询；检查返回的线程符合预期。

调用关系：测试运行器调用它；它覆盖 push_thread_filters 里的 cwd_filters 行为。

调用图：调用 4 个内部函数（from_string, init, test_thread_metadata, unique_temp_dir）；外部调用 3 个（from_timestamp, assert_eq!, vec!）。

async fn list_threads_uses_indexes_matching_cwd_filters()

作用：验证不同过滤组合下 SQLite 会使用合适索引。索引像书的目录，选错会让查询变慢。

数据流：测试构造 EXPLAIN QUERY PLAN 查询，针对不同排序和 cwd 过滤组合生成 SQL；读取查询计划，断言用了期望索引以及是否需要临时排序。

调用关系：测试运行器调用它；它直接调用 push_list_threads_query 来检查生成 SQL 的性能倾向。

调用图：调用 3 个内部函数（init, unique_temp_dir, push_list_threads_query）；外部调用 5 个（from_timestamp, from, new, assert!, assert_eq!）。

async fn list_threads_by_parent_filters_direct_children_with_keyset_pagination()

作用：验证按父线程列子线程时，只返回直接孩子，并且翻页锚点正常。孙子线程不应该混进直接孩子列表。

数据流：测试插入两个孩子和一个孙子，建立父子边；按创建时间倒序分页列父线程的孩子；检查两页分别返回正确孩子。

调用关系：测试运行器调用它；它覆盖 list_threads_by_parent、spawn edge 和 keyset pagination（基于锚点的翻页）。

调用图：调用 5 个内部函数（from_string, new, init, test_thread_metadata, unique_temp_dir）；外部调用 2 个（from_timestamp, assert_eq!）。

async fn apply_rollout_items_restores_memory_mode_from_session_meta()

作用：验证从 rollout 的 SessionMeta 里读到 memory_mode 后，会写回数据库。这样重建索引不会丢记忆模式。

数据流：测试先插入线程，再构造带 memory_mode 的 SessionMeta 流水项，调用 apply_rollout_items；最后读取 memory_mode 确认已更新。

调用关系：测试运行器调用它；它覆盖 apply_rollout_items、extract_memory_mode 和 set_thread_memory_mode。

调用图：调用 5 个内部函数（from_string, new, init, test_thread_metadata, unique_temp_dir）；外部调用 2 个（assert_eq!, vec!）。

async fn apply_rollout_items_preserves_existing_git_branch_and_fills_missing_git_fields()

作用：验证应用 rollout 时会保留数据库里已有的 Git 分支，同时补齐缺失的 Git 提交号和远端地址。

数据流：测试先写入带 sqlite-branch 的线程，再应用带完整 Git 信息的 SessionMeta；最后确认分支仍是旧值，其他缺失字段被补上。

调用关系：测试运行器调用它；它覆盖 apply_rollout_items 对现有 Git 信息的保护。

调用图：调用 5 个内部函数（from_string, new, init, test_thread_metadata, unique_temp_dir）；外部调用 2 个（assert_eq!, vec!）。

async fn upsert_thread_preserves_existing_git_fields_atomically()

作用：验证普通 upsert 不会用较旧 rollout 里的 Git 字段覆盖数据库已有 Git 字段。atomically 表示这件事在一条数据库写入里完成，减少并发风险。

数据流：测试先插入一组 Git 信息，再用另一组 Git 信息 upsert 同一线程；最后确认数据库仍保留第一次的 Git 字段。

调用关系：测试运行器调用它；它保护 upsert_thread_with_creation_memory_mode 里的 COALESCE 规则。

调用图：调用 4 个内部函数（from_string, init, test_thread_metadata, unique_temp_dir）；外部调用 1 个（assert_eq!）。

async fn upsert_thread_preserves_existing_preview_when_incoming_preview_is_empty()

作用：验证传入资料没有预览时，不会清掉数据库已有预览。这样迁移或回填时不会把有用摘要抹掉。

数据流：测试先插入带 preview 的线程，再用 preview 为空的资料 upsert；最后读取线程确认 preview 仍在。

调用关系：测试运行器调用它；它覆盖 upsert 时 preview 的保留逻辑。

调用图：调用 4 个内部函数（from_string, init, test_thread_metadata, unique_temp_dir）；外部调用 1 个（assert_eq!）。

async fn set_thread_preview_if_empty_only_fills_blank_preview()

作用：验证 set_thread_preview_if_empty 只填空白预览，不接受纯空白，也不会覆盖已有预览。

数据流：测试创建无预览线程，先传空白字符串确认不更新，再传有效预览确认填入，最后再传新预览确认不会覆盖。

调用关系：测试运行器调用它；它专门保护 set_thread_preview_if_empty 的安全填空行为。

调用图：调用 4 个内部函数（from_string, init, test_thread_metadata, unique_temp_dir）；外部调用 2 个（assert!, assert_eq!）。

async fn update_thread_git_info_preserves_newer_non_git_metadata()

作用：验证单独更新 Git 信息时，不会覆盖标题、token、预览、更新时间等非 Git 新数据。

数据流：测试先插入线程，再直接模拟另一个写入更新非 Git 字段；调用 update_thread_git_info 后读取线程，确认非 Git 字段仍是新值，Git 字段已更新。

调用关系：测试运行器调用它；它覆盖 update_thread_git_info 的“只改指定 Git 字段”设计。

调用图：调用 4 个内部函数（from_string, init, test_thread_metadata, unique_temp_dir）；外部调用 4 个（from_timestamp, assert!, assert_eq!, query）。

async fn insert_thread_if_absent_preserves_existing_metadata()

作用：验证 insert_thread_if_absent 遇到已有线程时不会覆盖已有资料。它防止备用插入把更准确的数据冲掉。

数据流：测试先插入较新的线程资料，再用较旧或空的备用资料调用 insert_thread_if_absent；确认返回未插入，数据库仍是旧的准确资料。

调用关系：测试运行器调用它；它覆盖 insert_thread_if_absent 的 ON CONFLICT DO NOTHING 行为。

调用图：调用 4 个内部函数（from_string, init, test_thread_metadata, unique_temp_dir）；外部调用 3 个（from_timestamp, assert!, assert_eq!）。

async fn update_thread_git_info_can_clear_fields()

作用：验证 Git 信息更新接口可以把字段清空。因为有时仓库信息确实需要移除，而不是只改成新值。

数据流：测试先插入带 Git 字段的线程，再调用 update_thread_git_info 并把三个字段都设成 None；最后确认数据库字段都变为空。

调用关系：测试运行器调用它；它覆盖 update_thread_git_info 对“明确清空”和“不更新”的区分。

调用图：调用 4 个内部函数（from_string, init, test_thread_metadata, unique_temp_dir）；外部调用 2 个（assert!, assert_eq!）。

async fn touch_thread_updated_at_updates_only_updated_at()

作用：验证刷新更新时间只改时间，不动标题和预览等内容字段。

数据流：测试插入线程后调用 touch_thread_updated_at；再读取线程，确认 updated_at 变了，标题、第一条用户消息和预览仍符合原逻辑。

调用关系：测试运行器调用它；它覆盖 touch_thread_updated_at 和 metadata_preview 的间接效果。

调用图：调用 4 个内部函数（from_string, init, test_thread_metadata, unique_temp_dir）；外部调用 3 个（from_timestamp, assert!, assert_eq!）。

async fn thread_updated_at_uses_unique_epoch_millis_and_reads_legacy_seconds()

作用：验证更新时间使用毫秒并能给同一毫秒的热写入分配唯一值，同时还能读取旧版只有秒的时间。

数据流：测试插入两个相同毫秒时间的线程，确认第二个被加 1 毫秒；再插入更旧时间确认不被强行推新；最后模拟旧秒字段写入并确认读取时能转成毫秒。

调用关系：测试运行器调用它；它重点覆盖 allocate_thread_updated_at 和时间列兼容。

调用图：调用 4 个内部函数（from_string, init, test_thread_metadata, unique_temp_dir）；外部调用 4 个（from_timestamp_millis, assert_eq!, query, query_as）。

async fn apply_rollout_items_uses_override_updated_at_when_provided()

作用：验证应用 rollout 时如果传入更新时间覆盖值，就用这个值而不是文件修改时间。这样批处理可以指定准确时间。

数据流：测试构造一个 token 统计流水项和指定更新时间，调用 apply_rollout_items；最后确认 token 用量更新且 updated_at 等于覆盖值。

调用关系：测试运行器调用它；它覆盖 apply_rollout_items 的 updated_at_override 分支。

调用图：调用 5 个内部函数（from_string, new, init, test_thread_metadata, unique_temp_dir）；外部调用 3 个（from_timestamp, assert_eq!, vec!）。

async fn thread_spawn_edges_track_directional_status()

作用：验证父子线程关系的方向和状态过滤都正确。方向很重要，因为“父启动子”和“子属于父”不能反过来理解。

数据流：测试建立父子和孙子关系，先查打开状态，再关闭一条边，分别查询打开、关闭、全部后代；检查结果符合树结构和状态。

调用关系：测试运行器调用它；它覆盖 upsert_thread_spawn_edge、set_thread_spawn_edge_status 以及 children/descendants 查询。

调用图：调用 3 个内部函数（from_string, init, unique_temp_dir）；外部调用 1 个（assert_eq!）。

async fn thread_spawn_children_without_status_filter_lists_all_statuses()

作用：验证不传状态过滤时，会列出所有状态的直接子线程，包括未来可能新增的未知状态字符串。

数据流：测试插入 open、closed 和 future 三种状态的子边；调用 list_thread_spawn_children；确认三个孩子都返回。

调用关系：测试运行器调用它；它保护 list_thread_spawn_children 不因状态枚举变化而漏数据。

调用图：调用 3 个内部函数（from_string, init, unique_temp_dir）；外部调用 2 个（assert_eq!, query）。

fn as_str(self) -> &'static str

作用：把目标状态变成固定的英文小写字符串，方便存进数据库、写进接口返回，或和外部系统对齐。比如 Active 会变成 active。

数据流：进去的是一个 ThreadGoalStatus 状态值 → 它按状态逐一匹配到约定好的文字 → 出来的是一个不会临时分配内存的字符串常量，原状态不被改动。

调用关系：它是状态枚举对外表达自己的出口。其他保存、展示或传输目标状态的代码会用它把程序内部的状态翻译成外部能存、能读的文字。

fn is_active(self) -> bool

作用：判断这个目标是不是“正在进行”。调用者不用自己写比较代码，直接问它就行。

数据流：进去的是一个状态值 → 它检查这个值是否等于 Active → 出来的是 true 或 false，不改动任何数据。

调用关系：它通常会出现在需要决定“这个目标还能不能继续推进”的地方，是业务判断里的一个小开关。

fn is_terminal(self) -> bool

作用：判断这个目标是不是已经到了“终点状态”。这里的终点包括预算用完和完成，意思是后续一般不该再按普通进行中目标处理。

数据流：进去的是一个状态值 → 它用模式匹配检查是否是 BudgetLimited 或 Complete → 出来的是 true 或 false，状态本身不变。

调用关系：它把“哪些状态算结束”集中写在一个地方。调用者不用到处重复判断，也避免有人漏掉某个结束状态。

调用图：外部调用 1 个（matches!）。

fn try_from(value: &str) -> Result<Self>

作用：把数据库或外部传来的状态字符串，转换成程序内部可靠的 ThreadGoalStatus。遇到不认识的文字会报错，而不是随便猜。

数据流：进去的是一段字符串，比如 active 或 complete → 它查这段文字是否属于已知状态 → 成功时出来一个对应的状态值；失败时出来一个错误，说明遇到了未知状态。

调用关系：它是从外部文字进入内部状态的入口。ThreadGoal::try_from 会调用它，把数据库行里的 status 字段变成真正的状态枚举。

调用图：外部调用 1 个（anyhow!）。

fn try_from_row(row: &SqliteRow) -> Result<Self>

作用：把 SQLite 数据库返回的一行原始数据，按字段名取出来，装进 ThreadGoalRow 这个临时容器里。SQLite 是这里使用的本地数据库。

数据流：进去的是一行 SqliteRow 数据库结果 → 它用字段名读取 thread_id、goal_id、objective、status、预算、用量、时间等列 → 出来的是 ThreadGoalRow；如果某列缺失或类型不对，就返回错误。

调用关系：它被 thread_goal_from_row 调用，处在“数据库结果刚回来”的第一步。它只负责把表格行拆开，不负责把字符串状态和毫秒时间变成最终类型。

调用图：被 1 处调用（thread_goal_from_row）；外部调用 1 个（try_get）。

fn try_from(row: ThreadGoalRow) -> Result<Self>

作用：把数据库用的 ThreadGoalRow，转换成程序真正使用的 ThreadGoal。它会顺手检查线程编号、状态和时间是否合法。

数据流：进去的是一个 ThreadGoalRow，里面很多值还是数据库形式，比如线程编号是字符串、时间是毫秒数字 → 它把线程编号转成 ThreadId，把状态字符串转成 ThreadGoalStatus，把毫秒时间转成 UTC 时间 → 出来的是完整的 ThreadGoal；任何一步失败都会返回错误。

调用关系：它接在 ThreadGoalRow::try_from_row 后面，是“原始数据库行”到“可安全使用的目标对象”的最后一道关。它会把状态转换交给 ThreadGoalStatus::try_from，把时间转换交给 epoch_millis_to_datetime。

调用图：调用 1 个内部函数（try_from）；外部调用 2 个（try_from, epoch_millis_to_datetime）。

fn new(pool: Arc<SqlitePool>) -> Self

作用：创建一个 GoalStore，让后续代码可以通过它访问线程目标表。它把数据库连接池收起来，之后所有目标读写都用这同一套连接。

数据流：进去的是一个 SQLite 连接池 → 函数把它放进 GoalStore 结构里 → 出来的是一个可复制使用的 GoalStore，不会立刻读写数据库。

调用关系：它在运行时初始化阶段被 init_inner 调用，相当于系统启动时把“目标账本”的入口装配好，后面的读写函数都依赖这个入口里的连接池。

调用图：被 1 处调用（init_inner）。

async fn close(&self)

作用：关闭 GoalStore 持有的数据库连接池。系统结束或清理资源时会用它，避免连接一直占着。

数据流：进去的是当前 GoalStore → 它通知连接池关闭并等待关闭完成 → 没有返回业务数据，只改变连接池状态。

调用关系：它被更外层的 close 调用，是运行结束时收尾的一环；它不处理目标内容，只负责把底层数据库通道关掉。

调用图：被 1 处调用（close）。

async fn get_thread_goal(
        &self,
        thread_id: ThreadId,
    ) -> anyhow::Result<Option<crate::ThreadGoal>>

作用：读取某个线程当前保存的目标。如果这个线程没有目标，就明确返回空。

数据流：进去的是 thread_id → 它把线程编号转成字符串，到 thread_goals 表按 thread_id 查询一行 → 查到就转换成 ThreadGoal，查不到就返回 None。

调用关系：很多函数在修改后会调用它拿最新结果，比如 update_thread_goal、update_active_thread_goal_status 和 account_thread_goal_usage；它是整个文件里最基础的“查账”动作。

调用图：被 3 处调用（account_thread_goal_usage, update_active_thread_goal_status, update_thread_goal）；外部调用 2 个（to_string, query）。

async fn replace_thread_goal(
        &self,
        thread_id: ThreadId,
        objective: &str,
        status: crate::ThreadGoalStatus,
        token_budget: Option<i64>,
    ) -> anyhow::Result<c

作用：强制替换某个线程的目标，不管之前有没有目标。它会把用量清零，并给新目标生成新的编号。

数据流：进去的是线程编号、目标文字、初始状态和可选 token 预算 → 它生成 goal_id 和当前时间，先检查初始状态是否已经碰到预算上限，再插入或覆盖数据库记录 → 出来的是数据库返回的新 ThreadGoal。

调用关系：它会调用 status_after_budget_limit 判断是否应立刻变成预算受限，再用 thread_goal_from_row 把数据库行转成业务对象；测试里大量使用它来搭建不同状态的目标。

调用图：调用 2 个内部函数（status_after_budget_limit, thread_goal_from_row）；外部调用 5 个（now, new_v4, as_str, to_string, query）。

async fn insert_thread_goal(
        &self,
        thread_id: ThreadId,
        objective: &str,
        status: crate::ThreadGoalStatus,
        token_budget: Option<i64>,
    ) -> anyhow::Result<Op

作用：尝试插入一个新目标，但不会随便覆盖正在进行的旧目标。只有原目标已经 complete（完成）时，才允许换成新目标。

数据流：进去的是线程编号、目标文字、状态和预算 → 它生成新 goal_id 和时间，计算预算后的状态，然后尝试插入；如果已有未完成目标，数据库不会返回新行 → 出来是 Some(目标) 或 None。

调用关系：它和 replace_thread_goal 很像，也会调用 status_after_budget_limit；区别是它更保守，适合“不想打断现有目标”的创建场景。

调用图：调用 1 个内部函数（status_after_budget_limit）；外部调用 5 个（now, new_v4, as_str, to_string, query）。

async fn update_thread_goal(
        &self,
        thread_id: ThreadId,
        update: GoalUpdate,
    ) -> anyhow::Result<Option<crate::ThreadGoal>>

作用：部分更新一个目标，比如只改目标文字、只改状态、只改预算，或几个一起改。它还支持 expected_goal_id，防止拿着旧目标编号的人误改新目标。

数据流：进去的是 thread_id 和 GoalUpdate → 它根据哪些字段有值选择不同 SQL 更新，只改指定字段，必要时按预算规则修正状态；如果 goal_id 不匹配或没有可改记录就返回 None → 成功后再读出最新目标返回。

调用关系：它在业务流程中用于改目标内容和状态；当没有实际字段要改时会退回调用 get_thread_goal 做安全读取，修改成功后也调用 get_thread_goal 拿最终状态。

调用图：调用 1 个内部函数（get_thread_goal）；外部调用 3 个（now, to_string, query）。

async fn pause_active_thread_goal(
        &self,
        thread_id: ThreadId,
    ) -> anyhow::Result<Option<crate::ThreadGoal>>

作用：把当前还在 active（活跃运行）的目标暂停。它不会改已经完成、阻塞或其他终止类状态的目标。

数据流：进去的是 thread_id → 它把目标状态请求包装成 Paused，交给 update_active_thread_goal_status → 出来是更新后的目标，或没有可暂停目标时返回 None。

调用关系：这是一个方便入口，真正的数据库更新交给 update_active_thread_goal_status；调用方不用自己写状态判断。

调用图：调用 1 个内部函数（update_active_thread_goal_status）。

async fn usage_limit_active_thread_goal(
        &self,
        thread_id: ThreadId,
    ) -> anyhow::Result<Option<crate::ThreadGoal>>

作用：把活跃目标标记为 usage_limited（外部使用限制导致停止）。如果目标已经是 budget_limited，也允许升级成 usage_limited，让真正的停止原因更清楚。

数据流：进去的是 thread_id → 它指定目标状态为 UsageLimited，交给 update_active_thread_goal_status → 出来是更新后的目标，或没有符合条件的目标时返回 None。

调用关系：它和 pause_active_thread_goal 共用同一个底层函数；上层遇到全局用量限制时会用这个入口记录停止原因。

调用图：调用 1 个内部函数（update_active_thread_goal_status）。

async fn update_active_thread_goal_status(
        &self,
        thread_id: ThreadId,
        status: crate::ThreadGoalStatus,
    ) -> anyhow::Result<Option<crate::ThreadGoal>>

作用：只在目标还适合被停止时更新状态，避免把已经完成的目标又改掉。它是暂停和使用限制两个公开方法背后的共同实现。

数据流：进去的是 thread_id 和目标新状态 → 它更新数据库里 active 的目标；如果新状态是 usage_limited，也允许从 budget_limited 改过去 → 更新成功后读回目标，失败则返回 None。

调用关系：它被 pause_active_thread_goal 和 usage_limit_active_thread_goal 调用；内部用 get_thread_goal 返回修改后的完整目标。

调用图：调用 1 个内部函数（get_thread_goal）；被 2 处调用（pause_active_thread_goal, usage_limit_active_thread_goal）；外部调用 4 个（now, as_str, to_string, query）。

async fn delete_thread_goal(
        &self,
        thread_id: ThreadId,
    ) -> anyhow::Result<Option<crate::ThreadGoal>>

作用：删除某个线程的目标，并把被删掉的目标返回给调用方。这样调用方既能清理，也能知道刚才清掉了什么。

数据流：进去的是 thread_id → 它从 thread_goals 表删除对应行，并用 RETURNING 取回被删除的数据 → 出来是 Some(旧目标) 或 None。

调用关系：它是目标生命周期的清理动作；测试也验证了删除线程时目标会一起消失，说明它配合更外层线程删除流程使用。

调用图：外部调用 2 个（to_string, query）。

async fn account_thread_goal_usage(
        &self,
        thread_id: ThreadId,
        time_delta_seconds: i64,
        token_delta: i64,
        mode: GoalAccountingMode,
        expected_goal_id: O

作用：给目标累计本次消耗的时间和 token，并在超过预算时自动把目标标记为 budget_limited。它是“记账”核心，保证用量不会因为并发或状态变化而算丢。

数据流：进去的是 thread_id、时间增量、token 增量、记账模式和可选 expected_goal_id → 它把负数增量当 0，按模式筛选哪些状态能记账，然后用一条数据库更新同时累加时间和 token，并检查预算 → 出来是 Updated(新目标)，或 Unchanged(当前目标)。

调用关系：它会在无需更新或更新没命中时调用 get_thread_goal，看当前到底是什么状态；更新命中后调用 thread_goal_from_row 转成 ThreadGoal。它是运行过程中每次模型或任务消耗资源后会走的关键路径。

调用图：调用 2 个内部函数（get_thread_goal, thread_goal_from_row）；外部调用 5 个（new, now, to_string, Unchanged, Updated）。

fn thread_goal_from_row(row: &sqlx::sqlite::SqliteRow) -> anyhow::Result<crate::ThreadGoal>

作用：把数据库查出来的一行目标记录，转换成代码里好用的 ThreadGoal 对象。它像把表格里的一行整理成一张标准任务卡。

数据流：进去的是 SQLite 的一行数据 → 先按 ThreadGoalRow 读取字段，再转换成 ThreadGoal → 出来是业务对象，字段不合法时返回错误。

调用关系：它被 replace_thread_goal 和 account_thread_goal_usage 调用；get_thread_goal、delete_thread_goal 等也通过同样模式使用它，保证数据库格式到业务格式的转换集中一致。

调用图：调用 1 个内部函数（try_from_row）；被 2 处调用（account_thread_goal_usage, replace_thread_goal）。

fn status_after_budget_limit(
    status: crate::ThreadGoalStatus,
    tokens_used: i64,
    token_budget: Option<i64>,
) -> crate::ThreadGoalStatus

作用：根据 token 预算判断初始或更新后的状态是否应该立刻变成 budget_limited。它防止出现“明明预算已经用完，却还显示 active”的矛盾。

数据流：进去的是目标状态、已用 token 和可选预算 → 如果状态是 Active 且已用量大于等于预算，就返回 BudgetLimited；否则原样返回状态。

调用关系：它被 replace_thread_goal 和 insert_thread_goal 调用，用在目标刚创建或替换时的第一道预算检查。

调用图：被 2 处调用（insert_thread_goal, replace_thread_goal）。

async fn test_runtime() -> std::sync::Arc<StateRuntime>

作用：为测试创建一个临时 StateRuntime，也就是带临时数据库和目录的运行环境。这样每个测试都能在干净环境里跑。

数据流：进去没有业务参数 → 它生成唯一临时目录并初始化 StateRuntime → 出来是可共享的 runtime，初始化失败会让测试报错。

调用关系：几乎所有测试都会先调用它；它调用 init 和 unique_temp_dir，把真实运行环境缩小成测试专用环境。

调用图：调用 2 个内部函数（init, unique_temp_dir）。

fn test_thread_id() -> ThreadId

作用：提供一个固定的测试线程编号。固定编号让断言更稳定，也方便所有测试复用。

数据流：进去没有参数 → 它把固定 UUID 字符串解析成 ThreadId → 出来是合法线程 ID。

调用关系：多数测试用它来指定要操作的线程；它调用 from_string，避免每个测试重复写解析代码。

调用图：调用 1 个内部函数（from_string）。

async fn upsert_test_thread(runtime: &StateRuntime, thread_id: ThreadId)

作用：在测试数据库里先放入一个线程记录。因为目标是挂在线程下面的，没有线程就无法可靠测试目标。

数据流：进去的是 runtime 和 thread_id → 它构造测试线程元数据，并调用 upsert_thread 写入或更新线程 → 没有返回业务数据，失败会让测试报错。

调用关系：几乎每个目标测试在操作 GoalStore 前都会调用它；它连接了测试辅助数据和真实的线程写入接口。

调用图：调用 2 个内部函数（codex_home, test_thread_metadata）；外部调用 1 个（upsert_thread）。

async fn replace_update_and_get_thread_goal()

作用：测试目标的基本生命周期：创建、读取、更新、替换、删除。它证明最常用的一整套操作能连起来正常工作。

数据流：进去是测试框架启动的空环境 → 它创建 runtime 和线程，写入目标，读回比对，再更新、替换和删除 → 最后通过断言确认每一步结果正确。

调用关系：它使用 test_runtime、test_thread_id 和 upsert_test_thread 搭环境，然后调用 GoalStore 的主要公开方法，是这个文件最完整的冒烟测试。

调用图：外部调用 4 个（assert_eq!, test_runtime, test_thread_id, upsert_test_thread）。

async fn replace_thread_goal_applies_budget_limit_immediately()

作用：测试替换目标时，如果预算一开始就是 0，目标会立刻变成预算受限。这样不会出现刚创建就超预算却仍然 active 的错误。

数据流：进去是干净测试环境 → 它创建一个 active 且预算为 0 的目标 → 断言返回状态是 BudgetLimited，用量仍为 0。

调用关系：它间接验证 replace_thread_goal 会使用 status_after_budget_limit；测试环境由通用测试辅助函数准备。

调用图：外部调用 4 个（assert_eq!, test_runtime, test_thread_id, upsert_test_thread）。

async fn insert_thread_goal_does_not_replace_existing_goal()

作用：测试保守插入不会覆盖已有未完成目标。这个规则防止新请求把正在做的目标悄悄换掉。

数据流：进去是测试环境和线程 → 先插入一个目标，再尝试插入第二个目标 → 第二次返回 None，数据库里仍是第一个目标。

调用关系：它主要验证 insert_thread_goal 的冲突处理逻辑；使用 get_thread_goal 确认最终保存内容没有变。

调用图：外部调用 4 个（assert_eq!, test_runtime, test_thread_id, upsert_test_thread）。

async fn insert_thread_goal_applies_budget_limit_immediately()

作用：测试插入新目标时也会立即检查预算上限。它和替换目标的预算规则保持一致。

数据流：进去是干净测试环境 → 插入一个 active 且预算为 0 的目标 → 断言目标状态立刻是 BudgetLimited，时间和 token 没有被错误增加。

调用关系：它验证 insert_thread_goal 调用 status_after_budget_limit 的效果，补齐 replace 场景之外的创建路径。

调用图：外部调用 4 个（assert_eq!, test_runtime, test_thread_id, upsert_test_thread）。

async fn update_thread_goal_ignores_replaced_goal_version()

作用：测试带旧 goal_id 的更新不会误改新目标。它保护并发场景下“旧页面、旧请求”不会覆盖当前目标。

数据流：进去是测试环境 → 先创建旧目标，再替换成新目标；随后用旧 goal_id 尝试完成目标 → 返回 None，当前目标不变；再用新 goal_id 更新才成功。

调用关系：它验证 update_thread_goal 的 expected_goal_id 保护机制，是防止版本错乱的重要测试。

调用图：外部调用 4 个（assert_eq!, test_runtime, test_thread_id, upsert_test_thread）。

async fn usage_accounting_ignores_replaced_goal_version()

作用：测试用量记账时也会尊重 goal_id 版本。旧目标的用量不能记到新目标身上。

数据流：进去是测试环境 → 创建旧目标并替换成新目标，然后用旧 goal_id 尝试增加用量 → 返回 Unchanged，当前新目标的用量仍为 0。

调用关系：它验证 account_thread_goal_usage 的 expected_goal_id 参数；通过 panic 和断言确保结果必须是未更新分支。

调用图：外部调用 6 个（assert_eq!, assert_ne!, panic!, test_runtime, test_thread_id, upsert_test_thread）。

async fn update_thread_goal_objective_preserves_usage_and_created_at()

作用：测试修改目标文字、状态和预算时，不会丢掉已有用量和创建时间。也就是说改说明不等于重开一张任务单。

数据流：进去是测试环境 → 创建目标并先记一笔用量，再更新目标文字、暂停状态和预算 → 断言 token、时间和 created_at 都沿用之前的值。

调用关系：它同时覆盖 account_thread_goal_usage 和 update_thread_goal 的配合，确保更新字段是“局部修改”而不是整体重置。

调用图：外部调用 5 个（assert_eq!, panic!, test_runtime, test_thread_id, upsert_test_thread）。

async fn concurrent_partial_updates_preserve_independent_fields()

作用：测试两个并发的局部更新不会互相覆盖。一个改状态，一个改预算，最后两个改动都应该留下。

数据流：进去是测试环境 → 创建目标，然后用 tokio::join! 同时发起状态更新和预算更新 → 最后读回目标，断言状态和预算都正确。

调用关系：它验证 update_thread_goal 的 SQL 只改指定字段；join! 用来模拟两个操作同时发生。

调用图：外部调用 5 个（assert_eq!, test_runtime, test_thread_id, upsert_test_thread, join!）。

async fn pause_active_thread_goal_does_not_clobber_terminal_status()

作用：测试暂停操作只影响 active 目标，不会把已经完成的目标改回暂停。它防止终态被晚到的暂停请求破坏。

数据流：进去是测试环境 → 创建 active 目标并暂停成功；再把目标改成 Complete，然后再次请求暂停 → 第二次返回 None，目标仍保持 Complete。

调用关系：它验证 pause_active_thread_goal 通过 update_active_thread_goal_status 做状态筛选，不会无脑覆盖所有状态。

调用图：外部调用 4 个（assert_eq!, test_runtime, test_thread_id, upsert_test_thread）。

async fn usage_limit_active_thread_goal_updates_active_or_budget_limited_goals()

作用：测试 usage_limited 状态能从 active 设置，也能从 budget_limited 设置。这样当系统发现外部用量限制时，可以覆盖预算限制，显示更准确的停止原因。

数据流：进去是测试环境 → 先让 active 目标变成 UsageLimited，并确认重复设置不会再改；再创建 BudgetLimited 目标并设置 UsageLimited → 断言两种允许路径都正确。

调用关系：它验证 usage_limit_active_thread_goal 和 update_active_thread_goal_status 里的特殊规则：usage_limited 可以接管 budget_limited。

调用图：外部调用 4 个（assert_eq!, test_runtime, test_thread_id, upsert_test_thread）。

async fn usage_accounting_updates_active_goals_and_accounts_budget_limited_in_flight_usage()

作用：测试活跃目标会正常累计用量，达到预算后变成 budget_limited，并且还能补记已经在路上的最后一段用量。

数据流：进去是测试环境 → 创建预算 20 的 active 目标；先记 5 token，再记 15 token 触顶；最后再记一笔进行中的用量 → 断言累计 token 和时间都正确。

调用关系：它重点验证 account_thread_goal_usage 在 ActiveOnly 模式下允许 active 和 budget_limited 的连续记账，避免刚触顶时漏掉未结算消耗。

调用图：外部调用 5 个（assert_eq!, panic!, test_runtime, test_thread_id, upsert_test_thread）。

async fn active_status_only_usage_accounting_does_not_update_budget_limited_goals()

作用：测试 ActiveStatusOnly 模式真的只更新 active 状态，不会更新 budget_limited。这个模式适合严格只给正在运行的目标记账。

数据流：进去是测试环境 → 创建一个 BudgetLimited 目标，然后尝试按 ActiveStatusOnly 记账 → 返回 Unchanged，用量保持 0。

调用关系：它验证 account_thread_goal_usage 的模式分支，和 ActiveOnly 模式形成对照。

调用图：外部调用 5 个（assert_eq!, panic!, test_runtime, test_thread_id, upsert_test_thread）。

async fn stopped_usage_accounting_promotes_paused_goal_over_budget()

作用：测试暂停后的目标如果补记最后用量并超过预算，也会变成 budget_limited。暂停不应该掩盖已经超预算的事实。

数据流：进去是测试环境 → 创建 active 目标并暂停，再用 ActiveOrStopped 模式记 25 token，超过预算 20 → 断言状态变为 BudgetLimited，并记录用量。

调用关系：它验证 account_thread_goal_usage 在 ActiveOrStopped 模式下，不只给停止状态补账，还会重新检查预算。

调用图：外部调用 5 个（assert_eq!, panic!, test_runtime, test_thread_id, upsert_test_thread）。

async fn budget_updates_immediately_stop_active_goals_already_over_budget()

作用：测试降低预算时，如果目标已用量超过新预算，会立刻变成 budget_limited。否则目标会继续显示 active，误导上层继续执行。

数据流：进去是测试环境 → 创建预算 100 的目标并消耗 50 token，然后把预算降到 40 → 断言状态马上变成 BudgetLimited，已用 token 保持 50。

调用关系：它验证 update_thread_goal 在只改预算时也会检查当前用量，而不是等下一次记账才发现超限。

调用图：外部调用 4 个（assert_eq!, test_runtime, test_thread_id, upsert_test_thread）。

async fn activating_goal_already_over_budget_keeps_it_budget_limited()

作用：测试一个已经超预算的目标，即使请求把它改回 active，也会保持 budget_limited。它防止人为状态更新绕过预算限制。

数据流：进去是测试环境 → 先让目标超过预算进入 BudgetLimited，再请求改状态为 Active 并改文字 → 断言文字更新成功，但状态仍是 BudgetLimited。

调用关系：它验证 update_thread_goal 的状态计算逻辑：预算规则优先于“重新激活”的请求。

调用图：外部调用 4 个（assert_eq!, test_runtime, test_thread_id, upsert_test_thread）。

async fn pausing_budget_limited_goal_preserves_terminal_status()

作用：测试对 budget_limited 目标发暂停请求，不会把它改成 Paused。预算受限这个停止原因需要保留下来。

数据流：进去是测试环境 → 让目标先超过预算，再用 update_thread_goal 请求 Paused → 断言状态仍是 BudgetLimited，预算和用量不变。

调用关系：它验证 update_thread_goal 中“预算受限遇到暂停或阻塞时保留原状态”的规则。

调用图：外部调用 4 个（assert_eq!, test_runtime, test_thread_id, upsert_test_thread）。

async fn blocking_budget_limited_goal_preserves_terminal_status()

作用：测试对 budget_limited 目标发阻塞请求，也不会覆盖预算受限状态。这样最终状态不会被较晚的普通停止原因冲掉。

数据流：进去是测试环境 → 让目标记账超过预算，得到 BudgetLimited；再请求改成 Blocked → 返回目标仍保持预算受限，只更新时间可能变化。

调用关系：它和 pausing_budget_limited_goal_preserves_terminal_status 配套，验证 update_thread_goal 对 Paused 和 Blocked 两种状态都保护预算受限。

调用图：外部调用 5 个（assert_eq!, panic!, test_runtime, test_thread_id, upsert_test_thread）。

async fn usage_accounting_can_finalize_completed_goal_for_completing_turn()

作用：测试已经 complete 的目标，在普通 active-only 记账下不会变，但在“完成这一轮的最后结算”模式下可以补记用量。

数据流：进去是测试环境 → 创建 Complete 目标；先用 ActiveOnly 记账，确认不变；再用 ActiveOrComplete 记账 → 断言完成目标保留 Complete 状态，同时累计时间和 token。

调用关系：它验证 account_thread_goal_usage 的 ActiveOrComplete 模式，用来处理目标刚完成但本轮消耗还没入账的情况。

调用图：外部调用 5 个（assert_eq!, panic!, test_runtime, test_thread_id, upsert_test_thread）。

async fn usage_accounting_can_finalize_stopped_goal_for_in_flight_turn()

作用：测试已经暂停的目标，在普通记账模式下不变，但可以用 ActiveOrStopped 模式补记正在进行中的最后一笔消耗。

数据流：进去是测试环境 → 创建 active 目标并暂停；ActiveOnly 记账返回不变；ActiveOrStopped 记账成功增加时间和 token，状态仍是 Paused。

调用关系：它验证 account_thread_goal_usage 对停止状态的补账能力，适合请求已经开始、后来目标被暂停的场景。

调用图：外部调用 5 个（assert_eq!, panic!, test_runtime, test_thread_id, upsert_test_thread）。

async fn usage_accounting_adds_concurrent_token_deltas()

作用：测试两个并发记账请求会相加，而不是后来的覆盖前面的。这个测试保护资源统计的准确性。

数据流：进去是测试环境 → 创建目标后同时记 40 token/4 秒和 60 token/6 秒 → 最后读回目标，断言总数是 100 token 和 10 秒。

调用关系：它验证 account_thread_goal_usage 使用数据库原子累加的效果；join! 模拟并发发生。

调用图：外部调用 5 个（assert_eq!, test_runtime, test_thread_id, upsert_test_thread, join!）。

async fn deleting_thread_deletes_goal()

作用：测试删除线程时，挂在这个线程上的目标也会被清掉。这样不会留下没有主人的目标记录。

数据流：进去是测试环境 → 创建线程和目标，然后调用 delete_thread 删除线程 → 再查询目标，断言已经不存在。

调用关系：它验证 GoalStore 和更外层线程删除流程的联动，说明目标表有随线程清理的约束或逻辑。

调用图：外部调用 4 个（assert_eq!, test_runtime, test_thread_id, upsert_test_thread）。

fn new(pool: Arc<SqlitePool>, state_pool: Arc<SqlitePool>) -> Self

作用：创建一个 MemoryStore，把两个 SQLite 数据库连接池装进去。一个池放记忆和任务，另一个池读线程信息。

数据流：输入两个数据库连接池 → 保存到结构体字段里 → 返回一个可以操作记忆数据的 MemoryStore。

调用关系：运行时初始化流程 init_inner 会调用它，把数据库连接交给记忆子系统，后面的所有记忆读写都从这个对象出发。

调用图：被 1 处调用（init_inner）。

async fn close(&self)

作用：关闭记忆数据库连接池，让后台连接干净退出。

数据流：输入当前 MemoryStore → 调用连接池的关闭动作 → 数据库连接停止接受新操作。

调用关系：整体运行时关闭时由 close 调用，属于收尾动作，避免资源一直占着。

调用图：被 1 处调用（close）。

async fn clear_memory_data(&self) -> anyhow::Result<()>

作用：清空所有已经保存的记忆结果和记忆任务记录。常用于重置记忆系统。

数据流：输入当前存储对象 → 把内部数据库连接交给清理函数 → 删除 stage1_outputs 和记忆相关 jobs 后返回成功或错误。

调用关系：它只是公开入口，真正删除工作交给 clear_memory_data_in_pool，测试包装函数也会通过它验证清理行为。

调用图：调用 1 个内部函数（clear_memory_data_in_pool）。

async fn record_stage1_output_usage(
        &self,
        thread_ids: &[ThreadId],
    ) -> anyhow::Result<usize>

作用：记录某些线程的小记忆被用过一次。这样后面挑选全局记忆时，会更偏向常用、最近用过的内容。

数据流：输入一组线程 ID → 如果为空直接返回 0；否则开启事务，逐个把对应记忆的使用次数加一、最后使用时间改成现在 → 返回实际更新了多少行。

调用关系：当系统引用了某些 stage-1 记忆后会调用它；它不再分发给别的本文件函数，只直接更新数据库。

调用图：外部调用 3 个（now, is_empty, query）。

async fn stage1_source_needs_update(
        &self,
        thread_id: ThreadId,
        source_updated_at: i64,
    ) -> anyhow::Result<bool>

作用：判断某个线程的第一阶段记忆是不是已经够新，不需要再提取。

数据流：输入线程 ID 和源线程更新时间 → 先看已有输出是否比源更新，再看任务成功水位是否覆盖这个时间 → 返回是否还需要重新跑。

调用关系：claim_stage1_jobs_for_startup 在启动扫描时会先问它，避免给已经处理过的线程重复派工。

调用图：被 1 处调用（claim_stage1_jobs_for_startup）；外部调用 3 个（as_str, to_string, query）。

async fn claim_stage1_jobs_for_startup(
        &self,
        current_thread_id: ThreadId,
        params: Stage1StartupClaimParams<'_>,
    ) -> anyhow::Result<Vec<Stage1JobClaim>>

作用：启动时批量找出适合补做记忆提取的旧线程，并尝试领取这些任务。

数据流：输入当前线程 ID 和扫描参数 → 在状态库里按来源、年龄、空闲时间、记忆模式过滤线程 → 再到记忆库确认是否过期 → 对合格线程调用领取函数 → 返回成功领取的任务列表。

调用关系：它串起线程筛选、过期判断和任务领取；内部使用 push_thread_filters、stage1_source_needs_update 和 try_claim_stage1_job。

调用图：调用 3 个内部函数（stage1_source_needs_update, try_claim_stage1_job, push_thread_filters）；外部调用 7 个（days, hours, new, now, new, try_from, as_str）。

async fn delete_thread_memory(&self, thread_id: ThreadId) -> anyhow::Result<()>

作用：删除某个线程对应的记忆数据和第一阶段任务记录。若这条记忆曾进入全局记忆，还会提醒第二阶段重新合并。

数据流：输入线程 ID → 查它是否被上次 phase-2 选中 → 删除 stage-1 输出和任务 → 如果删掉的是已选中内容，就加入全局合并任务 → 提交事务。

调用关系：线程删除流程会用到它；当需要重新合并时，它把后续工作交给 enqueue_global_consolidation_with_executor。

调用图：调用 1 个内部函数（enqueue_global_consolidation_with_executor）；外部调用 4 个（now, as_str, to_string, query）。

async fn list_stage1_outputs_for_global(
        &self,
        n: usize,
    ) -> anyhow::Result<Vec<Stage1Output>>

作用：列出最近的、非空的第一阶段记忆，给全局合并阶段使用。

数据流：输入最多要多少条 → 从记忆库按更新时间倒序取非空输出 → 每条再确认线程仍启用记忆并补齐线程信息 → 返回可见的 Stage1Output 列表。

调用关系：phase-2 准备输入时会调用它；它把每行转换工作交给 stage1_output_from_row_if_thread_enabled。

调用图：调用 1 个内部函数（stage1_output_from_row_if_thread_enabled）；外部调用 2 个（new, query）。

async fn prune_stage1_outputs_for_retention(
        &self,
        max_unused_days: i64,
        limit: usize,
    ) -> anyhow::Result<usize>

作用：清理太久没用、又没有被最新全局记忆选中的小记忆，防止数据库无限长大。

数据流：输入最大未使用天数和本次最多删除数 → 算出过期时间线 → 删除过期且未选中的输出，按最旧优先、数量受限 → 返回删除条数。

调用关系：保留策略或后台清理会调用它；它只清理输出，不动 stage-1 任务水位，避免破坏“已经处理过”的记录。

调用图：外部调用 3 个（days, now, query）。

async fn get_phase2_input_selection(
        &self,
        n: usize,
        max_unused_days: i64,
    ) -> anyhow::Result<Vec<Stage1Output>>

作用：为第二阶段挑选当前最应该参与全局合并的一批小记忆。

数据流：输入数量上限和最大未使用天数 → 分页挑选非空、仍新鲜的输出，排序优先常用和最近用过 → 排除禁用/污染线程 → 重新读取完整行并转换成输出对象 → 按线程 ID 稳定排序后返回。

调用关系：phase-2 真正生成输入文件前会调用它；内部会调用 enabled_thread_metadata 和 stage1_output_from_row_if_thread_enabled 确保只用仍有效的线程。

调用图：调用 3 个内部函数（try_from, enabled_thread_metadata, stage1_output_from_row_if_thread_enabled）；外部调用 6 个（days, now, new, with_capacity, try_from, query）。

async fn stage1_output_from_row_if_thread_enabled(
        &self,
        row: &sqlx::sqlite::SqliteRow,
    ) -> anyhow::Result<Option<Stage1Output>>

作用：把数据库里一行 stage-1 输出变成程序里的输出对象，但只接受记忆仍启用的线程。

数据流：输入一行数据库结果 → 取出线程 ID → 查询线程是否存在且 memory_mode 为 enabled → 若有效则合并线程信息和记忆内容 → 返回 Some，否则返回 None。

调用关系：list_stage1_outputs_for_global 和 get_phase2_input_selection 都靠它过滤掉不可再用的线程。

调用图：调用 3 个内部函数（try_from, enabled_thread_metadata, stage1_output_from_row_and_thread）；被 2 处调用（get_phase2_input_selection, list_stage1_outputs_for_global）；外部调用 1 个（try_get）。

async fn enabled_thread_metadata(
        &self,
        thread_id: ThreadId,
    ) -> anyhow::Result<Option<ThreadMetadata>>

作用：读取某个线程的元信息，并确认它的记忆模式仍是启用状态。

数据流：输入线程 ID → 到状态数据库查对应线程且 memory_mode = enabled → 把数据库行转成 ThreadMetadata → 没找到就返回空。

调用关系：它是记忆输出可见性的守门员，被 get_phase2_input_selection 和 stage1_output_from_row_if_thread_enabled 调用。

调用图：被 2 处调用（get_phase2_input_selection, stage1_output_from_row_if_thread_enabled）；外部调用 2 个（to_string, query）。

async fn mark_thread_memory_mode_polluted(
        &self,
        thread_id: ThreadId,
    ) -> anyhow::Result<bool>

作用：把一个线程标记成“污染”，意思是它的记忆不该再进入全局记忆。若它已经参与过全局记忆，还会触发重新合并来忘掉它。

数据流：输入线程 ID → 查它的 stage-1 输出是否被 phase-2 选中过 → 在状态库把线程 memory_mode 改为 polluted → 如曾被选中则排队全局合并 → 返回这次是否真的发生了状态变化。

调用关系：外部发现线程记忆不可信时会调用它；需要重新合并时通过 enqueue_global_consolidation 继续后续工作。

调用图：调用 1 个内部函数（enqueue_global_consolidation）；外部调用 4 个（now, as_str, to_string, query）。

async fn try_claim_stage1_job(
        &self,
        thread_id: ThreadId,
        worker_id: ThreadId,
        source_updated_at: i64,
        lease_seconds: i64,
        max_running_jobs: usize,

作用：尝试领取某个线程的第一阶段记忆提取任务。它像发工牌一样，成功时给调用者一个 ownership token。

数据流：输入线程、工作者、源更新时间、租约时长和并发上限 → 在事务里检查是否已最新、是否已有活跃任务、是否还在重试冷却、是否超过并发上限 → 成功则写入 running 任务和新票据 → 返回领取成功或跳过原因。

调用关系：claim_stage1_jobs_for_startup 会批量调用它；后续成功或失败函数必须带着这里发出的 token 才能结算任务。

调用图：被 1 处调用（claim_stage1_jobs_for_startup）；外部调用 5 个（now, new_v4, as_str, to_string, query）。

async fn mark_stage1_job_succeeded(
        &self,
        thread_id: ThreadId,
        ownership_token: &str,
        source_updated_at: i64,
        raw_memory: &str,
        rollout_summary: &str,

作用：把已领取的 stage-1 任务标记为成功，并保存提取出来的记忆内容。

数据流：输入线程 ID、领取票据、源更新时间、原始记忆、摘要和可选 slug → 只更新匹配 token 的 running 任务 → 插入或覆盖较新的 stage-1 输出 → 排队全局合并 → 返回是否成功结算。

调用关系：stage-1 工作器完成提取后调用它；它内部通过 enqueue_global_consolidation_with_executor 通知 phase-2 有新材料。

调用图：调用 1 个内部函数（enqueue_global_consolidation_with_executor）；外部调用 4 个（now, as_str, to_string, query）。

async fn mark_stage1_job_succeeded_no_output(
        &self,
        thread_id: ThreadId,
        ownership_token: &str,
    ) -> anyhow::Result<bool>

作用：把 stage-1 任务标记为成功，但表示这次没有提取到可保存的记忆。

数据流：输入线程 ID 和领取票据 → 只结算匹配 token 的 running 任务 → 读取本次处理的水位 → 删除旧输出 → 如果真的删掉了旧输出，就排队全局合并 → 返回是否成功结算。

调用关系：stage-1 工作器发现没有有效记忆时用它；若删除旧记忆，会通过 enqueue_global_consolidation_with_executor 让 phase-2 忘掉旧内容。

调用图：调用 1 个内部函数（enqueue_global_consolidation_with_executor）；外部调用 4 个（now, as_str, to_string, query）。

async fn mark_stage1_job_failed(
        &self,
        thread_id: ThreadId,
        ownership_token: &str,
        failure_reason: &str,
        retry_delay_seconds: i64,
    ) -> anyhow::Result<bool

作用：把已领取的 stage-1 任务标记为失败，并安排下次重试时间。

数据流：输入线程 ID、领取票据、失败原因和延迟秒数 → 只更新匹配 token 的 running 任务 → 状态改 error、清租约、记录错误、重试次数减一、设置 retry_at → 返回是否更新成功。

调用关系：stage-1 工作器出错时调用；之后 try_claim_stage1_job 会根据 retry_at 和 retry_remaining 决定能不能再领。

调用图：外部调用 4 个（now, as_str, to_string, query）。

async fn enqueue_global_consolidation(&self, input_watermark: i64) -> anyhow::Result<()>

作用：排队一次全局记忆合并任务，表示 phase-2 应该看看是否有新变化。

数据流：输入一个输入水位 → 调用通用入队函数写入或更新全局任务行 → 返回成功或错误。

调用关系：mark_thread_memory_mode_polluted 等公开流程会用它；实际数据库 upsert 交给 enqueue_global_consolidation_with_executor。

调用图：调用 1 个内部函数（enqueue_global_consolidation_with_executor）；被 1 处调用（mark_thread_memory_mode_polluted）。

async fn try_claim_global_phase2_job(
        &self,
        worker_id: ThreadId,
        lease_seconds: i64,
    ) -> anyhow::Result<Phase2JobClaimOutcome>

作用：尝试领取全局 phase-2 合并锁。因为全局记忆只能有一个合并者，所以它控制谁能跑。

数据流：输入工作者 ID 和租约时长 → 读取或创建唯一的 global 任务行 → 检查重试冷却、活跃租约和成功后的冷却期 → 成功则写入 running、票据和租约 → 返回领取成功及水位，或跳过原因。

调用关系：phase-2 工作器开始合并前调用它；成功拿到 token 后，后续心跳、成功、失败都要用这个 token。

调用图：外部调用 5 个（now, new_v4, as_str, to_string, query）。

async fn heartbeat_global_phase2_job(
        &self,
        ownership_token: &str,
        lease_seconds: i64,
    ) -> anyhow::Result<bool>

作用：给正在运行的全局合并任务续租，表示工作器还活着。

数据流：输入 ownership token 和租约秒数 → 计算新的过期时间 → 只更新匹配 token 的 running global 任务 → 返回是否续租成功。

调用关系：长时间运行的 phase-2 工作器会周期性调用它，防止别人误以为租约过期后接管。

调用图：外部调用 2 个（now, query）。

async fn mark_global_phase2_job_succeeded(
        &self,
        ownership_token: &str,
        completed_watermark: i64,
        selected_outputs: &[Stage1Output],
    ) -> anyhow::Result<bool>

作用：把全局合并任务标记为成功，并记录这次到底选中了哪些小记忆快照。

数据流：输入 token、完成水位和选中的输出列表 → 先结算 global 任务行 → 清掉旧的 selected 标记 → 对本次精确匹配的线程和更新时间打上选中标记 → 提交后返回是否成功。

调用关系：phase-2 合并完成后调用；任务行更新交给 mark_global_phase2_job_succeeded_row，然后自己维护 stage-1 输出的选中基线。

调用图：调用 1 个内部函数（mark_global_phase2_job_succeeded_row）；外部调用 1 个（query）。

async fn mark_global_phase2_job_failed(
        &self,
        ownership_token: &str,
        failure_reason: &str,
        retry_delay_seconds: i64,
    ) -> anyhow::Result<bool>

作用：把全局 phase-2 任务标记为失败，并安排稍后重试。

数据流：输入 token、失败原因和重试延迟 → 只更新匹配 token 的 running global 任务 → 状态改 error、清租约、记录错误、重试次数减一但不低于 0 → 返回是否更新成功。

调用关系：phase-2 工作器正常持有 token 但失败时调用；之后 try_claim_global_phase2_job 会按 retry_at 控制下一次领取。

调用图：外部调用 2 个（now, query）。

async fn mark_global_phase2_job_failed_if_unowned(
        &self,
        ownership_token: &str,
        failure_reason: &str,
        retry_delay_seconds: i64,
    ) -> anyhow::Result<bool>

作用：兜底处理全局 phase-2 失败：即使任务行丢了 ownership token，也能把卡住的 running 状态改成失败。

数据流：输入 token、失败原因和重试延迟 → 更新 matching token 或 token 为空的 running global 任务 → 写错误、清租约、设置重试时间 → 返回是否修复成功。

调用关系：当严格的 mark_global_phase2_job_failed 因 token 丢失无法更新时使用，帮助恢复卡住的全局任务。

调用图：外部调用 2 个（now, query）。

async fn mark_global_phase2_job_succeeded_row(
    executor: E,
    ownership_token: &str,
    completed_watermark: i64,
) -> anyhow::Result<u64>

作用：只负责更新 global phase-2 任务那一行，把它从 running 改成 done。

数据流：输入数据库执行器、token 和完成水位 → 写 finished_at、清租约和错误、推进成功水位 → 返回更新了几行。

调用关系：由 MemoryStore::mark_global_phase2_job_succeeded 调用；它处理任务状态，调用者再处理选中快照标记。

调用图：被 1 处调用（mark_global_phase2_job_succeeded）；外部调用 2 个（now, query）。

async fn clear_memory_data_in_pool(pool: &SqlitePool) -> anyhow::Result<()>

作用：在指定 SQLite 连接池里真正执行记忆数据清空。

数据流：输入数据库连接池 → 开启事务 → 删除所有 stage-1 输出和两类记忆任务 → 提交事务。

调用关系：MemoryStore::clear_memory_data 和其他清理入口会调用它，保证清理规则只有一份。

调用图：被 2 处调用（clear_memory_data_in_sqlite_home, clear_memory_data）；外部调用 2 个（begin, query）。

fn stage1_output_from_row_and_thread(
    row: &sqlx::sqlite::SqliteRow,
    thread: ThreadMetadata,
) -> anyhow::Result<Stage1Output>

作用：把数据库行和线程元信息拼成一个完整的 Stage1Output。

数据流：输入 stage-1 输出行和线程信息 → 读取时间、记忆文本、摘要、slug 等字段 → 把 Unix 秒时间转成日期对象 → 返回完整输出结构。

调用关系：由 stage1_output_from_row_if_thread_enabled 调用；它只负责转换，不负责判断线程是否可用。

调用图：调用 1 个内部函数（datetime_from_epoch_seconds）；被 1 处调用（stage1_output_from_row_if_thread_enabled）；外部调用 1 个（try_get）。

fn datetime_from_epoch_seconds(secs: i64) -> anyhow::Result<DateTime<Utc>>

作用：把 Unix 时间戳秒数转成程序使用的日期时间对象。

数据流：输入秒数 → 调用时间库转换 → 如果秒数非法就返回错误，否则返回 DateTime<Utc>。

调用关系：stage1_output_from_row_and_thread 用它处理数据库里的时间字段。

调用图：被 1 处调用（stage1_output_from_row_and_thread）；外部调用 1 个（from_timestamp）。

async fn enqueue_global_consolidation_with_executor(
    executor: E,
    input_watermark: i64,
) -> anyhow::Result<()>

作用：用任意数据库执行器排队或推进全局合并任务，是全局合并入队的底层实现。

数据流：输入执行器和输入水位 → 插入 pending global 任务；若已存在，则保持 running 不打断，否则设为 pending，并推进 bookkeeping 水位 → 返回成功或错误。

调用关系：被删除线程、stage-1 成功、stage-1 无输出删除、公开入队函数调用，是触发 phase-2 的共同通道。

调用图：被 4 处调用（delete_thread_memory, enqueue_global_consolidation, mark_stage1_job_succeeded, mark_stage1_job_succeeded_no_output）；外部调用 1 个（query）。

async fn clear_memory_data(&self) -> anyhow::Result<()>

作用：测试环境里的便捷入口，用运行时对象直接清空记忆数据。

数据流：输入 StateRuntime → 转交给内部 memories.clear_memory_data → 返回清理结果。

调用关系：只在测试编译时存在，让测试不用直接拿 MemoryStore。

async fn record_stage1_output_usage(&self, thread_ids: &[ThreadId]) -> anyhow::Result<usize>

作用：测试环境里的便捷入口，用来记录 stage-1 输出被使用。

数据流：输入线程 ID 列表 → 转交内部记忆存储 → 返回更新行数。

调用关系：只供测试调用，方便验证使用次数和最近使用时间的排序效果。

async fn claim_stage1_jobs_for_startup(
        &self,
        current_thread_id: ThreadId,
        params: Stage1StartupClaimParams<'_>,
    ) -> anyhow::Result<Vec<Stage1JobClaim>>

作用：测试环境里的便捷入口，用运行时对象模拟启动时领取 stage-1 任务。

数据流：输入当前线程和启动领取参数 → 转交 MemoryStore::claim_stage1_jobs_for_startup → 返回领取到的任务。

调用关系：测试用它检查年龄过滤、扫描上限、并发上限和禁用线程过滤。

async fn list_stage1_outputs_for_global(&self, n: usize) -> anyhow::Result<Vec<Stage1Output>>

作用：测试环境里的便捷入口，列出可供全局合并的 stage-1 输出。

数据流：输入数量上限 → 转交内部记忆存储 → 返回输出列表。

调用关系：测试 phase-2 选择和输出转换时会使用它。

async fn prune_stage1_outputs_for_retention(
        &self,
        max_unused_days: i64,
        limit: usize,
    ) -> anyhow::Result<usize>

作用：测试环境里的便捷入口，触发 stage-1 输出保留清理。

数据流：输入过期天数和删除上限 → 转交 MemoryStore → 返回实际删除数量。

调用关系：测试清理策略时调用，确认只删未选中且过期的输出。

async fn get_phase2_input_selection(
        &self,
        n: usize,
        max_unused_days: i64,
    ) -> anyhow::Result<Vec<Stage1Output>>

作用：测试环境里的便捷入口，取得 phase-2 当前输入选择。

数据流：输入数量上限和最大未使用天数 → 转交内部记忆存储 → 返回选择出的 Stage1Output。

调用关系：测试 phase-2 排序、污染线程排除、刷新后选择变化时使用。

async fn mark_thread_memory_mode_polluted(&self, thread_id: ThreadId) -> anyhow::Result<bool>

作用：测试环境里的便捷入口，把线程记忆标为污染。

数据流：输入线程 ID → 转交内部记忆存储 → 返回是否发生状态变更。

调用关系：测试污染线程是否触发重新合并时调用。

async fn try_claim_stage1_job(
        &self,
        thread_id: ThreadId,
        worker_id: ThreadId,
        source_updated_at: i64,
        lease_seconds: i64,
        max_running_jobs: usize,

作用：测试环境里的便捷入口，尝试领取单个 stage-1 任务。

数据流：输入线程、工作者、水位、租约和并发上限 → 转交 MemoryStore → 返回领取结果。

调用关系：大量测试用它搭建任务状态，比如成功、失败、并发抢占和重试耗尽。

async fn mark_stage1_job_succeeded(
        &self,
        thread_id: ThreadId,
        ownership_token: &str,
        source_updated_at: i64,
        raw_memory: &str,
        rollout_summary: &str,

作用：测试环境里的便捷入口，把 stage-1 任务结算为成功并写入输出。

数据流：输入线程、token、水位和输出文本 → 转交内部记忆存储 → 返回是否结算成功。

调用关系：测试用它生成 stage-1 输出，并观察是否触发 phase-2。

async fn mark_stage1_job_succeeded_no_output(
        &self,
        thread_id: ThreadId,
        ownership_token: &str,
    ) -> anyhow::Result<bool>

作用：测试环境里的便捷入口，把 stage-1 任务结算为成功但没有输出。

数据流：输入线程和 token → 转交内部记忆存储 → 返回是否结算成功。

调用关系：测试无输出时是否删除旧输出、是否需要触发 phase-2。

async fn mark_stage1_job_failed(
        &self,
        thread_id: ThreadId,
        ownership_token: &str,
        failure_reason: &str,
        retry_delay_seconds: i64,
    ) -> anyhow::Result<bool

作用：测试环境里的便捷入口，把 stage-1 任务标记失败。

数据流：输入线程、token、错误原因和重试延迟 → 转交内部记忆存储 → 返回是否更新成功。

调用关系：测试重试次数、重试冷却和新水位重置重试时会调用。

async fn enqueue_global_consolidation(&self, input_watermark: i64) -> anyhow::Result<()>

作用：测试环境里的便捷入口，手动排队一次全局合并。

数据流：输入水位 → 转交内部记忆存储 → 返回入队结果。

调用关系：测试 phase-2 锁、冷却和水位推进时用它制造待处理任务。

async fn try_claim_global_phase2_job(
        &self,
        worker_id: ThreadId,
        lease_seconds: i64,
    ) -> anyhow::Result<Phase2JobClaimOutcome>

作用：测试环境里的便捷入口，尝试领取全局 phase-2 锁。

数据流：输入工作者和租约 → 转交内部记忆存储 → 返回领取或跳过原因。

调用关系：phase-2 相关测试用它验证唯一运行者、租约过期接管、冷却和重试。

async fn mark_global_phase2_job_succeeded(
        &self,
        ownership_token: &str,
        completed_watermark: i64,
        selected_outputs: &[Stage1Output],
    ) -> anyhow::Result<bool>

作用：测试环境里的便捷入口，把全局 phase-2 任务标记为成功。

数据流：输入 token、完成水位和选中输出 → 转交内部记忆存储 → 返回是否结算成功。

调用关系：测试全局成功后 selected 标记和冷却行为时使用。

async fn mark_global_phase2_job_failed(
        &self,
        ownership_token: &str,
        failure_reason: &str,
        retry_delay_seconds: i64,
    ) -> anyhow::Result<bool>

作用：测试环境里的便捷入口，把全局 phase-2 任务标记失败。

数据流：输入 token、失败原因和重试延迟 → 转交内部记忆存储 → 返回是否更新成功。

调用关系：测试 phase-2 失败重试和重试次数耗尽时调用。

async fn mark_global_phase2_job_failed_if_unowned(
        &self,
        ownership_token: &str,
        failure_reason: &str,
        retry_delay_seconds: i64,
    ) -> anyhow::Result<bool>

作用：测试环境里的便捷入口，模拟兜底修复无 owner 的失败 phase-2 任务。

数据流：输入 token、失败原因和重试延迟 → 转交内部记忆存储 → 返回是否修复成功。

调用关系：专门给 fallback 失败处理测试使用。

fn stable_thread_id(value: &str) -> ThreadId

作用：测试辅助函数，把固定字符串变成稳定的线程 ID。

数据流：输入字符串 → 调用线程 ID 解析 → 返回测试可重复使用的 ThreadId。

调用关系：多个排序相关测试用它生成可预测的线程 ID，方便断言顺序。

调用图：调用 1 个内部函数（from_string）。

fn memory_pool(runtime: &StateRuntime) -> &sqlx::SqlitePool

作用：测试辅助函数，拿到运行时里的记忆数据库连接池。

数据流：输入运行时引用 → 访问 runtime.memories().pool → 返回底层 SQLite 连接池引用。

调用关系：测试需要直接查表或改表时会调用它。

调用图：调用 1 个内部函数（memories）。

async fn age_phase2_success_beyond_cooldown(runtime: &StateRuntime)

作用：测试辅助函数，把 phase-2 上次成功时间改得足够久远，绕过成功冷却期。

数据流：输入运行时 → 计算当前时间减去冷却时长再多一秒 → 直接更新 global job 的 finished_at → 测试随后就能再次领取 phase-2。

调用关系：多条 phase-2 测试用它从“刚成功不能跑”切换到“冷却已过可以跑”。

调用图：外部调用 3 个（now, query, memory_pool）。

async fn stage1_claim_skips_when_up_to_date()

作用：测试 stage-1 已经处理到同一水位时不会重复领取，源数据更新后才会重新领取。

数据流：创建运行时和线程 → 领取并成功完成水位 100 → 再用水位 100 领取应跳过 → 用水位 101 领取应成功。

调用关系：覆盖 try_claim_stage1_job 和 mark_stage1_job_succeeded 的基本去重关系。

调用图：调用 4 个内部函数（from_string, init, test_thread_metadata, unique_temp_dir）；外部调用 5 个（new_v4, assert!, assert_eq!, panic!, remove_dir_all）。

async fn stage1_running_stale_can_be_stolen_but_fresh_running_is_skipped()

作用：测试正在运行的 stage-1 任务：租约还新时别人不能抢，租约过期后可以接管。

数据流：创建线程并由 owner A 领取 → owner B 立即领取会被跳过 → 手动把租约改过期 → owner B 再领取成功。

调用关系：验证 try_claim_stage1_job 对 running 状态和 lease_until 的判断。

调用图：调用 4 个内部函数（from_string, init, test_thread_metadata, unique_temp_dir）；外部调用 6 个（new_v4, assert!, assert_eq!, query, memory_pool, remove_dir_all）。

async fn stage1_concurrent_claim_for_same_thread_is_conflict_safe()

作用：测试两个任务同时抢同一个线程时，最终只有一个能成功。

数据流：创建同一线程 → 两个异步任务同时尝试领取 → 收集结果 → 断言只有一个 Claimed，另一个是 SkippedRunning。

调用关系：验证 try_claim_stage1_job 的事务和冲突处理不会发出两张有效工牌。

调用图：调用 4 个内部函数（from_string, init, test_thread_metadata, unique_temp_dir）；外部调用 7 个（clone, new_v4, assert!, assert_eq!, remove_dir_all, join!, vec!）。

async fn stage1_concurrent_claims_respect_running_cap()

作用：测试 stage-1 全局并发上限生效，即使抢的是不同线程，也不能超过上限。

数据流：创建两个线程 → 并发领取且最大运行数设为 1 → 断言只有一个成功，另一个因运行上限被跳过。

调用关系：直接检验 try_claim_stage1_job 里统计 running job 的逻辑。

调用图：调用 4 个内部函数（from_string, init, test_thread_metadata, unique_temp_dir）；外部调用 7 个（clone, new_v4, assert!, assert_eq!, remove_dir_all, join!, vec!）。

async fn claim_stage1_jobs_filters_by_age_idle_and_current_thread()

作用：测试启动扫描只挑合适年龄、足够空闲、且不是当前线程的候选。

数据流：创建当前、新鲜、未够空闲、合格、过旧几种线程 → 调用启动领取 → 断言只领到合格线程。

调用关系：覆盖 claim_stage1_jobs_for_startup 的时间窗口和排除当前线程规则。

调用图：调用 4 个内部函数（from_string, init, test_thread_metadata, unique_temp_dir）；外部调用 8 个（days, hours, minutes, now, new_v4, assert_eq!, remove_dir_all, vec!）。

async fn claim_stage1_jobs_bounds_state_scan_before_memory_probes()

作用：测试启动扫描先受 scan_limit 限制，再去记忆库检查是否过期。

数据流：创建一个较新的已处理线程和一个较旧待处理线程 → scan_limit 为 1 时只看到已处理线程所以无领取 → scan_limit 为 2 时能领到旧线程。

调用关系：验证 claim_stage1_jobs_for_startup 的扫描边界和 stage1_source_needs_update 配合方式。

调用图：调用 4 个内部函数（from_string, init, test_thread_metadata, unique_temp_dir）；外部调用 8 个（hours, now, new_v4, assert!, assert_eq!, panic!, remove_dir_all, vec!）。

async fn claim_stage1_jobs_skips_threads_with_disabled_memory_mode()

作用：测试启动扫描不会给记忆模式禁用的线程派 stage-1 任务。

数据流：创建一个 disabled 线程和一个 enabled 线程 → 调用启动领取 → 断言只领到 enabled 线程。

调用关系：覆盖 claim_stage1_jobs_for_startup 中 memory_mode = enabled 的过滤。

调用图：调用 4 个内部函数（from_string, init, test_thread_metadata, unique_temp_dir）；外部调用 7 个（hours, now, new_v4, assert_eq!, query, remove_dir_all, vec!）。

async fn clear_memory_data_clears_rows_and_preserves_thread_memory_modes()

作用：测试清空记忆数据只删记忆输出和记忆任务，不改线程自己的记忆模式。

数据流：制造 stage-1 输出、global job、enabled/disabled 线程 → 调用清理 → 查表确认输出和任务为 0，线程模式仍原样。

调用关系：验证 clear_memory_data 和 clear_memory_data_in_pool 的删除范围。

调用图：调用 4 个内部函数（from_string, init, test_thread_metadata, unique_temp_dir）；外部调用 10 个（hours, now, new_v4, assert!, assert_eq!, panic!, query, query_scalar, memory_pool, remove_dir_all）。

async fn claim_stage1_jobs_enforces_global_running_cap()

作用：测试启动批量领取也遵守 stage-1 全局运行数量上限。

数据流：预先插入 10 个 running 任务，再创建 80 个候选 → 最大运行数 64 → 第一次只能补领到 54 个，第二次上限满后领不到。

调用关系：覆盖 claim_stage1_jobs_for_startup 调用 try_claim_stage1_job 时的上限传递。

调用图：调用 4 个内部函数（from_string, init, test_thread_metadata, unique_temp_dir）；外部调用 10 个（hours, seconds, now, new_v4, assert_eq!, format!, query, memory_pool, remove_dir_all, vec!）。

async fn claim_stage1_jobs_processes_two_full_batches_across_startup_passes()

作用：测试启动领取可以分批处理大量线程，不会只处理第一批。

数据流：创建 200 个合格线程 → 第一次领取 64 个并全部标成功 → 第二次再领取 64 个 → 断言两批都满额。

调用关系：验证 claim_stage1_jobs_for_startup、成功结算和过期判断配合后能继续推进队列。

调用图：调用 4 个内部函数（from_string, init, test_thread_metadata, unique_temp_dir）；外部调用 9 个（hours, seconds, now, new_v4, assert!, assert_eq!, format!, remove_dir_all, vec!）。

async fn delete_thread_removes_stage1_output_and_enqueues_phase2_when_selected()

作用：测试删除已进入全局记忆的线程时，会删掉它的小记忆并触发 phase-2 重新合并。

数据流：创建线程、写 stage-1 输出、完成一次 phase-2 选中 → 删除线程 → 确认输出消失，global job 变 pending 且水位更新。

调用关系：覆盖 delete_thread_memory 和 enqueue_global_consolidation_with_executor 的联动。

调用图：调用 4 个内部函数（from_string, init, test_thread_metadata, unique_temp_dir）；外部调用 8 个（now, new_v4, assert!, assert_eq!, panic!, query, memory_pool, remove_dir_all）。

async fn mark_stage1_job_succeeded_no_output_skips_phase2_when_output_was_already_absent()

作用：测试 stage-1 没有输出且本来就没有旧输出时，不应该触发全局合并。

数据流：创建线程并领取 stage-1 → 用 no_output 成功结算 → 确认没有 stage1_outputs 行，再领取同水位会跳过，global job 不存在。

调用关系：验证 mark_stage1_job_succeeded_no_output 在没有删除旧输出时不会多派 phase-2。

调用图：调用 4 个内部函数（from_string, init, test_thread_metadata, unique_temp_dir）；外部调用 7 个（new_v4, assert!, assert_eq!, panic!, query, memory_pool, remove_dir_all）。

async fn mark_stage1_job_succeeded_no_output_enqueues_phase2_when_deleting_output()

作用：测试 stage-1 新结果为空但删掉了旧输出时，会触发 phase-2 忘掉旧记忆。

数据流：先写入旧输出并完成一次 phase-2 → 新水位用 no_output 结算 → 确认旧输出删除 → 冷却后领取 phase-2，水位等于新水位。

调用关系：覆盖 mark_stage1_job_succeeded_no_output 与全局合并入队的条件。

调用图：调用 4 个内部函数（from_string, init, test_thread_metadata, unique_temp_dir）；外部调用 8 个（new_v4, assert!, assert_eq!, panic!, query, age_phase2_success_beyond_cooldown, memory_pool, remove_dir_all）。

async fn stage1_retry_exhaustion_does_not_block_newer_watermark()

作用：测试 stage-1 同一水位重试耗尽后，新水位仍能重新领取并重置重试次数。

数据流：同一线程连续失败 3 次 → 再领同水位应返回重试耗尽 → 用更新的 source_updated_at 领取成功 → 查表确认 retry_remaining 重置。

调用关系：验证 try_claim_stage1_job 对 input_watermark 前进的特殊放行逻辑。

调用图：调用 4 个内部函数（from_string, init, test_thread_metadata, unique_temp_dir）；外部调用 7 个（new_v4, assert!, assert_eq!, panic!, query, memory_pool, remove_dir_all）。

async fn phase2_global_lock_respects_success_cooldown()

作用：测试 phase-2 成功后有冷却期，短时间内即使再次入队也不能马上跑。

数据流：入队并完成一次 phase-2 → 立即再次领取被冷却拦住 → 再入队仍被拦住 → 手动老化成功时间后可以领取。

调用关系：覆盖 try_claim_global_phase2_job 的成功冷却规则和测试辅助老化函数。

调用图：调用 3 个内部函数（from_string, init, unique_temp_dir）；外部调用 6 个（new_v4, assert!, assert_eq!, panic!, age_phase2_success_beyond_cooldown, remove_dir_all）。

async fn phase2_global_lock_can_be_claimed_after_retry_budget_is_exhausted()

作用：测试 phase-2 即使失败重试次数用完，也仍可领取锁，因为是否有实际工作由外部文件差异判断。

数据流：入队 global 任务 → 连续失败 3 次把 retry_remaining 降到 0 → 再次尝试领取仍成功。

调用关系：验证 try_claim_global_phase2_job 不用 retry_remaining 阻断 phase-2 领取。

调用图：调用 3 个内部函数（from_string, init, unique_temp_dir）；外部调用 7 个（new_v4, assert!, assert_eq!, panic!, query, memory_pool, remove_dir_all）。

async fn list_stage1_outputs_for_global_returns_latest_outputs()

作用：测试列出全局合并输入时按最新输出排序，并补齐 cwd、git_branch、slug 等信息。

数据流：创建两个线程并写不同水位输出 → 调用列表函数 → 断言较新的在前，字段都来自正确来源。

调用关系：覆盖 list_stage1_outputs_for_global 和输出转换函数。

调用图：调用 4 个内部函数（from_string, init, test_thread_metadata, unique_temp_dir）；外部调用 5 个（new_v4, assert!, assert_eq!, panic!, remove_dir_all）。

async fn list_stage1_outputs_for_global_skips_empty_payloads()

作用：测试 raw_memory 和 rollout_summary 都为空的输出不会进入全局合并列表。

数据流：手动插入一条非空输出和一条空输出 → 请求 1 条 → 只返回非空那条。

调用关系：验证 list_stage1_outputs_for_global 的非空内容过滤。

调用图：调用 4 个内部函数（from_string, init, test_thread_metadata, unique_temp_dir）；外部调用 5 个（new_v4, assert_eq!, query, memory_pool, remove_dir_all）。

async fn list_stage1_outputs_for_global_skips_polluted_threads()

作用：测试被标记为 polluted 的线程，即使有 stage-1 输出，也不会给全局合并使用。

数据流：创建两个线程并各写输出 → 把其中一个设为 polluted → 列出全局输出 → 只剩 enabled 线程。

调用关系：覆盖 stage1_output_from_row_if_thread_enabled 和 enabled_thread_metadata 的过滤。

调用图：调用 4 个内部函数（from_string, init, test_thread_metadata, unique_temp_dir）；外部调用 5 个（new_v4, assert!, assert_eq!, panic!, remove_dir_all）。

async fn get_phase2_input_selection_returns_current_selected_rows()

作用：测试 phase-2 输入选择会返回当前按规则选中的输出集合。

数据流：创建三条输出并完成一次 phase-2 选择 → 再调用当前选择函数取 2 条 → 断言返回预期线程和路径信息。

调用关系：验证 get_phase2_input_selection 的基本选择和输出补全。

调用图：调用 4 个内部函数（from_string, init, test_thread_metadata, unique_temp_dir）；外部调用 6 个（new_v4, assert!, assert_eq!, panic!, stable_thread_id, remove_dir_all）。

async fn get_phase2_input_selection_excludes_polluted_previous_selection()

作用：测试即使某条记忆曾被 phase-2 选中过，只要线程后来污染，也会从当前输入选择中排除。

数据流：创建两个输出并完成 phase-2 选择 → 把其中一个线程设 polluted → 重新获取选择 → 只返回未污染线程。

调用关系：覆盖 get_phase2_input_selection 对线程当前状态的二次确认。

调用图：调用 4 个内部函数（from_string, init, test_thread_metadata, unique_temp_dir）；外部调用 5 个（new_v4, assert!, assert_eq!, panic!, remove_dir_all）。

async fn mark_thread_memory_mode_polluted_enqueues_phase2_for_selected_threads()

作用：测试把已被全局记忆选中的线程标污染，会触发下一次 phase-2。

数据流：写入输出并完成 phase-2 选中 → 调用污染标记 → 老化冷却 → 再领取 phase-2 应成功。

调用关系：验证 mark_thread_memory_mode_polluted 与 enqueue_global_consolidation 的联动。

调用图：调用 4 个内部函数（from_string, init, test_thread_metadata, unique_temp_dir）；外部调用 5 个（new_v4, assert!, panic!, age_phase2_success_beyond_cooldown, remove_dir_all）。

async fn mark_thread_memory_mode_polluted_enqueues_phase2_when_already_polluted()

作用：测试线程已经是 polluted 时，再调用污染标记虽然不算状态变化，但若它曾被选中仍会触发 phase-2。

数据流：先让线程输出被 phase-2 选中 → 手动把线程设 polluted → 调用污染函数返回 false → 冷却后 phase-2 仍可领取。

调用关系：覆盖 mark_thread_memory_mode_polluted 的“已污染但仍需忘记”场景。

调用图：调用 4 个内部函数（from_string, init, test_thread_metadata, unique_temp_dir）；外部调用 6 个（new_v4, assert!, panic!, query, age_phase2_success_beyond_cooldown, remove_dir_all）。

async fn get_phase2_input_selection_returns_regenerated_selected_rows()

作用：测试同一线程重新生成更新的 stage-1 输出后，phase-2 输入会使用最新内容。

数据流：先写水位 100 并完成选择 → 再写水位 101 → 获取选择 → 返回水位 101；同时旧 selected 快照时间仍记录为 100。

调用关系：验证 mark_stage1_job_succeeded 和 get_phase2_input_selection 对刷新输出的处理。

调用图：调用 4 个内部函数（from_string, init, test_thread_metadata, unique_temp_dir）；外部调用 6 个（new_v4, assert!, assert_eq!, panic!, memory_pool, remove_dir_all）。

async fn get_phase2_input_selection_uses_current_ranking_after_refreshes()

作用：测试多个线程刷新后，phase-2 输入按当前最新排序重新选择，而不是死守旧选择。

数据流：先创建四条输出并选中前两条 → 刷新其中几条到更高水位 → 获取当前选择 → 断言选出现在排名最高的两条。

调用关系：覆盖 get_phase2_input_selection 的当前排名规则。

调用图：调用 4 个内部函数（from_string, init, test_thread_metadata, unique_temp_dir）；外部调用 6 个（new_v4, assert!, assert_eq!, panic!, stable_thread_id, remove_dir_all）。

async fn mark_global_phase2_job_succeeded_updates_selected_snapshot_timestamp()

作用：测试 phase-2 成功后，会把被选中输出的精确 source_updated_at 记录下来。

数据流：先选中水位 100 → 刷新到 101 → 冷却后再次 phase-2 成功 → 查表确认 selected_for_phase2_source_updated_at 更新到 101。

调用关系：验证 mark_global_phase2_job_succeeded 写入选中快照时间的行为。

调用图：调用 4 个内部函数（from_string, init, test_thread_metadata, unique_temp_dir）；外部调用 7 个（new_v4, assert!, assert_eq!, panic!, age_phase2_success_beyond_cooldown, memory_pool, remove_dir_all）。

async fn mark_global_phase2_job_succeeded_only_marks_exact_selected_snapshots()

作用：测试 phase-2 成功只标记它真正看到的那份快照；如果中途输出刷新了，不会错误标记新快照。

数据流：phase-2 领取并保存选择时看到水位 100 → 在结算前 stage-1 刷新到 101 → phase-2 成功 → 查表确认新行没有被标为旧选择。

调用关系：覆盖 mark_global_phase2_job_succeeded 的精确匹配条件。

调用图：调用 4 个内部函数（from_string, init, test_thread_metadata, unique_temp_dir）；外部调用 6 个（new_v4, assert!, assert_eq!, panic!, memory_pool, remove_dir_all）。

async fn record_stage1_output_usage_updates_usage_metadata()

作用：测试记录使用情况会增加 usage_count，并写入 last_usage。

数据流：创建两条输出 → 记录 A 两次、B 一次、缺失线程一次 → 返回更新 3 行 → 查表确认 A=2、B=1，last_usage 有值。

调用关系：验证 record_stage1_output_usage 对存在和不存在输出的处理。

调用图：调用 4 个内部函数（from_string, init, test_thread_metadata, unique_temp_dir）；外部调用 7 个（new_v4, assert!, assert_eq!, panic!, query, memory_pool, remove_dir_all）。

async fn get_phase2_input_selection_prioritizes_usage_count_then_recent_usage()

作用：测试 phase-2 选择优先看使用次数；次数相同时看最近使用时间。

数据流：创建三条输出并手动设置 usage_count 和 last_usage → 请求 1 条 → 选中使用次数高且最近用过的线程。

调用关系：覆盖 get_phase2_input_selection 的排序优先级。

调用图：调用 4 个内部函数（from_string, init, test_thread_metadata, unique_temp_dir）；外部调用 11 个（days, hours, now, new_v4, assert!, assert_eq!, panic!, query, memory_pool, stable_thread_id (+1 more)）。

async fn get_phase2_input_selection_excludes_stale_used_memories_but_keeps_fresh_never_used()

作用：测试太久没用的旧记忆会被排除，但从没用过且本身较新的记忆会保留。

数据流：创建三条不同新旧和使用状态的输出 → 设置 last_usage → 获取选择 → 排除过期使用记录，保留新鲜未使用和最近使用的输出。

调用关系：验证 get_phase2_input_selection 的新鲜度判断。

调用图：调用 4 个内部函数（from_string, init, test_thread_metadata, unique_temp_dir）；外部调用 10 个（days, now, new_v4, assert!, assert_eq!, panic!, query, memory_pool, stable_thread_id, remove_dir_all）。

async fn get_phase2_input_selection_prefers_recent_thread_updates_over_recent_generation()

作用：测试 phase-2 排名看源线程更新时间，而不是生成输出的时间。

数据流：创建两个输出：一个 source_updated_at 旧但 generated_at 新，一个 source_updated_at 新但 generated_at 旧 → 请求 1 条 → 选中源更新时间新的线程。

调用关系：覆盖 get_phase2_input_selection 排序中 source_updated_at 的作用。

调用图：调用 4 个内部函数（from_string, init, test_thread_metadata, unique_temp_dir）；外部调用 7 个（new_v4, assert!, assert_eq!, panic!, query, memory_pool, remove_dir_all）。

async fn prune_stage1_outputs_for_retention_prunes_stale_unselected_rows_only()

作用：测试保留清理只删除过期且未被 phase-2 选中的输出，不删已选中的基线。

数据流：创建过期未用、过期用过、过期已选中、新鲜用过几类输出 → 执行清理 → 只删两个过期未选中行，任务记录数量不变。

调用关系：验证 prune_stage1_outputs_for_retention 的删除条件和不影响 jobs 的要求。

调用图：调用 4 个内部函数（from_string, init, test_thread_metadata, unique_temp_dir）；外部调用 10 个（days, now, new_v4, assert!, assert_eq!, panic!, query, memory_pool, remove_dir_all, vec!）。

async fn prune_stage1_outputs_for_retention_respects_batch_limit()

作用：测试保留清理遵守单次删除上限。

数据流：创建三条都应过期删除的输出 → limit 设为 2 → 清理后只删两条，剩一条。

调用关系：覆盖 prune_stage1_outputs_for_retention 的批量限制。

调用图：调用 4 个内部函数（from_string, init, test_thread_metadata, unique_temp_dir）；外部调用 9 个（days, now, new_v4, assert!, assert_eq!, panic!, query_scalar, memory_pool, remove_dir_all）。

async fn mark_stage1_job_succeeded_enqueues_global_consolidation()

作用：测试 stage-1 成功后会推进全局合并任务，并使用最新水位。

数据流：两个线程分别完成水位 100 和 101 的 stage-1 → 尝试领取 phase-2 → 返回 input_watermark 为 101。

调用关系：验证 mark_stage1_job_succeeded 调用全局合并入队函数。

调用图：调用 4 个内部函数（from_string, init, test_thread_metadata, unique_temp_dir）；外部调用 5 个（new_v4, assert!, assert_eq!, panic!, remove_dir_all）。

async fn phase2_global_lock_allows_only_one_fresh_runner()

作用：测试全局 phase-2 锁同一时间只允许一个有效运行者。

数据流：入队 global 任务 → owner A 领取成功 → owner B 立即领取返回 SkippedRunning。

调用关系：覆盖 try_claim_global_phase2_job 对活跃租约的互斥控制。

调用图：调用 3 个内部函数（from_string, init, unique_temp_dir）；外部调用 4 个（new_v4, assert!, assert_eq!, remove_dir_all）。

async fn phase2_global_lock_creates_missing_job_row()

作用：测试没有 global job 行时，领取 phase-2 会自动创建并占有它。

数据流：新运行时不先入队 → 直接领取 phase-2 得到水位 0 → 第二个 owner 被拦住 → 成功结算后立即再领会被冷却拦住。

调用关系：验证 try_claim_global_phase2_job 的缺省创建路径和冷却规则。

调用图：调用 3 个内部函数（from_string, init, unique_temp_dir）；外部调用 5 个（new_v4, assert!, assert_eq!, panic!, remove_dir_all）。

async fn phase2_global_lock_stale_lease_allows_takeover()

作用：测试 phase-2 租约过期后，新 owner 可以接管，旧 owner 不能再结算。

数据流：owner A 领取 → 手动让租约过期 → owner B 接管并拿到新 token → A 结算失败，B 结算成功。

调用关系：覆盖 try_claim_global_phase2_job 和 mark_global_phase2_job_succeeded 对 token 所有权的保护。

调用图：调用 3 个内部函数（from_string, init, unique_temp_dir）；外部调用 9 个（now, new_v4, assert!, assert_eq!, assert_ne!, panic!, query, memory_pool, remove_dir_all）。

async fn enqueue_global_consolidation_keeps_phase2_input_watermark_monotonic()

作用：测试即使后来入队的水位更低，phase-2 的 bookkeeping 水位也会继续前进，不会倒退。

数据流：先以水位 500 入队并成功 → 再以水位 400 入队 → 冷却后领取 → 断言拿到的水位大于 500。

调用关系：验证 enqueue_global_consolidation_with_executor 的水位推进策略。

调用图：调用 3 个内部函数（from_string, init, unique_temp_dir）；外部调用 6 个（new_v4, assert!, assert_eq!, panic!, age_phase2_success_beyond_cooldown, remove_dir_all）。

async fn phase2_failure_fallback_updates_unowned_running_job()

作用：测试 global 任务处于 running 但 ownership_token 丢失时，兜底失败函数能把它改成 error。

数据流：领取 phase-2 → 手动把 ownership_token 设为空 → 严格失败更新返回 false → fallback 更新返回 true → 再领取时因 retry_at 返回重试不可用。

调用关系：覆盖 mark_global_phase2_job_failed 和 mark_global_phase2_job_failed_if_unowned 的差别。

调用图：调用 4 个内部函数（from_string, new, init, unique_temp_dir）；外部调用 7 个（new_v4, assert!, assert_eq!, panic!, query, memory_pool, remove_dir_all）。

fn as_str(self) -> &'static str

作用：把任务状态从程序里的固定选项，变成数据库或接口里常用的小写文字。别人需要保存、展示或传输任务状态时会用它。

数据流：进去的是一个任务状态，比如 Running → 它按固定对应关系查表 → 出来的是字符串，比如 "running"，不会改动任何数据。

调用关系：它是状态枚举的对外翻译器，通常会在需要把程序内部状态写到外部世界时派上用场。调用图里没有显示它再调用别的函数，说明它只是一个很轻的小工具。

fn parse(value: &str) -> Result<Self>

作用：把外部来的状态文字，翻译成程序内部认可的任务状态。这样可以在读数据库或接收输入时，及时发现拼错或不支持的状态。

数据流：进去的是一段文字，比如 "completed" → 它检查这段文字是否属于允许的状态 → 如果合法，出来对应的 AgentJobStatus；如果不合法，出来一个错误，说明状态无效。

调用关系：AgentJob::try_from 在把数据库行变成 AgentJob 时会用它；另一个检查任务是否取消的流程 is_agent_job_cancelled 也会用它。它遇到坏状态时会交给 anyhow! 生成一条错误信息。

调用图：被 2 处调用（try_from, is_agent_job_cancelled）；外部调用 1 个（anyhow!）。

fn is_final(self) -> bool

作用：判断一个任务是不是已经走到终点。完成、失败、取消都算终点；等待中和运行中不算。

数据流：进去的是一个任务状态 → 它检查这个状态是不是 Completed、Failed 或 Cancelled → 出来 true 或 false，不改动任何东西。

调用关系：它提供给上层流程做判断：任务如果已经是最终状态，就不应该再继续推进。它内部只用 matches! 这种模式匹配工具做简单判断。

调用图：外部调用 1 个（matches!）。

fn as_str(self) -> &'static str

作用：把单个任务项的状态变成小写文字。任务项就是大任务里的某一行工作，保存或展示它的状态时需要这个转换。

数据流：进去的是一个任务项状态，比如 Failed → 它按固定对应关系转换 → 出来的是 "failed" 这样的字符串。

调用关系：它服务于任务项状态的对外表达，和 AgentJobStatus::as_str 类似。调用图没有显示它依赖其他函数，说明它只做直接转换。

fn parse(value: &str) -> Result<Self>

作用：把数据库或外部输入里的任务项状态文字，变成程序内部的固定状态。这样每一行工作的状态不会被随便写成奇怪的值。

数据流：进去的是一段状态文字，比如 "running" → 它核对是否是 pending、running、completed、failed 之一 → 合法就输出对应状态，不合法就输出错误。

调用关系：AgentJobItem::try_from 在读取数据库里的任务项时会调用它。遇到不认识的状态时，它用 anyhow! 生成错误，让坏数据不能继续往下走。

调用图：被 1 处调用（try_from）；外部调用 1 个（anyhow!）。

fn try_from(value: AgentJobRow) -> Result<Self, Self::Error>

作用：把数据库查出来的一行 AgentJobRow，转换成程序真正使用的 AgentJob。它像一道“入库检查门”，把字符串、数字等粗糙数据整理成可靠对象。

数据流：进去的是数据库行，里面有状态字符串、JSON 字符串、Unix 时间戳、0/1 表示的布尔值等 → 它解析 JSON，转换最大运行秒数，把状态文字变成枚举，把时间戳变成 UTC 时间，把 auto_export 的数字变成真假值 → 出来一个完整的 AgentJob；如果任何字段不合法，就出来错误。

调用关系：它是数据库层和业务层之间的转换口。它会调用 AgentJobStatus::parse 识别任务状态，调用 epoch_seconds_to_datetime 转换时间，也会用 serde_json 的 from_str 解析 JSON 字符串。

调用图：调用 2 个内部函数（parse, epoch_seconds_to_datetime）；外部调用 1 个（from_str）。

fn try_from(value: AgentJobItemRow) -> Result<Self, Self::Error>

作用：把数据库里的某个任务项记录 AgentJobItemRow，转换成程序里可直接使用的 AgentJobItem。它保证每一行工作的原始输入、结果、状态和时间都被正确还原。

数据流：进去的是数据库行，包含 row_json、result_json、状态文字、时间戳、错误信息等 → 它把 JSON 字符串解析成 JSON 值，把状态文字解析成固定状态，把时间戳转成 UTC 时间 → 出来一个 AgentJobItem；如果 JSON、状态或时间有问题，就返回错误。

调用关系：它在读取任务明细时发挥作用，把存储格式变成业务代码能放心使用的格式。它会调用 AgentJobItemStatus::parse、epoch_seconds_to_datetime，以及 serde_json 的 from_str。

调用图：调用 2 个内部函数（parse, epoch_seconds_to_datetime）；外部调用 1 个（from_str）。

fn epoch_seconds_to_datetime(secs: i64) -> Result<DateTime<Utc>>

作用：把 Unix 时间戳转换成真正的 UTC 日期时间。Unix 时间戳就是从 1970 年 1 月 1 日开始数到现在的秒数，数据库常用它来存时间。

数据流：进去的是一个秒数，比如 1710000000 → 它尝试用 chrono 库把秒数变成 UTC 时间 → 成功就输出 DateTime<Utc>，失败就输出“时间戳非法”的错误。

调用关系：AgentJob::try_from 和 AgentJobItem::try_from 都会用它来还原创建时间、更新时间、完成时间等字段。它把具体转换工作交给 chrono 的 from_timestamp。

调用图：被 2 处调用（try_from, try_from）；外部调用 1 个（from_timestamp）。

async fn create_agent_job(
        &self,
        params: &AgentJobCreateParams,
        items: &[AgentJobItemCreateParams],
    ) -> anyhow::Result<AgentJob>

作用：新建一个代理任务，并把这个任务下面的所有小工作一起写进数据库。有人提交一批要处理的数据时，会先走这里登记。

数据流：输入是任务参数和一组任务项；它先把表头、输出格式、每行数据等转成可存数据库的字符串，再开启一个数据库事务（一组要么全成功、要么全失败的操作），插入任务主记录和所有任务项；最后提交事务，并重新读出刚创建好的任务作为结果返回。

调用关系：它是创建任务的起点。写完数据库后，它会调用 StateRuntime::get_agent_job 把刚插入的任务再查出来，确保返回的是数据库里真实保存下来的版本。

调用图：调用 1 个内部函数（get_agent_job）；外部调用 4 个（now, from, to_string, query）。

async fn get_agent_job(&self, job_id: &str) -> anyhow::Result<Option<AgentJob>>

作用：按任务 ID 查一个代理任务。调用者想知道某个任务是否存在、当前是什么状态、路径和错误信息是什么时会用它。

数据流：输入是 job_id；它到 agent_jobs 表里查对应记录，如果查到就把数据库行转换成程序里的 AgentJob 对象，如果没查到就返回空；它不会改动数据库。

调用关系：它是读取单个任务的基础查询函数。StateRuntime::create_agent_job 在新建任务后会调用它，用来取回完整任务信息。

调用图：被 1 处调用（create_agent_job）。

async fn list_agent_job_items(
        &self,
        job_id: &str,
        status: Option<AgentJobItemStatus>,
        limit: Option<usize>,
    ) -> anyhow::Result<Vec<AgentJobItem>>

作用：列出一个任务下面的小工作项，可以按状态筛选，也可以限制返回数量。比如调度器要拿一批“等待处理”的行，就会用它。

数据流：输入是 job_id、可选状态和可选数量上限；它动态拼出查询条件，从 agent_job_items 表里按 row_index 顺序读出任务项；最后把数据库行转换成 AgentJobItem 列表返回，不修改数据。

调用关系：它通常服务于调度或界面展示：上层需要知道某个任务里有哪些行、哪些还没做时，就通过它从数据库取清单。

调用图：外部调用 1 个（new）。

async fn get_agent_job_item(
        &self,
        job_id: &str,
        item_id: &str,
    ) -> anyhow::Result<Option<AgentJobItem>>

作用：按任务 ID 和小项 ID 查一个具体的小工作项。适合用来查看某一行的状态、结果或错误。

数据流：输入是 job_id 和 item_id；它查询 agent_job_items 表中这一个小项，如果存在就转换成 AgentJobItem 返回，如果不存在就返回空；数据库内容不变。

调用关系：它是读取单个任务项的基础函数。测试和业务代码可以在更新后调用它，确认某个小项现在到底处于什么状态。

async fn mark_agent_job_running(&self, job_id: &str) -> anyhow::Result<()>

作用：把整个代理任务标记为“运行中”。当任务正式开始被执行时，会用它更新任务状态。

数据流：输入是 job_id；它取当前时间，把任务状态改成 Running，更新 updated_at，如果 started_at 还没有值就写入开始时间，同时清掉完成时间和上次错误；输出为空，只表示操作成功或失败。

调用关系：它通常在调度器开始处理任务前调用。之后任务里的小项才会陆续通过运行、完成或失败相关函数改变状态。

调用图：外部调用 2 个（now, query）。

async fn mark_agent_job_completed(&self, job_id: &str) -> anyhow::Result<()>

作用：把整个代理任务标记为“已完成”。当所有小工作都处理完并且任务整体成功时，会调用它。

数据流：输入是 job_id；它取当前时间，把任务状态改成 Completed，写入完成时间，清掉 last_error；输出为空，不返回任务内容。

调用关系：它处在任务生命周期的收尾阶段。通常是在进度统计显示没有待处理或运行中的小项后，由上层流程决定调用。

调用图：外部调用 2 个（now, query）。

async fn mark_agent_job_failed(
        &self,
        job_id: &str,
        error_message: &str,
    ) -> anyhow::Result<()>

作用：把整个代理任务标记为“失败”，并保存失败原因。遇到无法继续的错误时会用它留下事故记录。

数据流：输入是 job_id 和错误文字；它取当前时间，把任务状态改成 Failed，记录完成时间和错误信息；输出为空，只表示数据库更新是否成功执行。

调用关系：它是任务整体失败的收尾入口。上层运行流程捕获到严重错误后，会把错误消息交给它保存。

调用图：外部调用 2 个（now, query）。

async fn mark_agent_job_cancelled(
        &self,
        job_id: &str,
        reason: &str,
    ) -> anyhow::Result<bool>

作用：尝试取消一个还没结束的代理任务，并记录取消原因。它只会取消“等待中”或“运行中”的任务，避免改掉已经结束的任务。

数据流：输入是 job_id 和取消原因；它取当前时间，只有当任务状态还是 Pending 或 Running 时，才把状态改成 Cancelled，写入完成时间和原因；输出 true 表示真的改到了记录，false 表示没改到，可能任务已经结束或不存在。

调用关系：它用于用户或系统发起取消时。后续代码可以再用 StateRuntime::is_agent_job_cancelled 检查任务是否已经进入取消状态。

调用图：外部调用 2 个（now, query）。

async fn is_agent_job_cancelled(&self, job_id: &str) -> anyhow::Result<bool>

作用：检查某个代理任务是不是已经被取消。长时间运行的流程可以用它定期看一眼，决定要不要停下来。

数据流：输入是 job_id；它读取任务状态，如果任务不存在就当作没有取消，存在时把状态字符串解析成 AgentJobStatus，再判断是否等于 Cancelled；输出是布尔值。

调用关系：它是运行过程中的“是否该停”检查点。它会使用状态解析函数把数据库里的文字变成程序能判断的状态值。

调用图：调用 1 个内部函数（parse）；外部调用 1 个（query）。

async fn mark_agent_job_item_running(
        &self,
        job_id: &str,
        item_id: &str,
    ) -> anyhow::Result<bool>

作用：把某个小工作项从“等待中”抢成“运行中”。它用于没有指定线程 ID 的简单领取场景。

数据流：输入是 job_id 和 item_id；它只有在该小项当前还是 Pending 时才更新为 Running，尝试次数加一，更新时间，清掉错误，并把已分配线程清空；输出 true 表示领取成功，false 表示它已经不是等待中或不存在。

调用关系：它是任务项开始执行前的状态门槛。通过数据库条件保证同一个待处理小项不会被多个执行者轻易重复领取。

调用图：外部调用 2 个（now, query）。

async fn mark_agent_job_item_running_with_thread(
        &self,
        job_id: &str,
        item_id: &str,
        thread_id: &str,
    ) -> anyhow::Result<bool>

作用：把某个小工作项标记为“运行中”，同时记下由哪个线程负责。这里的线程 ID 可以理解为“当前处理人编号”。

数据流：输入是 job_id、item_id 和 thread_id；它只允许 Pending 状态的小项变成 Running，写入负责的线程 ID，尝试次数加一，更新时间并清掉旧错误；输出 true 表示成功领取，false 表示没有符合条件的记录。

调用关系：它比 StateRuntime::mark_agent_job_item_running 多保存一个负责人。后面的 StateRuntime::report_agent_job_item_result 会检查上报结果的线程 ID 是否匹配，防止别人替它乱交结果。

调用图：外部调用 2 个（now, query）。

async fn mark_agent_job_item_pending(
        &self,
        job_id: &str,
        item_id: &str,
        error_message: Option<&str>,
    ) -> anyhow::Result<bool>

作用：把正在运行的小工作项退回“等待中”，可选地留下错误信息。比如执行中临时失败、想之后重试时会用它。

数据流：输入是 job_id、item_id 和可选错误文字；它只会把当前 Running 的小项改回 Pending，清掉分配线程，更新时间，并记录错误信息；输出 true 表示成功退回，false 表示小项状态不符合或不存在。

调用关系：它用于重试流程。某次运行失败但还不想判死刑时，上层可以调用它把小项放回队列，之后再由运行标记函数重新领取。

调用图：外部调用 2 个（now, query）。

async fn set_agent_job_item_thread(
        &self,
        job_id: &str,
        item_id: &str,
        thread_id: &str,
    ) -> anyhow::Result<bool>

作用：给一个正在运行的小工作项补记或更新负责线程 ID。它用于运行中才知道具体线程的情况。

数据流：输入是 job_id、item_id 和 thread_id；它只更新当前状态为 Running 的小项，把 assigned_thread_id 写成新值并更新时间；输出 true 表示改到了记录，false 表示没改到。

调用关系：它服务于执行过程中的绑定动作。绑定后，StateRuntime::report_agent_job_item_result 可以用这个线程 ID 判断结果是不是由正确的执行者交上来的。

调用图：外部调用 2 个（now, query）。

async fn report_agent_job_item_result(
        &self,
        job_id: &str,
        item_id: &str,
        reporting_thread_id: &str,
        result_json: &Value,
    ) -> anyhow::Result<bool>

作用：让负责某个小工作项的线程提交处理结果，并把这个小项一次性标记为完成。它是防止“迟到结果乱覆盖状态”的关键函数。

数据流：输入是 job_id、item_id、上报线程 ID 和 JSON 结果；它把结果转成字符串，只在小项仍是 Running 且 assigned_thread_id 等于上报者时，写入结果、上报时间、完成时间，把状态改成 Completed，清掉错误和线程占用；输出 true 表示结果被接受，false 表示状态或线程不匹配，数据库不被改动。

调用关系：它位于任务项执行完成后的上报环节。测试专门验证它的原子性：成功上报会直接完成小项；如果小项已经失败，迟来的上报会被拒绝。

调用图：外部调用 3 个（now, to_string, query）。

async fn mark_agent_job_item_completed(
        &self,
        job_id: &str,
        item_id: &str,
    ) -> anyhow::Result<bool>

作用：把一个正在运行且已经有结果的小工作项标记为完成。它适合“结果已经先写好了，现在只补状态”的场景。

数据流：输入是 job_id 和 item_id；它只在小项状态为 Running 且 result_json 不为空时，改成 Completed，写入完成时间和更新时间，并清掉分配线程；输出 true 表示成功完成，false 表示条件不满足。

调用关系：它是任务项完成流程的另一种收尾方式。和 StateRuntime::report_agent_job_item_result 相比，它不接收新结果，只确认已有结果的小项可以结束。

调用图：外部调用 2 个（now, query）。

async fn mark_agent_job_item_failed(
        &self,
        job_id: &str,
        item_id: &str,
        error_message: &str,
    ) -> anyhow::Result<bool>

作用：把一个正在运行的小工作项标记为失败，并保存失败原因。某一行处理失败且不再继续时会调用它。

数据流：输入是 job_id、item_id 和错误文字；它只更新当前 Running 的小项，把状态改成 Failed，写入完成时间、更新时间和错误信息，并清掉分配线程；输出 true 表示成功标记失败，false 表示小项不是运行中或不存在。

调用关系：它是单个任务项失败的收尾入口。StateRuntime::report_agent_job_item_result 的测试会先调用它把小项变成失败，再确认迟到的结果不会被接受。

调用图：外部调用 2 个（now, query）。

async fn get_agent_job_progress(&self, job_id: &str) -> anyhow::Result<AgentJobProgress>

作用：统计一个代理任务的总体进度：总共有多少小项，分别有多少等待、运行、完成和失败。界面进度条或调度判断都会需要它。

数据流：输入是 job_id；它在 agent_job_items 表里按状态做计数，把数据库里的数字转换成程序里的 usize 数字；输出是 AgentJobProgress，里面装着各类数量，不修改数据库。

调用关系：它是进度汇总函数。任务执行过程中或完成后，上层可以调用它判断任务还剩多少工作，也能给用户展示当前完成情况。

调用图：外部调用 2 个（query, try_from）。

async fn create_running_single_item_job(
        runtime: &StateRuntime,
    ) -> anyhow::Result<(String, String, String)>

作用：这是测试用的小帮手，用来快速造出一个“已经开始运行、只有一个小项、并绑定了线程”的任务。这样每个测试不用重复写一大段准备代码。

数据流：输入是测试里的 StateRuntime；它创建一个固定 ID 的任务和一个任务项，然后把任务标记为运行中，再把这个小项用 thread-1 标记为运行中；输出 job_id、item_id 和 thread_id，供测试继续操作。

调用关系：它被下面两个测试函数调用。它内部会调用创建任务、标记任务运行、标记任务项运行并绑定线程这些真实函数，所以测试环境和实际流程很接近。

调用图：外部调用 6 个（assert!, json!, create_agent_job, mark_agent_job_item_running_with_thread, mark_agent_job_running, vec!）。

async fn report_agent_job_item_result_completes_item_atomically() -> anyhow::Result<()>

作用：这个测试确认：任务项上报结果时，会同时保存结果、标记完成、清掉线程占用，并正确更新进度。重点是这些变化要像一次动作一样完成。

数据流：它先创建临时目录和运行时环境，再用 tests::create_running_single_item_job 准备一个运行中的小项；接着上报 JSON 结果；最后重新查询小项和进度，检查状态、结果、时间、错误和统计数量都符合预期。

调用关系：它主要验证 StateRuntime::report_agent_job_item_result 的成功路径，同时也间接使用 StateRuntime::get_agent_job_item 和 StateRuntime::get_agent_job_progress 来确认数据库里的最终状态。

调用图：调用 2 个内部函数（init, unique_temp_dir）；外部调用 4 个（assert!, assert_eq!, json!, create_running_single_item_job）。

async fn report_agent_job_item_result_rejects_late_reports() -> anyhow::Result<()>

作用：这个测试确认：如果一个小工作项已经被标记失败，后来才到的结果不能把它改成成功。这样可以避免慢半拍的执行者覆盖正确的失败记录。

数据流：它先准备一个运行中的单项任务，然后把该小项标记为失败；随后尝试用原来的线程 ID 上报一个迟到结果；函数期望上报被拒绝，最后查询小项，确认状态仍是 Failed，结果为空，错误信息还在。

调用关系：它验证 StateRuntime::report_agent_job_item_result 的保护条件，也用到 StateRuntime::mark_agent_job_item_failed 制造“已经失败”的场景，确保状态转换不会被迟到上报打乱。

调用图：调用 2 个内部函数（init, unique_temp_dir）；外部调用 4 个（assert!, assert_eq!, json!, create_running_single_item_job）。

fn default() -> Self

作用：给回填任务创建一个最初始的状态。它表示任务还没开始，没有处理进度，也没有成功完成过。

数据流：进去不需要任何输入 → 它填好一份默认记录：状态是 pending，last_watermark 为空，last_success_at 为空 → 出来一个新的 BackfillState，供后续保存或使用。

调用关系：当 backfill_sessions_with_lease 需要一份初始回填状态时会用到它。它不再把工作交给别的函数，只是提供一个安全的起点。

调用图：被 1 处调用（backfill_sessions_with_lease）。

fn try_from_row(row: &SqliteRow) -> Result<Self>

作用：把数据库里查出来的一行回填状态，翻译成程序能直接使用的 BackfillState。它会顺便检查状态文字和时间是不是合法。

数据流：进去是一行 SQLite 查询结果 → 它读取 status、last_watermark、last_success_at 这些字段；把 status 交给 BackfillStatus::parse 变成固定枚举；把 last_success_at 这种 Unix 时间戳（从 1970 年开始数的秒数）转成真正的 UTC 时间 → 出来是 BackfillState；如果字段读不出、状态不认识、时间不合法，就返回错误。

调用关系：get_backfill_state 从数据库拿到原始行后会调用它。它内部会调用 BackfillStatus::parse 来校验状态，也会通过数据库读取接口 try_get 取字段。

调用图：调用 1 个内部函数（parse）；被 1 处调用（get_backfill_state）；外部调用 1 个（try_get）。

fn as_str(self) -> &'static str

作用：把程序里的状态值变成适合写进数据库或日志的短文字。比如 Pending 会变成 "pending"。

数据流：进去是一个 BackfillStatus 状态 → 它按照三种固定状态做对应转换 → 出来是一个静态字符串，不会分配新内存，也不会修改任何东西。

调用关系：它通常用于需要把状态保存或展示的时候。这个函数本身不调用别的函数，是状态和文字之间的简单对照表。

fn parse(value: &str) -> Result<Self>

作用：把数据库或外部读到的状态文字，检查并转成程序认可的状态。它能防止拼错或未知状态混进系统。

数据流：进去是一段文字，比如 "pending" → 它只接受 "pending"、"running"、"complete" 三种 → 出来是对应的 BackfillStatus；如果文字不在名单里，就用 anyhow! 生成一个带说明的错误。

调用关系：BackfillState::try_from_row 在读取数据库状态时会调用它。它是数据库文字进入程序内部之前的一道门卫。

调用图：被 1 处调用（try_from_row）；外部调用 1 个（anyhow!）。

fn epoch_seconds_to_datetime(secs: i64) -> Result<DateTime<Utc>>

作用：把数据库里保存的秒级时间戳转换成 UTC 时间对象。这样后面的代码就不用直接处理一串数字。

数据流：进去是一个 i64 秒数，表示从 1970-01-01 00:00:00 UTC 到某个时间经过了多少秒 → 它调用时间库的 from_timestamp 做转换 → 出来是 DateTime<Utc>；如果这个秒数超出合法范围，就返回错误。

调用关系：BackfillState::try_from_row 在读取 last_success_at 时会间接使用它。它不关心数据库，只负责把数字时间变成安全、明确的时间类型。

调用图：外部调用 1 个（from_timestamp）。

async fn get_backfill_state(&self) -> anyhow::Result<crate::BackfillState>

作用：读取当前补历史数据的状态。外部代码用它来知道补数是还没开始、正在跑、还是已经完成，以及上次处理到哪个位置。

数据流：输入是当前 StateRuntime，它里面带着数据库连接池。函数先确认 backfill_state 表里那条 id=1 的记录存在，然后从数据库读出 status、last_watermark、last_success_at，最后交给 try_from_row 转成程序里好用的 BackfillState 对象返回；它本身不会推进进度。

调用关系：它是查询入口。调用时先把“确保单行记录存在”的小活交给 StateRuntime::ensure_backfill_state_row，再用 SQLx（Rust 里访问数据库的工具）查询数据库，最后让 try_from_row 把数据库行翻译成业务对象。

调用图：调用 2 个内部函数（try_from_row, ensure_backfill_state_row）；外部调用 1 个（query）。

async fn try_claim_backfill(&self, lease_seconds: i64) -> anyhow::Result<bool>

作用：尝试抢到补历史数据的执行权。这样可以防止多个程序实例同时补同一批数据，造成重复工作或互相覆盖。

数据流：输入是 lease_seconds，也就是这次占用有效多久。函数先确保状态行存在，取当前时间，算出“多久没更新就算过期”的时间点；然后尝试把状态改成 Running。只有在任务未完成，并且没有有效的 Running 占用时，数据库更新才会成功。输出 true 表示抢到了，false 表示已经完成或别人还在做。

调用关系：它通常在后台补数 worker 准备开工前被调用。它先调用 StateRuntime::ensure_backfill_state_row 做安全垫，再用 now 取时间、用 SQL 更新状态；后续如果抢到，调用方才会继续做补数并调用 checkpoint 或 complete 之类的方法。

调用图：调用 1 个内部函数（ensure_backfill_state_row）；外部调用 2 个（now, query）。

async fn mark_backfill_running(&self) -> anyhow::Result<()>

作用：把补历史数据明确标记为“正在运行”。当程序已经决定开始干活时，用它把数据库里的进度卡更新成运行中。

数据流：输入是当前 StateRuntime。函数先确保那条唯一状态记录存在，然后把 status 写成 Running，并把 updated_at 写成当前时间。输出只是成功或失败；成功后数据库中的状态已经改变。

调用关系：它是状态切换的小工具。调用时先经过 StateRuntime::ensure_backfill_state_row，然后用 now 取当前时间，再用 SQLx 执行数据库更新。它通常出现在补数流程开始或重新确认运行状态的时候。

调用图：调用 1 个内部函数（ensure_backfill_state_row）；外部调用 2 个（now, query）。

async fn checkpoint_backfill(&self, watermark: &str) -> anyhow::Result<()>

作用：保存补历史数据的中途进度点。这样程序崩了或重启后，可以从最后处理到的位置附近继续，而不是完全重来。

数据流：输入是 watermark，也就是“处理到哪里了”的标记，例如某个文件路径或排序位置。函数确保状态行存在后，把状态写成 Running，把 last_watermark 写成传入的进度点，并刷新 updated_at。输出成功或失败；成功后数据库记住了新的进度。

调用关系：它通常由正在跑的补数 worker 周期性调用。它先依赖 StateRuntime::ensure_backfill_state_row 保证记录存在，再用 now 和 SQLx 把新的进度写入数据库。

调用图：调用 1 个内部函数（ensure_backfill_state_row）；外部调用 2 个（now, query）。

async fn mark_backfill_complete(&self, last_watermark: Option<&str>) -> anyhow::Result<()>

作用：把补历史数据标记为完成。完成后，后续再尝试抢任务的人会被挡住，不会重复补已经补完的历史数据。

数据流：输入是可选的 last_watermark。如果传了最后进度点，函数会把它写入数据库；如果没传，就保留原来的进度点。函数还会把 status 改成 Complete，并把 last_success_at 和 updated_at 都写成当前时间。输出成功或失败。

调用关系：它通常在补数 worker 成功跑完全部历史数据后调用。它先调用 StateRuntime::ensure_backfill_state_row，再用 now 生成完成时间，并通过 SQLx 写入最终状态；之后 StateRuntime::try_claim_backfill 会看到 Complete 而拒绝新的抢占。

调用图：调用 1 个内部函数（ensure_backfill_state_row）；外部调用 2 个（now, query）。

async fn ensure_backfill_state_row(&self) -> anyhow::Result<()>

作用：保证 backfill_state 表里那条 id=1 的唯一进度记录存在。它像先检查“进度表有没有贴在墙上”，没有就补一张。

数据流：输入是当前 StateRuntime 和它的数据库连接池。函数把实际工作交给 ensure_backfill_state_row_in_pool：检查或插入那条默认状态记录。输出成功或失败；成功后，后面的读取和更新都能放心针对 id=1 操作。

调用关系：它是本文件所有正式方法的共同前置步骤。StateRuntime::get_backfill_state、try_claim_backfill、mark_backfill_running、checkpoint_backfill、mark_backfill_complete 都会先调用它，避免因为单例记录缺失导致查询或更新出错。

调用图：被 5 处调用（checkpoint_backfill, get_backfill_state, mark_backfill_complete, mark_backfill_running, try_claim_backfill）。

async fn backfill_state_persists_progress_and_completion()

作用：测试补历史数据的状态能从初始、运行中、保存进度，一直到完成，全部正确写进数据库并读回来。

数据流：测试先创建临时目录并初始化 StateRuntime，读取初始状态，确认是 Pending 且没有进度。接着标记运行中，写入一个 checkpoint，再读回来确认状态和进度都对。最后标记完成并传入最终 watermark，再读回来确认状态为 Complete、最终进度被保存、成功时间已经写入。结束时删除临时目录。

调用关系：它像一次完整演练，会间接覆盖 init、StateRuntime::get_backfill_state、mark_backfill_running、checkpoint_backfill、mark_backfill_complete 等行为，并用 assert_eq 和 assert 检查结果是不是符合预期。

调用图：调用 2 个内部函数（init, unique_temp_dir）；外部调用 3 个（assert!, assert_eq!, remove_dir_all）。

async fn get_backfill_state_succeeds_while_another_connection_holds_writer_slot()

作用：测试即使另一个数据库连接占着写入位置，读取补数状态也能成功。这个很重要，因为状态查询不应该轻易被别的写操作拖死。

数据流：测试先初始化运行环境，再手动打开另一个 SQLite 数据库连接，并用 BEGIN IMMEDIATE 占住写锁。随后通过 runtime 读取 backfill 状态，确认能拿到默认状态。最后回滚事务释放写锁并清理临时目录。

调用关系：它主要验证 StateRuntime::get_backfill_state 在数据库有并发写锁时仍能工作。过程中会用 init、base_sqlite_options、state_db_path 和 connect_with 建立测试场景，再用断言确认读取结果。

调用图：调用 2 个内部函数（init, unique_temp_dir）；外部调用 5 个（assert_eq!, state_db_path, connect_with, base_sqlite_options, remove_dir_all）。

async fn get_backfill_state_repairs_a_missing_singleton_row()

作用：测试如果 backfill_state 里那条唯一记录被删了，读取状态时会自动修复。这样系统不会因为一条缺失的进度记录就坏掉。

数据流：测试先初始化 StateRuntime，然后直接用 SQL 删除 id=1 的状态行。接着调用 get_backfill_state，期望它返回默认状态。最后再查数据库里的行数，确认 id=1 的记录确实被补回来了，并清理临时目录。

调用关系：它重点验证 StateRuntime::get_backfill_state 会先触发 StateRuntime::ensure_backfill_state_row。测试通过 query 手动破坏数据库，再观察读取动作是否把数据库修好。

调用图：调用 2 个内部函数（init, unique_temp_dir）；外部调用 3 个（assert_eq!, query, remove_dir_all）。

async fn backfill_claim_is_singleton_until_stale_and_blocked_when_complete()

作用：测试补历史数据的执行权同一时间只能被一个人拿到，旧占用过期后可以被接手，但完成后谁都不能再抢。

数据流：测试初始化运行环境后，第一次 try_claim_backfill 应该成功；马上再抢一次应该失败。然后它手动把 updated_at 改成很久以前，模拟上一个 worker 卡死或失联，再抢应该成功。最后标记完成，再尝试抢占应失败。结束时删除临时目录。

调用关系：它集中验证 StateRuntime::try_claim_backfill 和 StateRuntime::mark_backfill_complete 的配合。测试用 now 和 query 人为制造“租约过期”的数据库状态，再用断言确认抢占规则符合预期。

调用图：调用 2 个内部函数（init, unique_temp_dir）；外部调用 4 个（now, assert_eq!, query, remove_dir_all）。

async fn record_external_agent_config_import_completed(
        &self,
        import_id: &str,
        successes: &[ExternalAgentConfigImportSuccessRecord],
        failures: &[ExternalAgentConfigImp

作用：这个函数在一次外部代理配置导入结束时使用，把这次导入的成功清单和失败清单写进数据库。它的作用是留下可追溯的记录，之后用户或系统才能知道这次导入到底发生了什么。

数据流：进去的是一个导入编号 import_id、一组成功记录和一组失败记录。函数先取当前时间，并把时间转成毫秒时间戳；再把成功和失败列表转成 JSON 文本；最后把这些内容写入 external_agent_config_imports 表。如果数据库里已经有同一个 import_id，它会覆盖更新时间、成功列表和失败列表。出来的是一个成功或失败的结果；成功时数据库被更新，失败时返回错误。

调用关系：它通常由执行导入流程的上层代码在“导入完成”那一刻调用。它自己不判断哪些项目成功或失败，只负责把别人整理好的结果安全落库。过程中它会调用当前时间函数、时间转换函数、JSON 序列化函数，以及 sqlx 的 query 来执行数据库写入。

调用图：外部调用 4 个（now, datetime_to_epoch_millis, to_string, query）。

async fn external_agent_config_import_details_record(
        &self,
        import_id: &str,
    ) -> anyhow::Result<Option<ExternalAgentConfigImportDetailsRecord>>

作用：这个函数用来按导入编号查看某一次导入的详细结果。有人想点开某条导入记录，看里面哪些成功、哪些失败，就会用到它。

数据流：进去的是 import_id。函数拿这个编号去数据库查对应行，只取 successes 和 failures 两列。查到后，它把数据库里的 JSON 文本还原成成功记录列表和失败记录列表，并装成 ExternalAgentConfigImportDetailsRecord 返回。没查到就返回空；查数据库或解析 JSON 出错就返回错误。

调用关系：它位于“查看单次导入详情”的读取路径上，通常会被界面接口或业务层调用。它把活交给 sqlx 的 query 去查数据库，自己负责把查出的文本变回结构化记录，方便调用方直接展示或处理。

调用图：外部调用 1 个（query）。

async fn external_agent_config_import_history_records(
        &self,
    ) -> anyhow::Result<Vec<ExternalAgentConfigImportHistoryRecord>>

作用：这个函数用来列出所有外部代理配置导入的历史记录。它适合给历史页面使用，让用户按时间看到最近做过哪些导入，以及每次导入的结果摘要。

数据流：进去不需要额外参数。函数从 external_agent_config_imports 表取出所有导入记录，包括导入编号、完成时间、成功列表和失败列表，并按完成时间从新到旧排序；时间一样时再按 import_id 排序。然后它逐行把 JSON 文本还原成成功和失败记录，组成一组 ExternalAgentConfigImportHistoryRecord 返回。数据库读取或 JSON 解析失败时返回错误。

调用关系：它处在“展示导入历史”的读取路径上，通常由需要列表数据的上层代码调用。它通过 sqlx 的 query 读取数据库，不负责生成导入结果，只负责把过去保存的结果整理成调用方好用的历史记录列表。

调用图：外部调用 1 个（query）。

fn remote_control_app_server_client_name_key(app_server_client_name: Option<&str>) -> &str

作用：把可有可无的客户端名字变成数据库里能用来查找的一段固定文字。没有客户端名字时，它用空字符串代替。

数据流：进去的是一个可能为空的客户端名字；如果有名字，就原样返回；如果没有，就返回预设的空字符串。它不改数据库，只是统一“查找钥匙”的写法。

调用关系：查、存、改开关、删除登记记录时都会先经过它。这样所有数据库操作都用同一种规则处理“没有客户端名字”，避免保存时一个样、查询时另一个样。

调用图：被 4 处调用（delete_remote_control_enrollment, get_remote_control_enrollment, set_remote_control_enabled, upsert_remote_control_enrollment）。

fn app_server_client_name_from_key(app_server_client_name: String) -> Option<String>

作用：把数据库里存的客户端名字钥匙还原成程序里更自然的表示。空字符串会被还原成“没有客户端名字”。

数据流：进去的是从数据库读出的字符串；如果字符串为空，就输出 None，表示没有值；如果不为空，就包成 Some，表示真的有客户端名字。它只做格式转换，不访问外部资源。

调用关系：它主要接在读取数据库之后使用，配合 remote_control_app_server_client_name_key 形成一进一出的转换。前者负责写入和查询前的统一，后者负责读出来后恢复给调用者。

async fn get_remote_control_enrollment(
        &self,
        websocket_url: &str,
        account_id: &str,
        app_server_client_name: Option<&str>,
    ) -> anyhow::Result<Option<RemoteControl

作用：按 websocket 地址、账号、客户端名字这三个条件，从本地数据库找一条远程控制登记信息。找不到时不会报错，而是返回“没有”。

数据流：进去的是远程控制地址、账号 ID、可选的客户端名字；它先把客户端名字转成数据库用的钥匙，再执行一条 SELECT 查询；如果数据库有匹配行，就组装成 RemoteControlEnrollmentRecord 返回，如果没有就返回 None。数据库本身不会被改动。

调用关系：这是外部代码想恢复远程控制登记状态时会调用的入口。它把查找钥匙的细节交给 remote_control_app_server_client_name_key，把真正的数据库读取交给 sqlx 的 query。

调用图：调用 1 个内部函数（remote_control_app_server_client_name_key）；外部调用 1 个（query）。

async fn upsert_remote_control_enrollment(
        &self,
        enrollment: &RemoteControlEnrollmentRecord,
    ) -> anyhow::Result<()>

作用：保存一条远程控制登记信息；如果同一个地址、账号、客户端名字已经存在，就更新服务器资料。upsert 的意思就是“没有就插入，有就更新”。

数据流：进去的是一整条登记记录；它把可选客户端名字转成数据库钥匙，填入 INSERT 语句，并写入当前时间；如果数据库发现同一个唯一键已经存在，就只更新服务器 ID、环境 ID、服务器名和更新时间。操作成功后不返回具体记录，只表示完成。

调用关系：当远程控制登记成功或服务器资料刷新时会用它。它会调用 remote_control_app_server_client_name_key 统一查找键，并调用 now 取当前时间，再让 sqlx 执行数据库写入。注意它不会在冲突更新时改 remote_control_enabled，这个开关由 StateRuntime::set_remote_control_enabled 单独负责。

调用图：调用 1 个内部函数（remote_control_app_server_client_name_key）；外部调用 2 个（now, query）。

async fn set_remote_control_enabled(
        &self,
        websocket_url: &str,
        account_id: &str,
        app_server_client_name: Option<&str>,
        remote_control_enabled: bool,
    ) ->

作用：只修改某条远程控制登记的“是否启用”开关。这样用户开关偏好不会被普通的登记资料更新顺手覆盖。

数据流：进去的是地址、账号、可选客户端名字，以及新的开关值；它把客户端名字转成数据库钥匙，执行 UPDATE，把 remote_control_enabled 和更新时间改掉；出来的是受影响的行数，通常 1 表示改到了，0 表示没找到匹配记录。

调用关系：当用户或上层流程要打开、关闭远程控制时会调用它。它和 upsert_remote_control_enrollment 分工明确：upsert 管服务器资料，这个函数管开关；底层同样把查询条件交给 remote_control_app_server_client_name_key，数据库操作交给 sqlx。

调用图：调用 1 个内部函数（remote_control_app_server_client_name_key）；外部调用 2 个（now, query）。

async fn delete_remote_control_enrollment(
        &self,
        websocket_url: &str,
        account_id: &str,
        app_server_client_name: Option<&str>,
    ) -> anyhow::Result<u64>

作用：删除指定的一条远程控制登记记录，不会把同一服务器地址下其他账号或其他客户端的记录一起删掉。

数据流：进去的是地址、账号、可选客户端名字；它把客户端名字转成数据库钥匙，然后按这三个条件执行 DELETE；出来的是删除了几行，1 通常表示成功删掉一条，0 表示原本就没有这条。

调用关系：当账号退出、取消登记或需要清理某个远程控制绑定时会用它。它依赖 remote_control_app_server_client_name_key 保证删除条件和保存、查询时完全一致，再通过 sqlx 执行真正的数据库删除。

调用图：调用 1 个内部函数（remote_control_app_server_client_name_key）；外部调用 1 个（query）。

async fn remote_control_enrollment_round_trips_by_target_and_account()

作用：这个测试确认登记信息能“存进去再读出来”，并且不同账号、不同客户端名字不会互相串号。

数据流：它先创建一个临时目录，初始化一套测试用状态数据库；然后插入两条地址相同但账号不同的登记；接着读取指定账号的记录，确认内容完整一致；再查不存在的账号和错误客户端名字，确认返回 None；最后清理临时目录。

调用关系：它模拟正常使用 StateRuntime::upsert_remote_control_enrollment 和 StateRuntime::get_remote_control_enrollment 的场景。测试通过说明查找钥匙确实包含地址、账号、客户端名字这几个关键条件。

调用图：调用 2 个内部函数（init, unique_temp_dir）；外部调用 2 个（assert_eq!, remove_dir_all）。

async fn delete_remote_control_enrollment_removes_only_matching_entry()

作用：这个测试确认删除操作只删除完全匹配的那一条记录，不会误删同地址下另一个账号的登记。

数据流：它建立临时数据库，插入两条客户端名字都为空但账号不同的登记；然后删除 account-a 的那条；之后再查 account-a，应该没有了；查 account-b，应该还完整保留；最后删除临时目录。

调用关系：它围绕 StateRuntime::delete_remote_control_enrollment 展开，同时用 StateRuntime::get_remote_control_enrollment 验证删除后的结果。它特别覆盖了“没有客户端名字会存成空字符串”这种情况。

调用图：调用 2 个内部函数（init, unique_temp_dir）；外部调用 2 个（assert_eq!, remove_dir_all）。

async fn migration_preserves_legacy_remote_control_preference_as_null()

作用：这个测试确认老版本数据库升级后，旧登记记录的远程控制开关保持为空，而不是被偷偷改成开或关。

数据流：它先创建临时目录和一个老版本状态数据库，只运行旧迁移；然后手工插入一条旧格式的远程控制登记，旧格式里没有 remote_control_enabled；接着用新版 StateRuntime 初始化，让迁移跑完；最后读出这条记录，确认 remote_control_enabled 是 None，并清理临时目录。

调用关系：它把数据库迁移流程和 StateRuntime::get_remote_control_enrollment 串起来检查。这个测试保护一个兼容性承诺：老用户升级后，程序不能擅自猜测他们的远程控制开关偏好。

调用图：调用 2 个内部函数（init, unique_temp_dir）；外部调用 8 个（Owned, new, connect_with, assert_eq!, state_db_path, query, create_dir_all, remove_dir_all）。

fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result

作用：这个函数决定调试打印 LocalAgentGraphStore 时显示什么内容。它不会把整个数据库对象打印出来，只露出 codex_home 这个关键路径，方便排查问题又避免信息太乱。

数据流：进去的是一个 LocalAgentGraphStore 和一个格式化输出器 → 它读取 state_db 里的 codex_home 路径，并用 debug_struct 组装成调试文本 → 出来的是格式化是否成功的结果，不改动数据库。

调用关系：当开发者或日志系统用调试格式打印 LocalAgentGraphStore 时会走到这里。它只把活儿交给标准的 debug_struct 来生成好看的调试结构。

调用图：外部调用 1 个（debug_struct）。

fn new(state_db: Arc<StateRuntime>) -> Self

作用：这个函数用一个已经准备好的 StateRuntime 创建本地图存储对象。别人想通过 AgentGraphStore 接口读写本地数据库时，就从这里拿到 LocalAgentGraphStore。

数据流：进去的是一个共享的 StateRuntime，也就是已经初始化好的状态数据库运行对象 → 它把这个对象放进 LocalAgentGraphStore 里 → 出来的是一个可以被复制、共享使用的本地图存储实例。

调用关系：测试里的 local_store_upserts_and_lists_direct_children_with_status_filters、local_store_updates_edge_status 和 local_store_lists_descendants_breadth_first_with_status_filters 都先调用它创建 store，然后再验证各种读写行为。实际运行中，它通常会在数据库初始化完成后被创建。

调用图：被 3 处调用（local_store_lists_descendants_breadth_first_with_status_filters, local_store_updates_edge_status, local_store_upserts_and_lists_direct_children_with_status_filters）。

async fn upsert_thread_spawn_edge(
        &self,
        parent_thread_id: ThreadId,
        child_thread_id: ThreadId,
        status: ThreadSpawnEdgeStatus,
    ) -> AgentGraphStoreResult<()>

作用：这个函数记录或更新一条“父线程启动了子线程”的关系。upsert 的意思是：没有就插入，有了就更新，避免重复关系乱掉。

数据流：进去的是父线程 ID、子线程 ID、以及这条关系的状态，比如打开或关闭 → 它先用 to_state_status 把外层状态翻译成数据库层认识的状态，再交给 state_db 写入 → 成功时出来一个空结果，表示事情办完；失败时把底层错误转成 AgentGraphStoreError。

调用关系：这是 AgentGraphStore 接口的一部分，外部代码在新增或刷新线程父子关系时会用它。它自己不直接操作 SQL，而是把状态翻译后交给 StateRuntime 的 upsert_thread_spawn_edge。

调用图：调用 1 个内部函数（to_state_status）。

async fn set_thread_spawn_edge_status(
        &self,
        child_thread_id: ThreadId,
        status: ThreadSpawnEdgeStatus,
    ) -> AgentGraphStoreResult<()>

作用：这个函数只修改某个子线程对应父子关系的状态。比如一个子线程关系原来是打开的，现在要标成关闭，就用它。

数据流：进去的是子线程 ID 和新的关系状态 → 它用 to_state_status 把状态翻译成 StateRuntime 认识的版本，再让 state_db 按子线程 ID 更新记录 → 成功时没有额外数据返回；失败时包装成统一错误。

调用关系：这是 AgentGraphStore 接口提供的更新入口。测试 local_store_updates_edge_status 会先插入一条打开关系，再调用它改成关闭，随后用查询函数确认状态真的变了。

调用图：调用 1 个内部函数（to_state_status）。

async fn list_thread_spawn_children(
        &self,
        parent_thread_id: ThreadId,
        status_filter: Option<ThreadSpawnEdgeStatus>,
    ) -> AgentGraphStoreResult<Vec<ThreadId>>

作用：这个函数查询某个父线程的直接孩子。它还可以按状态过滤，只看打开的孩子，或只看关闭的孩子。

数据流：进去的是父线程 ID，以及一个可选的状态过滤条件 → 如果有过滤条件，它先用 to_state_status 翻译状态，再调用带状态筛选的数据库查询；如果没有过滤条件，就查询所有直接孩子 → 出来的是子线程 ID 列表，底层出错时转成统一错误。

调用关系：这是外部代码查看线程图第一层关系时用的入口。它根据有没有状态过滤，把活儿分给 StateRuntime 的不同查询方法；测试 local_store_upserts_and_lists_direct_children_with_status_filters 和 local_store_updates_edge_status 都用它确认结果。

调用图：调用 1 个内部函数（to_state_status）。

async fn list_thread_spawn_descendants(
        &self,
        root_thread_id: ThreadId,
        status_filter: Option<ThreadSpawnEdgeStatus>,
    ) -> AgentGraphStoreResult<Vec<ThreadId>>

作用：这个函数查询某个线程下面的所有后代，不只是直接孩子，还包括孙子、曾孙等。它也支持按关系状态过滤。

数据流：进去的是根线程 ID，以及一个可选状态过滤条件 → 有过滤条件时先翻译状态，再查符合状态的后代；没有过滤条件时查全部后代 → 出来的是后代线程 ID 列表，查询失败会变成统一错误返回。

调用关系：这是需要看完整线程树时用的入口。它把具体的递归或图遍历工作交给 StateRuntime；测试 local_store_lists_descendants_breadth_first_with_status_filters 会验证它能按预期列出后代，并支持状态过滤。

调用图：调用 1 个内部函数（to_state_status）。

fn to_state_status(status: ThreadSpawnEdgeStatus) -> codex_state::DirectionalThreadSpawnEdgeStatus

作用：这个小函数负责把 agent-graph-store 自己的状态枚举，翻译成 codex_state 数据库层使用的状态枚举。它让外层接口和底层存储可以各用各的类型，但意思保持一致。

数据流：进去的是 ThreadSpawnEdgeStatus，比如 Open 或 Closed → 它做一个一一对应的匹配：Open 变成数据库层 Open，Closed 变成数据库层 Closed → 出来的是 codex_state::DirectionalThreadSpawnEdgeStatus。

调用关系：所有需要把状态传给 StateRuntime 的函数都会用它，包括 upsert_thread_spawn_edge、set_thread_spawn_edge_status、list_thread_spawn_children 和 list_thread_spawn_descendants。它是这个文件里的“状态翻译表”。

调用图：被 4 处调用（list_thread_spawn_children, list_thread_spawn_descendants, set_thread_spawn_edge_status, upsert_thread_spawn_edge）。

fn internal_error(err: impl std::fmt::Display) -> AgentGraphStoreError

作用：这个函数把底层数据库或运行时错误，包装成 AgentGraphStore 自己统一使用的内部错误。这样外部调用者不用理解 StateRuntime 的各种错误细节。

数据流：进去的是任何能显示成文字的错误 → 它调用 to_string 把错误变成一段说明文字，再放进 AgentGraphStoreError::Internal 的 message 字段 → 出来的是统一格式的 AgentGraphStoreError。

调用关系：这个文件里的数据库调用失败时会用它做错误转换。它的位置像一个“错误海关”，把底层来的错误换成外层接口能识别的证件。

调用图：外部调用 1 个（to_string）。

fn thread_id(suffix: u128) -> ThreadId

作用：这个测试辅助函数用一个数字后缀造出稳定的 ThreadId。测试需要可预测的线程 ID，这样断言结果顺序时不会乱。

数据流：进去的是一个数字后缀 → 它把数字拼进固定格式的 UUID 字符串里，再用 ThreadId::from_string 解析 → 出来的是一个合法的 ThreadId；如果格式不对，测试会直接失败。

调用关系：三个测试函数都会用它准备父线程、子线程、孙线程等 ID。它不参与正式运行，只是让测试数据更清楚、更容易重复。

调用图：调用 1 个内部函数（from_string）；外部调用 1 个（format!）。

async fn state_runtime() -> TestRuntime

作用：这个测试辅助函数创建一个临时的本地状态数据库。每个测试都拿到一套干净环境，互相不会污染。

数据流：进去没有业务输入 → 它先创建临时目录，再用 StateRuntime::init 在这个目录里初始化测试数据库 → 出来的是 TestRuntime，里面有 state_db 和临时目录；临时目录被保留下来，保证测试期间数据库文件还在。

调用关系：三个异步测试开始时都会调用它。它为 LocalAgentGraphStore::new 准备底层 StateRuntime，让测试能真正读写 SQLite-backed 的状态存储。

调用图：调用 1 个内部函数（init）；外部调用 1 个（new）。

async fn local_store_upserts_and_lists_direct_children_with_status_filters()

作用：这个测试确认本地 store 能插入父子关系，并且能按状态查询直接孩子。它验证“全部孩子”“打开孩子”“关闭孩子”三种情况都对。

数据流：进去没有外部输入，测试自己创建临时数据库和几个线程 ID → 它插入两条父子关系，一条关闭、一条打开，然后分别查询全部、打开、关闭的直接孩子 → 出来是测试断言结果：顺序和过滤结果必须符合预期，否则测试失败。

调用关系：它调用 state_runtime 准备数据库，调用 LocalAgentGraphStore::new 创建 store，调用 thread_id 生成测试 ID。它主要覆盖 upsert_thread_spawn_edge 和 list_thread_spawn_children 的配合效果。

调用图：调用 1 个内部函数（new）；外部调用 3 个（state_runtime, thread_id, assert_eq!）。

async fn local_store_updates_edge_status()

作用：这个测试确认一条父子关系的状态可以从打开改成关闭。它防止出现“更新函数看似成功，但查询结果还是旧状态”的问题。

数据流：进去没有外部输入 → 它创建临时数据库和一对父子线程 ID，先插入打开关系，再调用 set_thread_spawn_edge_status 改成关闭 → 最后查询打开孩子应为空，查询关闭孩子应包含这个子线程。

调用关系：它调用 state_runtime、LocalAgentGraphStore::new 和 thread_id 来搭测试环境。它重点验证 set_thread_spawn_edge_status 和 list_thread_spawn_children 之间的一致性。

调用图：调用 1 个内部函数（new）；外部调用 3 个（state_runtime, thread_id, assert_eq!）。

async fn local_store_lists_descendants_breadth_first_with_status_filters()

作用：这个测试确认 store 能列出一个线程下面所有层级的后代，并且能按状态过滤。它还检查返回顺序符合预期的广度优先顺序，也就是先看近的孩子，再看更深的后代。

数据流：进去没有外部输入 → 它创建一棵小的线程关系树，里面有打开和关闭的边，再查询全部后代、打开后代、关闭后代 → 出来是多组断言，确认列表内容、过滤结果和顺序都正确。

调用关系：它调用 state_runtime 创建数据库，调用 LocalAgentGraphStore::new 创建 store，调用 thread_id 生成多个节点。它主要覆盖 upsert_thread_spawn_edge 和 list_thread_spawn_descendants 的组合行为，并和 StateRuntime 的带状态查询结果做对照。

调用图：调用 1 个内部函数（new）；外部调用 3 个（state_runtime, thread_id, assert_eq!）。