共享的 project.jsonl 让两条对话链互读对方的行,混合历史不再双向失败关闭

- project/conversation.rs:read_persisted_local_conversation_records_unlocked 跳过 type=response_item 且带 payload 的行(DirectProject 写给同一份「项目主对话」的行),其余解析失败仍按「解析对话记录失败」失败关闭
- 新增 is_direct_project_history_row 做行信封白名单,只认对方那一种明确枚举的形状,不放宽成「什么都跳过」
- 不把 response_item 行二次投影成 LocalConversationMessageRecord:DirectProject 侧已经拥有那份 Responses item → 聊天内容的投影,再造一份平行投影会与它漂移
- 不加「文件已属于 DirectProject 就拒绝追加旧行」的硬报错:runtime_driver/task_start.rs 在 ensure_..._accepted_public_status_at 返回 Err 时会中止后台任务(「后台任务启动确认落盘失败,任务未执行」),runtime_protocol/steering.rs:572/604/1300 也用 ? 上抛,加硬报错会把旧行噪声换成任务起不来;毒化本身已由读侧白名单堵住
- 补断言:混合文件里通用链只读自己的 legacy 行且坏行仍失败关闭、混合文件从两条链都能读(DirectProject 把两种行都读出来)、纯旧格式文件的写入行为不变
This commit is contained in:
2026-09-11 18:36:07 +08:00
parent bfb928295e
commit f0d5687ab7
2 changed files with 133 additions and 2 deletions
@@ -958,8 +958,19 @@ fn read_persisted_local_conversation_records_unlocked(
if line.is_empty() {
continue;
}
let record = serde_json::from_str::<PersistedLocalConversationMessageRecord>(line)
.map_err(|error| format!("解析对话记录失败:{}: {error}", path.display()))?;
let record = match serde_json::from_str::<PersistedLocalConversationMessageRecord>(
line,
) {
Ok(record) => record,
// 项目主对话(agent_id=None)与 DirectProject 共用同一份
// `project.jsonl`DirectProject 写的是 Codex rollout 的
// `response_item` 行。通用对话链只认自己的 legacy 行,遇到对方那种
// 行就跳过;其余任何坏行继续失败关闭。
Err(_) if is_direct_project_history_row(line) => continue,
Err(error) => {
return Err(format!("解析对话记录失败:{}: {error}", path.display()))
}
};
records.push(record);
}
}
@@ -969,6 +980,15 @@ fn read_persisted_local_conversation_records_unlocked(
Ok(records)
}
/// 只认 DirectProject 那一种明确枚举的行信封:`type=response_item` 且带 `payload`。
/// 其余任何形状都不算“对方的行”,仍由上方按坏行失败关闭。
fn is_direct_project_history_row(line: &str) -> bool {
serde_json::from_str::<serde_json::Value>(line).is_ok_and(|parsed| {
parsed.get("type").and_then(serde_json::Value::as_str) == Some("response_item")
&& parsed.get("payload").is_some()
})
}
fn persisted_local_conversation_message_index_by_id(
records: &[PersistedLocalConversationMessageRecord],
expected_agent_id: Option<&str>,
@@ -225,3 +225,114 @@ fn finalization_conversation_retry_rejects_conflicting_audit_payload() {
std::fs::remove_dir_all(root).ok();
}
const DIRECT_PROJECT_ROW: &str = r#"{"type":"response_item","payload":{"type":"message","role":"user","id":"codex-item-1","content":[{"type":"input_text","text":"再加一个按钮"}]}}"#;
const LEGACY_PROJECT_ROW: &str = r#"{"schemaVersion":"game-creator-conversation.v1","role":"assistant","content":"已创建菜单","agentId":null,"updatedAt":1757000000}"#;
fn project_conversation_path(root: &Path) -> PathBuf {
root.join(".agent/conversations/project.jsonl")
}
fn write_project_history(root: &Path, lines: &[&str]) {
let path = project_conversation_path(root);
std::fs::create_dir_all(path.parent().expect("history parent")).expect("history dir");
std::fs::write(&path, format!("{}\n", lines.join("\n"))).expect("write project history");
}
fn append_project_message(root: &Path, content: &str) -> Result<LocalConversationResult, String> {
append_local_conversation_message_at(
root,
None,
LocalConversationMessage {
role: "assistant".to_string(),
content: content.to_string(),
agent_id: None,
},
)
}
#[test]
fn project_history_skips_direct_project_rows_but_keeps_other_broken_rows_failing_closed() {
let root = unique_conversation_test_root();
init_local_game_project_at(&root, "project-1", "混合行形状读取测试").expect("init project");
write_project_history(&root, &[DIRECT_PROJECT_ROW, LEGACY_PROJECT_ROW]);
let conversation =
read_local_conversation_at(&root, None).expect("read mixed project conversation");
assert_eq!(conversation.messages.len(), 1);
assert_eq!(conversation.messages[0].role, "assistant");
assert_eq!(conversation.messages[0].content, "已创建菜单");
write_project_history(&root, &[LEGACY_PROJECT_ROW, r#"{"unexpected":true}"#]);
let error = read_local_conversation_at(&root, None)
.expect_err("unrecognized row must keep failing closed");
assert!(error.starts_with("解析对话记录失败"), "{error}");
std::fs::remove_dir_all(root).ok();
}
#[test]
fn mixed_project_history_rows_stay_readable_from_both_sides() {
let root = unique_conversation_test_root();
init_local_game_project_at(&root, "project-1", "写侧统一测试").expect("init project");
write_project_history(&root, &[DIRECT_PROJECT_ROW]);
append_project_message(&root, "模式切换后由通用写入器补写的回复").expect("append legacy row");
append_local_conversation_message_for_session_idempotent_at(
&root,
None,
None,
LocalConversationMessage {
role: "assistant".to_string(),
content: "带 id 的旧格式回复".to_string(),
agent_id: None,
},
"direct-codex:turn-0002:assistant",
)
.expect("append id-bearing legacy row");
let history = std::fs::read_to_string(project_conversation_path(&root)).expect("read history");
assert_eq!(history.lines().count(), 3, "{history}");
assert!(history.contains("模式切换后由通用写入器补写的回复"), "{history}");
// 通用对话链只读自己的 legacy 行,跳过 DirectProject 的行。
let conversation = read_local_conversation_at(&root, None).expect("read project conversation");
assert_eq!(
conversation
.messages
.iter()
.map(|message| message.content.as_str())
.collect::<Vec<_>>(),
vec!["模式切换后由通用写入器补写的回复", "带 id 的旧格式回复"]
);
// 反向:DirectProject 链把同一份文件里的两种行都读出来,混合文件不构成毒化。
let direct_items = crate::agent::read_direct_project_history_items_at(&root)
.expect("DirectProject must keep reading the mixed history");
assert_eq!(
direct_items
.iter()
.map(|item| item["content"][0]["text"].as_str().unwrap_or_default())
.collect::<Vec<_>>(),
vec![
"再加一个按钮",
"模式切换后由通用写入器补写的回复",
"带 id 的旧格式回复",
]
);
std::fs::remove_dir_all(root).ok();
}
#[test]
fn project_history_keeps_accepting_legacy_rows_while_no_direct_project_row_exists() {
let root = unique_conversation_test_root();
init_local_game_project_at(&root, "project-1", "纯旧格式项目写入测试").expect("init project");
append_project_message(&root, "非 direct 模式的项目对话回复").expect("append legacy row");
let conversation = read_local_conversation_at(&root, None).expect("read legacy conversation");
assert_eq!(conversation.messages.len(), 1);
assert_eq!(conversation.messages[0].content, "非 direct 模式的项目对话回复");
std::fs::remove_dir_all(root).ok();
}