GenarrativeAI/Genarrative
10
0
Fork 0
Code Issues 22 Pull Requests 5 Actions Packages Projects 1 Releases Wiki Activity

宿主：首页「运行中的项目」改由 Thread Manager 的逻辑回合导出

Browse Source 
活动回合表的唯一事实源从"调用身份守卫"搬进逻辑回合占用，任务侧不再另建一张表。
- `ActiveDirectTurn` 带上快照字段（回合身份 / 项目名 / 起点 / 状态 / 活动 / 序号），
  接单时初始化，收口时随占用一起消失
- 新增 `update_direct_thread_active_turn`（进度回填，只认身份一致且序号不倒退）与
  `list_direct_active_turns`（只导出仍有未收口回合的 thread）
- `DirectActiveTurnSnapshot` 移进 `direct_thread_manager`，`projectPath` 用线程身份，
  与事件流里的项目身份是同一个字符串
- `DirectTaonierActiveInvocation` 退回纯单飞锁：只留调用身份与登记时刻
- 回合更新发射器不再按项目路径 canonicalize 找表，改为持线程身份回填
- `DirectTurnReservation::accept` 多带一个 `clientTurnId`（快照与进度匹配用），
  与占用 token 是两个身份
- 上下文身份的两个测试补上"逻辑回合也接单"这一步：身份来自接单，不是调用守卫
This commit is contained in:
王德宇
2026-09-23 21:39:01 +08:00
parent d31a758c9c
commit af5fdf8a0e
7 changed files with 237 additions and 128 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/ai-game-creator-shell/src-tauri/src/agent/codex_app_server/mod.rs
+1
View File
@@ -7791,6 +7791,7 @@ done
// 这里补上同一步，于是这一轮的边界仍在同一个订阅里成对出现，历史里也有那条用户消息。
let _reservation = crate::agent::direct_turn_accept::DirectTurnReservation::accept(
&thread_id,
"turn-0001",
Some("direct-codex:turn-0001:user"),
)
.expect("accept logical turn");
apps/ai-game-creator-shell/src-tauri/src/agent/direct_project_context.rs
+22 -2
View File
@@ -435,8 +435,14 @@ mod tests {
async fn a_replaced_active_turn_marks_the_batch_stale() {
let (_temp, root) = project();
std::fs::write(root.join("code.js"), "unchanged").unwrap();
// 身份来自逻辑回合（Thread Manager）：接单才是"这一轮在跑"的唯一登记。
let owner = Arc::new(std::sync::Mutex::new(Some(
DirectTaonierActiveInvocationGuard::enter(&root, "turn-before").unwrap(),
DirectTurnReservation::accept(
&direct_thread_id_for_project(&root),
"turn-before",
None,
)
.unwrap(),
)));
let swap = Arc::clone(&owner);
let result = read_batch_with(
@@ -447,7 +453,14 @@ mod tests {
let result = read_file(r, f, b);
let mut guard = swap.lock().unwrap();
drop(guard.take());
*guard = Some(DirectTaonierActiveInvocationGuard::enter(r, "turn-after").unwrap());
*guard = Some(
DirectTurnReservation::accept(
&direct_thread_id_for_project(r),
"turn-after",
None,
)
.unwrap(),
);
result
},
)
@@ -581,7 +594,14 @@ mod tests {
#[tokio::test]
async fn host_prefetch_keeps_data_out_of_system_rules_and_matches_active_turn() {
let (_temp, root) = project();
// 调用身份（预取闸门）与逻辑回合（上下文身份）是两件事，生产入口两步都做。
let _guard = DirectTaonierActiveInvocationGuard::enter(&root, "prefetch-turn").unwrap();
let _turn = DirectTurnReservation::accept(
&direct_thread_id_for_project(&root),
"prefetch-turn",
None,
)
.unwrap();
let data = prefetch_turn_input(&root, "prefetch-turn")
.await
.unwrap()
apps/ai-game-creator-shell/src-tauri/src/agent/direct_runtime/mod.rs
-107
View File
@@ -472,25 +472,7 @@ fn direct_taonier_regeneration_invocation_sha256(invocation_id: &str) -> String
#[derive(Debug)]
struct DirectTaonierActiveInvocation {
invocation_id: String,
project_name: Option<String>,
started_at: u64,
status: String,
activity: Option<String>,
updated_at: u64,
sequence: u64,
}
#[derive(Clone, Debug, serde::Serialize)]
#[serde(rename_all = "camelCase")]
pub(crate) struct DirectActiveTurnSnapshot {
pub(crate) project_path: String,
pub(crate) project_name: Option<String>,
pub(crate) turn_id: String,
pub(crate) started_at: u64,
pub(crate) status: String,
pub(crate) activity: Option<String>,
pub(crate) updated_at: u64,
pub(crate) sequence: u64,
}
static DIRECT_TAONIER_ACTIVE_INVOCATIONS: OnceLock<
@@ -542,15 +524,7 @@ impl DirectTaonierActiveInvocationGuard {
root.clone(),
DirectTaonierActiveInvocation {
invocation_id: invocation_id.to_string(),
project_name: root
.file_name()
.and_then(|name| name.to_str())
.map(str::to_string),
started_at,
status: "accepted".to_string(),
activity: Some("request-accepted".to_string()),
updated_at: started_at,
sequence: 0,
},
);
}
@@ -579,57 +553,6 @@ impl Drop for DirectTaonierActiveInvocationGuard {
}
}
pub(crate) fn list_direct_active_turns() -> Result<Vec<DirectActiveTurnSnapshot>, String> {
let active = DIRECT_TAONIER_ACTIVE_INVOCATIONS
.get_or_init(|| Mutex::new(HashMap::new()))
.lock()
.map_err(|_| "Direct 调用身份锁已损坏".to_string())?;
let mut turns = active
.iter()
.map(|(root, invocation)| DirectActiveTurnSnapshot {
project_path: root.to_string_lossy().into_owned(),
project_name: invocation.project_name.clone(),
turn_id: invocation.invocation_id.clone(),
started_at: invocation.started_at,
status: invocation.status.clone(),
activity: invocation.activity.clone(),
updated_at: invocation.updated_at,
sequence: invocation.sequence,
})
.collect::<Vec<_>>();
turns.sort_by(|left, right| left.project_path.cmp(&right.project_path));
Ok(turns)
}
pub(crate) fn update_direct_active_turn(
root: &Path,
turn_id: &str,
status: &str,
activity: Option<&str>,
sequence: u64,
updated_at: u64,
) {
let Ok(root) = root.canonicalize() else {
return;
};
let Some(active) = DIRECT_TAONIER_ACTIVE_INVOCATIONS.get() else {
return;
};
let Ok(mut active) = active.lock() else {
return;
};
let Some(invocation) = active.get_mut(&root) else {
return;
};
if invocation.invocation_id != turn_id || sequence < invocation.sequence {
return;
}
invocation.status = status.to_string();
invocation.activity = activity.map(str::to_string);
invocation.updated_at = updated_at;
invocation.sequence = sequence;
}
pub(crate) fn direct_taonier_active_invocation_id_at(root: &Path) -> Result<String, String> {
let root = root
.canonicalize()
@@ -5972,36 +5895,6 @@ mod tests {
entry.started_at = entry.started_at.saturating_sub(age_ms);
}
#[test]
fn active_turn_snapshot_tracks_progress_and_is_removed_after_drop() {
let root = tempfile::tempdir().expect("active snapshot root");
let turn_id = "client-turn-snapshot-0001";
let guard = DirectTaonierActiveInvocationGuard::enter(root.path(), turn_id)
.expect("active snapshot turn");
update_direct_active_turn(
root.path(),
turn_id,
"streaming",
Some("response-finalization"),
3,
42,
);
let snapshot = list_direct_active_turns()
.expect("list active turns")
.into_iter()
.find(|turn| turn.turn_id == turn_id)
.expect("snapshot entry");
assert_eq!(snapshot.status, "streaming");
assert_eq!(snapshot.activity.as_deref(), Some("response-finalization"));
assert_eq!(snapshot.sequence, 3);
assert_eq!(snapshot.updated_at, 42);
drop(guard);
assert!(list_direct_active_turns()
.expect("list after completion")
.into_iter()
.all(|turn| turn.turn_id != turn_id));
}
#[test]
fn direct_success_reply_is_persisted_once_with_the_stable_client_turn_identity() {
let root = tempfile::tempdir().expect("temp dir");
apps/ai-game-creator-shell/src-tauri/src/agent/direct_runtime/user_input.rs
+2 -2
View File
@@ -106,7 +106,7 @@ async fn chat_with_game_creator_direct_codex_typed(
// ——开始事件发生在用户条目落盘之前，而落盘本身也可能失败。
let thread_id = direct_thread_id_for_project(root);
let user_item_id = direct_codex_user_item_id_for_client_turn_id(&turn_id);
let reservation = DirectTurnReservation::accept(&thread_id, user_item_id.as_deref())?;
let reservation = DirectTurnReservation::accept(&thread_id, &turn_id, user_item_id.as_deref())?;
// 落盘即接单：接单成功就必须在历史里留下这条用户消息，哪怕这一轮随后失败。
if let Err(error) = append_direct_project_user_message_at(root, &canonical_user_item) {
let failure = DirectTurnError::EnvironmentNotReady {
@@ -206,7 +206,7 @@ mod tests {
let subscription = subscribe_direct_thread(&thread_id);
let _ = consume_direct_thread(&subscription.subscription_id);
let reservation =
DirectTurnReservation::accept(&thread_id, Some("direct-codex:turn-1:user"))
DirectTurnReservation::accept(&thread_id, "turn-1", Some("direct-codex:turn-1:user"))
.expect("accept logical turn");
let invocation =
DirectTaonierActiveInvocationGuard::enter(&root, "turn-1").expect("enter invocation");
apps/ai-game-creator-shell/src-tauri/src/agent/direct_thread_manager.rs
+181 -3
View File
@@ -37,11 +37,42 @@ struct SubscriberState {
/// 一条正在跑的逻辑回合的占用：接单时登记，终态写出时解除。
///
/// `token` 是这一次接单的稳定身份：终态出口只有拿着同一个 token 的占用对象才能写兜底终态，
/// 避免迟到的旧占用把新回合的边界顶掉。
/// 它同时是首页「运行中的项目」快照的**唯一事实源**（[`list_direct_active_turns`]）：这一格的
/// 生命周期就是"这一轮在不在跑"，进度字段由运行时那一侧经 [`update_direct_thread_active_turn`]
/// 回填。任务侧不再另建一张活动回合表——同一件事只许有一处真相。
///
/// 两个身份别混：
/// - `token` 是这一次接单的占用身份：终态出口只有拿着同一个 token 的占用对象才能写兜底终态，
/// 避免迟到的旧占用把新回合的边界顶掉。它不对外。
/// - `turn_id` 是给界面看的回合身份（`clientTurnId` 派生），只服务快照与进度回填的匹配。
#[derive(Clone, Debug)]
struct ActiveDirectTurn {
token: String,
turn_id: String,
project_name: Option<String>,
started_at: u64,
status: String,
activity: Option<String>,
updated_at: u64,
sequence: u64,
}
/// 首页「运行中的项目」的一条快照。
///
/// `project_path` 与线上其它地方的项目身份取同一个字符串：Thread Manager 的线程身份就是项目的
/// canonical 路径（见 `direct_thread_id_for_project`），所以快照里的项目身份与事件流里的身份
/// 永远能对上，不需要调用方再做一次归一。
#[derive(Clone, Debug, serde::Serialize)]
#[serde(rename_all = "camelCase")]
pub(crate) struct DirectActiveTurnSnapshot {
pub(crate) project_path: String,
pub(crate) project_name: Option<String>,
pub(crate) turn_id: String,
pub(crate) started_at: u64,
pub(crate) status: String,
pub(crate) activity: Option<String>,
pub(crate) updated_at: u64,
pub(crate) sequence: u64,
}
#[derive(Clone, Debug)]
@@ -175,6 +206,7 @@ impl DirectThreadManager {
&mut self,
thread_id: &str,
token: &str,
turn_id: &str,
user_item_id: Option<&str>,
started_at_ms: u64,
) -> Result<DirectThreadEvent, String> {
@@ -185,6 +217,17 @@ impl DirectThreadManager {
}
thread.active_turn = Some(ActiveDirectTurn {
token: token.to_string(),
turn_id: turn_id.to_string(),
project_name: std::path::Path::new(thread_id)
.file_name()
.and_then(|name| name.to_str())
.map(str::to_string),
started_at: started_at_ms,
// 与"还没有任何进度事件"的状态一致：运行时给出的第一条进度会覆盖它。
status: "accepted".to_string(),
activity: Some("request-accepted".to_string()),
updated_at: started_at_ms,
sequence: 0,
});
}
Ok(self.append(
@@ -193,6 +236,56 @@ impl DirectThreadManager {
))
}
/// 运行时回填这一轮的进度。只认"仍在跑 + 回合身份一致 + 序号不倒退"的那一次。
///
/// 返回是否真的写进去了：没有未收口的回合、身份对不上（上一轮迟到的进度）、序号倒退
/// （乱序到达的旧进度）都必须原地丢弃，不能把快照改成过期的样子。
fn update_active_turn(
&mut self,
thread_id: &str,
turn_id: &str,
status: &str,
activity: Option<&str>,
sequence: u64,
updated_at: u64,
) -> bool {
let Some(active) = self
.threads
.get_mut(thread_id)
.and_then(|thread| thread.active_turn.as_mut())
else {
return false;
};
if active.turn_id != turn_id || sequence < active.sequence {
return false;
}
active.status = status.to_string();
active.activity = activity.map(str::to_string);
active.updated_at = updated_at;
active.sequence = sequence;
true
}
/// 首页快照：只导出仍有未收口逻辑回合的 thread。
fn active_turn_snapshots(&self) -> Vec<DirectActiveTurnSnapshot> {
self.threads
.iter()
.filter_map(|(thread_id, thread)| {
let active = thread.active_turn.as_ref()?;
Some(DirectActiveTurnSnapshot {
project_path: thread_id.clone(),
project_name: active.project_name.clone(),
turn_id: active.turn_id.clone(),
started_at: active.started_at,
status: active.status.clone(),
activity: active.activity.clone(),
updated_at: active.updated_at,
sequence: active.sequence,
})
})
.collect()
}
/// 深层的终态出口：解除占用并写下 `turn.completed`。
///
/// 不校验 token：这一条由真正跑完这一轮的代码调用，终态就是它算出来的那个（CLI 这类没有
@@ -472,6 +565,7 @@ pub(crate) fn append_direct_thread_event(
pub(crate) fn accept_direct_thread_turn(
thread_id: &str,
token: &str,
turn_id: &str,
user_item_id: Option<&str>,
started_at_ms: u64,
) -> Result<(), String> {
@@ -479,12 +573,37 @@ pub(crate) fn accept_direct_thread_turn(
global_direct_thread_manager()
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner())
.accept_turn(thread_id, token, user_item_id, started_at_ms)?;
.accept_turn(thread_id, token, turn_id, user_item_id, started_at_ms)?;
}
notify_direct_thread_subscribers(thread_id);
Ok(())
}
/// 运行时回填某一轮逻辑回合的进度（状态 / 活动 / 序号）。返回是否真的写进去了。
pub(crate) fn update_direct_thread_active_turn(
thread_id: &str,
turn_id: &str,
status: &str,
activity: Option<&str>,
sequence: u64,
updated_at: u64,
) -> bool {
global_direct_thread_manager()
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner())
.update_active_turn(thread_id, turn_id, status, activity, sequence, updated_at)
}
/// 首页「运行中的项目」快照：逻辑回合的唯一导出口（见 [`DirectActiveTurnSnapshot`]）。
pub(crate) fn list_direct_active_turns() -> Result<Vec<DirectActiveTurnSnapshot>, String> {
let mut turns = global_direct_thread_manager()
.lock()
.map_err(|_| "Direct 线程管理器已损坏".to_string())?
.active_turn_snapshots();
turns.sort_by(|left, right| left.project_path.cmp(&right.project_path));
Ok(turns)
}
/// 深层终态出口：解除占用并写 `turn.completed`。
pub(crate) fn complete_direct_thread_turn(thread_id: &str, event: DirectThreadEvent) {
{
@@ -890,4 +1009,63 @@ mod tests {
"unfinished item at queue head blocks middle cleanup"
);
}
/// 首页快照就是逻辑回合的导出：接单即出现、进度按序号回填、收口即消失。
fn snapshot_of(
manager: &DirectThreadManager,
thread_id: &str,
) -> Option<DirectActiveTurnSnapshot> {
manager
.active_turn_snapshots()
.into_iter()
.find(|turn| turn.project_path == thread_id)
}
#[test]
fn active_turn_snapshot_follows_the_logical_turn_lifecycle() {
let mut manager = DirectThreadManager::with_limits(100, 100_000);
let thread_id = "/tmp/快照项目";
assert!(snapshot_of(&manager, thread_id).is_none());
manager
.accept_turn(thread_id, "token-1", "turn-1", Some("u-1"), FIXED_AT_MS)
.expect("accept");
let accepted = snapshot_of(&manager, thread_id).expect("accepted turn is visible");
assert_eq!(accepted.turn_id, "turn-1");
assert_eq!(accepted.project_name.as_deref(), Some("快照项目"));
assert_eq!(accepted.started_at, FIXED_AT_MS);
assert_eq!(accepted.status, "accepted");
assert_eq!(accepted.activity.as_deref(), Some("request-accepted"));
assert_eq!(accepted.sequence, 0);
assert!(manager.update_active_turn(
thread_id,
"turn-1",
"streaming",
Some("file-write"),
3,
42,
));
let running = snapshot_of(&manager, thread_id).expect("running turn is visible");
assert_eq!(running.status, "streaming");
assert_eq!(running.activity.as_deref(), Some("file-write"));
assert_eq!(running.sequence, 3);
assert_eq!(running.updated_at, 42);
// 序号倒退与身份对不上的进度都不许改快照。
assert!(!manager.update_active_turn(thread_id, "turn-1", "failed", None, 2, 99));
assert!(!manager.update_active_turn(thread_id, "turn-2", "failed", None, 4, 99));
assert_eq!(
snapshot_of(&manager, thread_id)
.expect("snapshot unchanged")
.status,
"streaming"
);
manager.complete_turn(
thread_id,
DirectThreadEvent::turn_completed("completed".to_string(), 5_000),
);
assert!(snapshot_of(&manager, thread_id).is_none());
}
}
apps/ai-game-creator-shell/src-tauri/src/agent/direct_turn_accept.rs
+26 -12
View File
@@ -30,16 +30,25 @@ impl DirectTurnReservation {
/// 接单：登记占用并发出逻辑回合开始事件。
///
/// 失败表示这个 thread 已经有一条没收口的回合（并发接单），此时不改队列、不发事件。
/// `client_turn_id` 是给界面看的回合身份（首页快照与进度回填按它匹配），与占用身份 `token`
/// 是两件事：前者来自调用方，后者只活在这个进程里。
pub(crate) fn accept(
thread_id: &str,
client_turn_id: &str,
user_item_id: Option<&str>,
) -> Result<Self, DirectTurnError> {
let token = Uuid::new_v4().to_string();
accept_direct_thread_turn(thread_id, &token, user_item_id, direct_tool_call_now_ms())
.map_err(|existing| DirectTurnError::TurnAlreadyRunning {
existing_invocation_id: existing,
incoming_invocation_id: token.clone(),
})?;
accept_direct_thread_turn(
thread_id,
&token,
client_turn_id,
user_item_id,
direct_tool_call_now_ms(),
)
.map_err(|existing| DirectTurnError::TurnAlreadyRunning {
existing_invocation_id: existing,
incoming_invocation_id: token.clone(),
})?;
Ok(Self {
thread_id: thread_id.to_string(),
token,
@@ -108,7 +117,8 @@ mod tests {
let thread = unique_thread("started");
let subscription = watch(&thread);
let reservation = DirectTurnReservation::accept(&thread, Some("u-1")).expect("accept");
let reservation =
DirectTurnReservation::accept(&thread, "turn-1", Some("u-1")).expect("accept");
let events = pending(&subscription);
assert_eq!(events.len(), 1, "{events:?}");
@@ -126,11 +136,12 @@ mod tests {
fn a_second_accept_is_rejected_while_the_turn_is_open() {
let thread = unique_thread("busy");
let subscription = watch(&thread);
let reservation = DirectTurnReservation::accept(&thread, Some("u-1")).expect("accept");
let reservation =
DirectTurnReservation::accept(&thread, "turn-1", Some("u-1")).expect("accept");
// 先取走第一条接单自己的开始事件，之后的"空"才只说明被拒的这一次没写东西。
assert_eq!(pending(&subscription).len(), 1);
let rejected = DirectTurnReservation::accept(&thread, Some("u-2"));
let rejected = DirectTurnReservation::accept(&thread, "turn-2", Some("u-2"));
assert!(matches!(
rejected,
@@ -145,7 +156,8 @@ mod tests {
fn drop_without_a_terminal_writes_a_host_dropped_terminal() {
let thread = unique_thread("drop");
let subscription = watch(&thread);
let reservation = DirectTurnReservation::accept(&thread, Some("u-1")).expect("accept");
let reservation =
DirectTurnReservation::accept(&thread, "turn-1", Some("u-1")).expect("accept");
assert!(reservation.finish_if_unfinished(DirectTurnTerminal::host_dropped()));
assert!(!direct_thread_turn_is_active(&thread));
@@ -172,7 +184,8 @@ mod tests {
fn the_deep_terminal_wins_and_the_fallback_stays_silent() {
let thread = unique_thread("deep");
let subscription = watch(&thread);
let reservation = DirectTurnReservation::accept(&thread, Some("u-1")).expect("accept");
let reservation =
DirectTurnReservation::accept(&thread, "turn-1", Some("u-1")).expect("accept");
// 深层收口：真正跑完这一轮的代码算出来的终态。
let deep = DirectThreadEvent::turn_completed_failed(
@@ -202,10 +215,11 @@ mod tests {
#[test]
fn the_thread_can_be_accepted_again_after_the_turn_is_settled() {
let thread = unique_thread("again");
let first = DirectTurnReservation::accept(&thread, Some("u-1")).expect("accept");
let first = DirectTurnReservation::accept(&thread, "turn-1", Some("u-1")).expect("accept");
drop(first);
let second = DirectTurnReservation::accept(&thread, Some("u-2")).expect("second accept");
let second =
DirectTurnReservation::accept(&thread, "turn-2", Some("u-2")).expect("second accept");
assert!(direct_thread_turn_is_active(&thread));
drop(second);
apps/ai-game-creator-shell/src-tauri/src/agent/runtime_driver/entrypoints.rs
+5 -2
View File
@@ -32,6 +32,8 @@ pub(crate) fn emit_direct_game_creator_progress(root: &Path, stage: &str, messag
#[derive(Clone)]
pub(crate) struct DirectGameCreatorTurnUpdateEmitter {
project_path: String,
/// Thread Manager 的线程身份：进度只回填到"这一轮仍被占用"的那一格上。
thread_id: String,
turn_id: String,
sequence: Arc<AtomicU64>,
}
@@ -40,6 +42,7 @@ impl DirectGameCreatorTurnUpdateEmitter {
pub(crate) fn new(root: &Path, turn_id: String) -> Self {
Self {
project_path: root.to_string_lossy().into_owned(),
thread_id: crate::agent::direct_thread_id_for_project(root),
turn_id,
sequence: Arc::new(AtomicU64::new(0)),
}
@@ -136,8 +139,8 @@ impl DirectGameCreatorTurnUpdateEmitter {
.unwrap_or_default()
.as_millis()
.min(u64::MAX as u128) as u64;
update_direct_active_turn(
Path::new(&self.project_path),
update_direct_thread_active_turn(
&self.thread_id,
&self.turn_id,
status,
activity,