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

保证混合协作回执完整观察

Browse Source 
为静态回执和隔离 join 分配独立预算，拒绝静默截断

按精确 delegationId 集合推进 claim 观察并收紧恢复锁边界

补充 V1.36 回归测试与技术决策文档
This commit is contained in:
AIGameCreator App
2026-07-18 13:10:53 +08:00
parent 6299faa81e
commit 27a35c2e2c
5 changed files with 1141 additions and 65 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/ai-game-creator-shell/src-tauri/src/agent.rs
+256 -33
View File
@@ -9113,6 +9113,8 @@ pub(crate) const AGENT_RUNTIME_PLAN_STATUS_COMPLETED: &str = "completed";
const AGENT_RUNTIME_PLAN_STATUS_FAILED: &str = "failed";
const AGENT_RUNTIME_TOOL_OBSERVATION_MAX_CHARS: usize = 900;
const AGENT_RUNTIME_RUN_STATUS_OBSERVATION_MAX_CHARS: usize = 16_000;
const AGENT_RUNTIME_RUN_STATUS_BASE_DETAIL_MAX_CHARS: usize = 3_500;
const AGENT_RUNTIME_READY_ISOLATED_JOIN_MIXED_PAYLOAD_MAX_CHARS: usize = 6_000;
const AGENT_RUNTIME_READY_ISOLATED_JOIN_PAYLOAD_MAX_CHARS: usize = 10_000;
const AGENT_RUNTIME_DELEGATE_CONTRACT_OBSERVATION_MAX_CHARS: usize = 20_000;
const AGENT_RUNTIME_COMMAND_OUTPUT_CONTEXT_MAX_CHARS: usize = 64_000;
@@ -15682,14 +15684,17 @@ fn mark_supervisor_delivery_claims_observed_for_pending_action_at(
if pending.action.tool != "agent.run_status" || observation.status != "ok" {
return Ok(());
}
let observed_delegate_receipt_ids =
observed_delegate_receipt_ids_from_run_status(observation.detail.as_deref())?;
let observed_isolated_groups =
observed_isolated_join_group_ids_from_run_status(observation.detail.as_deref())?;
if pending.agent_id == GAME_CREATOR_PROJECT_SUPERVISOR_AGENT_ID {
mark_static_delegate_claim_observed_at(
mark_static_delegate_claim_observed_for_receipts_at(
root,
&pending.agent_id,
&pending.run_id,
&pending.action_id,
&observed_delegate_receipt_ids,
)?;
}
mark_unobserved_isolated_join_claims_for_parent_at(
@@ -15701,21 +15706,85 @@ fn mark_supervisor_delivery_claims_observed_for_pending_action_at(
Ok(())
}
fn run_status_prefixed_payload<'a>(
detail: Option<&'a str>,
prefix: &str,
) -> Result<Option<&'a str>, String> {
let Some(mut remaining) = detail else {
return Ok(None);
};
let mut found = None;
loop {
let (block, next) = remaining
.split_once("\n\n")
.map_or((remaining, ""), |(block, next)| (block, next));
if let Some(payload) = block.strip_prefix(prefix) {
if found.replace(payload).is_some() {
return Err(format!(
"agent.run_status observation 含重复前置区块：{}",
prefix.trim_end()
));
}
}
if !matches!(
block.split_once(':').map(|(name, _)| name),
Some(
"readyIsolatedJoins"
| "readyDelegateReceipts"
| "claimedDelegateContracts"
| "claimedIsolatedJoins"
)
) {
return Ok(found);
}
if next.is_empty() {
return Ok(found);
}
remaining = next;
}
}
fn observed_delegate_receipt_ids_from_run_status(
detail: Option<&str>,
) -> Result<BTreeSet<String>, String> {
let Some(payload) = run_status_prefixed_payload(detail, "readyDelegateReceipts: ")? else {
return Ok(BTreeSet::new());
};
let payload = serde_json::from_str::<serde_json::Value>(payload)
.map_err(|error| format!("解析已观察专业 Agent ready receipts 失败：{error}"))?;
if payload.get("ready").and_then(serde_json::Value::as_bool) != Some(true) {
return Err("已观察专业 Agent ready receipts 结果未标记 ready=true".to_string());
}
let receipts = payload
.get("receipts")
.and_then(serde_json::Value::as_array)
.ok_or_else(|| "已观察专业 Agent ready receipts 结果缺少 receipts".to_string())?;
let mut delegation_ids = BTreeSet::new();
for receipt in receipts {
let delegation_id = receipt
.get("delegationId")
.and_then(serde_json::Value::as_str)
.ok_or_else(|| "已观察专业 Agent ready receipts 结果缺少 delegationId".to_string())?;
if !delegation_ids.insert(delegation_id.to_string()) {
return Err(format!(
"已观察专业 Agent ready receipts 结果含重复 delegationId：{delegation_id}"
));
}
}
Ok(delegation_ids)
}
fn observed_isolated_join_group_ids_from_run_status(
detail: Option<&str>,
) -> Result<BTreeSet<String>, String> {
let Some(payload) = detail
.and_then(|detail| detail.strip_prefix("readyIsolatedJoins: "))
.map(|detail| {
detail
.split_once("\n\n")
.map_or(detail, |(payload, _)| payload)
})
else {
let Some(payload) = run_status_prefixed_payload(detail, "readyIsolatedJoins: ")? else {
return Ok(BTreeSet::new());
};
let payload = serde_json::from_str::<serde_json::Value>(payload)
.map_err(|error| format!("解析已观察动态隔离 Agent join 结果失败：{error}"))?;
if payload.get("ready").and_then(serde_json::Value::as_bool) != Some(true) {
return Err("已观察动态隔离 Agent join 结果未标记 ready=true".to_string());
}
let joins = payload
.get("joins")
.and_then(serde_json::Value::as_array)
@@ -15735,6 +15804,51 @@ fn observed_isolated_join_group_ids_from_run_status(
Ok(group_ids)
}
#[cfg(test)]
mod run_status_observation_tests {
use super::*;
#[test]
fn mixed_ready_prefixes_parse_exact_static_and_isolated_ids() {
let detail = concat!(
"readyIsolatedJoins: {\"ready\":true,\"joins\":[{\"delegationGroupId\":\"group-a\"}]}\n\n",
"readyDelegateReceipts: {\"ready\":true,\"receipts\":[{\"delegationId\":\"delivery-a\"}]}\n\n",
"agentId: project-supervisor"
);
assert_eq!(
observed_isolated_join_group_ids_from_run_status(Some(detail))
.expect("parse mixed isolated prefix"),
BTreeSet::from(["group-a".to_string()])
);
assert_eq!(
observed_delegate_receipt_ids_from_run_status(Some(detail))
.expect("parse mixed static prefix"),
BTreeSet::from(["delivery-a".to_string()])
);
}
#[test]
fn ready_prefix_parser_rejects_false_and_duplicate_blocks() {
let not_ready =
"readyDelegateReceipts: {\"ready\":false,\"receipts\":[{\"delegationId\":\"delivery-a\"}]}";
assert!(
observed_delegate_receipt_ids_from_run_status(Some(not_ready))
.expect_err("ready=false must fail closed")
.contains("ready=true")
);
let duplicate = concat!(
"readyDelegateReceipts: {\"ready\":true,\"receipts\":[{\"delegationId\":\"delivery-a\"}]}\n\n",
"readyDelegateReceipts: {\"ready\":true,\"receipts\":[{\"delegationId\":\"delivery-a\"}]}"
);
assert!(
observed_delegate_receipt_ids_from_run_status(Some(duplicate))
.expect_err("duplicate ready receipt blocks must fail closed")
.contains("重复前置区块")
);
}
}
impl AgentRuntimeToolObservation {
pub(crate) fn summary(&self) -> String {
format!("{}：{} · {}", self.tool, self.status, self.summary)
@@ -30203,8 +30317,30 @@ pub(crate) fn observe_agent_runtime_run_status(
})
}
.and_then(|mut detail| {
let ready_joins =
ready_isolated_join_status_for_parent_at(root, agent_id, run_id, action_id)?;
let claim_action_id = action_id.map(str::trim).filter(|value| !value.is_empty());
let static_delegate_output_may_be_present =
if agent_id == GAME_CREATOR_PROJECT_SUPERVISOR_AGENT_ID {
static_delegate_run_status_may_include_receipts_at(
root,
agent_id,
run_id,
claim_action_id,
)?
} else {
false
};
let isolated_join_payload_limit = if static_delegate_output_may_be_present {
AGENT_RUNTIME_READY_ISOLATED_JOIN_MIXED_PAYLOAD_MAX_CHARS
} else {
AGENT_RUNTIME_READY_ISOLATED_JOIN_PAYLOAD_MAX_CHARS
};
let ready_joins = ready_isolated_join_status_for_parent_with_budget_at(
root,
agent_id,
run_id,
action_id,
isolated_join_payload_limit,
)?;
let ready_join_count = ready_joins.len();
let ready_join_payload = if ready_join_count > 0 {
let payload = serde_json::json!({
@@ -30226,15 +30362,21 @@ pub(crate) fn observe_agent_runtime_run_status(
.map_err(|error| format!("序列化动态隔离 Agent claimed join 失败：{error}"))?;
detail = format!("claimedIsolatedJoins: {payload}\n\n{detail}");
}
let ready_delegate_receipts = if agent_id == GAME_CREATOR_PROJECT_SUPERVISOR_AGENT_ID {
let barrier = static_delegate_completion_barrier_at(root, agent_id, run_id)?;
let claim_action_id = action_id.map(str::trim).filter(|value| !value.is_empty());
if barrier.ready_unclaimed_count > 0 && claim_action_id.is_none() {
return Err("agent.run_status 认领专业 Agent 回执必须绑定 actionId".to_string());
}
claim_action_id
.map(|action_id| {
claim_ready_static_delegate_receipts_at(root, agent_id, run_id, action_id)
claim_ready_static_delegate_receipts_with_budget_at(
root,
agent_id,
run_id,
action_id,
STATIC_DELEGATE_READY_RECEIPTS_PAYLOAD_MAX_CHARS,
)
})
.transpose()?
.unwrap_or_default()
@@ -30269,13 +30411,8 @@ pub(crate) fn observe_agent_runtime_run_status(
}
Ok(())
})();
let payload = serde_json::to_string(&serde_json::json!({
"ready": true,
"receipts": ready_delegate_receipts,
}))
.map_err(|error| format!("序列化专业 Agent ready receipts 失败：{error}"))?;
detail = format!("readyDelegateReceipts: {payload}\n\n{detail}");
}
let claimed_delegate_deliveries = if agent_id == GAME_CREATOR_PROJECT_SUPERVISOR_AGENT_ID {
claimed_static_delegate_deliveries_at(root, agent_id, run_id)?
} else {
@@ -30307,9 +30444,33 @@ pub(crate) fn observe_agent_runtime_run_status(
.map_err(|error| format!("序列化专业 Agent claimed contracts 失败：{error}"))?;
detail = format!("claimedDelegateContracts: {payload}\n\n{detail}");
}
// Ready payloads are complete evidence. The ordinary status summary may be shortened,
// but evidence must fit its budget before the corresponding claim is committed.
let base_detail = truncate_agent_runtime_text(
sanitize_prompt_context(&detail).as_str(),
AGENT_RUNTIME_RUN_STATUS_BASE_DETAIL_MAX_CHARS,
);
let mut detail = base_detail;
if ready_delegate_count > 0 {
let payload = serde_json::to_string(&serde_json::json!({
"ready": true,
"receipts": ready_delegate_receipts,
}))
.map_err(|error| format!("序列化专业 Agent ready receipts 失败：{error}"))?;
detail = format!("readyDelegateReceipts: {payload}\n\n{detail}");
}
if let Some(payload) = ready_join_payload {
detail = format!("readyIsolatedJoins: {payload}\n\n{detail}");
}
let detail = sanitize_prompt_context(&detail);
let detail_chars = detail.chars().count();
if detail_chars > AGENT_RUNTIME_RUN_STATUS_OBSERVATION_MAX_CHARS {
return Err(format!(
"agent.run_status 完整 observation 超过上限，拒绝静默截断：{} > {}",
detail_chars, AGENT_RUNTIME_RUN_STATUS_OBSERVATION_MAX_CHARS
));
}
Ok((
detail,
ready_join_count,
@@ -30426,12 +30587,44 @@ fn ready_isolated_join_status_for_parent_at(
parent_run_id: &str,
action_id: Option<&str>,
) -> Result<Vec<serde_json::Value>, String> {
let joins = claim_ready_isolated_joins_at(root, parent_agent_id, parent_run_id, action_id)?;
render_isolated_join_status_batch(&joins)
ready_isolated_join_status_for_parent_with_budget_at(
root,
parent_agent_id,
parent_run_id,
action_id,
AGENT_RUNTIME_READY_ISOLATED_JOIN_PAYLOAD_MAX_CHARS,
)
}
fn ready_isolated_join_status_for_parent_with_budget_at(
root: &Path,
parent_agent_id: &str,
parent_run_id: &str,
action_id: Option<&str>,
max_payload_chars: usize,
) -> Result<Vec<serde_json::Value>, String> {
let joins = claim_ready_isolated_joins_with_budget_at(
root,
parent_agent_id,
parent_run_id,
action_id,
max_payload_chars,
)?;
render_isolated_join_status_batch_with_limit(&joins, max_payload_chars)
}
pub(crate) fn render_isolated_join_status_batch(
joins: &[JoinDispatch],
) -> Result<Vec<serde_json::Value>, String> {
render_isolated_join_status_batch_with_limit(
joins,
AGENT_RUNTIME_READY_ISOLATED_JOIN_PAYLOAD_MAX_CHARS,
)
}
fn render_isolated_join_status_batch_with_limit(
joins: &[JoinDispatch],
max_payload_chars: usize,
) -> Result<Vec<serde_json::Value>, String> {
let rendered = joins
.iter()
@@ -30442,11 +30635,11 @@ pub(crate) fn render_isolated_join_status_batch(
"joins": &rendered,
}))
.map_err(|error| format!("序列化动态隔离 Agent ready join 失败：{error}"))?;
if payload.chars().count() > AGENT_RUNTIME_READY_ISOLATED_JOIN_PAYLOAD_MAX_CHARS {
if payload.chars().count() > max_payload_chars {
return Err(format!(
"动态隔离 Agent ready join 结果超过单次完整观察上限：{} > {}",
payload.chars().count(),
AGENT_RUNTIME_READY_ISOLATED_JOIN_PAYLOAD_MAX_CHARS
max_payload_chars
));
}
Ok(rendered)
@@ -30503,6 +30696,22 @@ fn claim_ready_isolated_joins_at(
parent_agent_id: &str,
parent_run_id: &str,
action_id: Option<&str>,
) -> Result<Vec<JoinDispatch>, String> {
claim_ready_isolated_joins_with_budget_at(
root,
parent_agent_id,
parent_run_id,
action_id,
AGENT_RUNTIME_READY_ISOLATED_JOIN_PAYLOAD_MAX_CHARS,
)
}
fn claim_ready_isolated_joins_with_budget_at(
root: &Path,
parent_agent_id: &str,
parent_run_id: &str,
action_id: Option<&str>,
max_payload_chars: usize,
) -> Result<Vec<JoinDispatch>, String> {
let action_id = action_id.map(str::trim).filter(|value| !value.is_empty());
let mut unobserved_claims = list_isolated_join_claims_at(root)?
@@ -30520,7 +30729,7 @@ fn claim_ready_isolated_joins_at(
})?;
let mut recovered = std::collections::BTreeMap::<String, JoinDispatch>::new();
for claim in unobserved_claims {
render_isolated_join_status_batch(&claim.joins)?;
render_isolated_join_status_batch_with_limit(&claim.joins, max_payload_chars)?;
let claim_lock = acquire_isolated_join_claim_lock_at(
root,
&claim.parent_agent_id,
@@ -30541,13 +30750,16 @@ fn claim_ready_isolated_joins_at(
}
}
let recovered = recovered.into_values().collect::<Vec<_>>();
render_isolated_join_status_batch(&recovered)?;
render_isolated_join_status_batch_with_limit(&recovered, max_payload_chars)?;
return Ok(recovered);
}
if action_id.is_some() {
if let Some(recovered) =
synthesize_next_legacy_isolated_join_claim_at(root, parent_agent_id, parent_run_id)?
{
if let Some(recovered) = synthesize_next_legacy_isolated_join_claim_at(
root,
parent_agent_id,
parent_run_id,
max_payload_chars,
)? {
return Ok(recovered);
}
}
@@ -30578,7 +30790,7 @@ fn claim_ready_isolated_joins_at(
}
let action_id =
action_id.ok_or_else(|| "agent.run_status 认领 all-join 必须绑定 actionId".to_string())?;
let candidates = select_isolated_join_claim_batch(candidates)?;
let candidates = select_isolated_join_claim_batch_with_limit(candidates, max_payload_chars)?;
let claim_lock =
acquire_isolated_join_claim_lock_at(root, parent_agent_id, parent_run_id, action_id)?;
if let Some(claim) =
@@ -30616,6 +30828,7 @@ fn synthesize_next_legacy_isolated_join_claim_at(
root: &Path,
parent_agent_id: &str,
parent_run_id: &str,
max_payload_chars: usize,
) -> Result<Option<Vec<JoinDispatch>>, String> {
let claims = list_isolated_join_claims_at(root)?;
let parent_claims = claims
@@ -30691,7 +30904,7 @@ fn synthesize_next_legacy_isolated_join_claim_at(
"动态隔离 Agent 旧认领 action 无法完整恢复：{legacy_action_id} 的 group 超过 16 个"
));
}
render_isolated_join_status_batch(&joins).map_err(|error| {
render_isolated_join_status_batch_with_limit(&joins, max_payload_chars).map_err(|error| {
format!("动态隔离 Agent 旧认领 action 无法完整观察：{legacy_action_id}：{error}")
})?;
let claim_lock = acquire_isolated_join_claim_lock_at(
@@ -30712,7 +30925,7 @@ fn synthesize_next_legacy_isolated_join_claim_at(
return Ok(None);
}
let recovered = commit_isolated_join_claim_locked_at(root, existing, &claim_lock)?;
render_isolated_join_status_batch(&recovered)?;
render_isolated_join_status_batch_with_limit(&recovered, max_payload_chars)?;
return Ok(Some(recovered));
}
let join_locks = acquire_isolated_join_locks_at(root, &joins)?;
@@ -30739,12 +30952,22 @@ fn synthesize_next_legacy_isolated_join_claim_at(
};
write_isolated_join_claim_at(root, &claim)?;
let recovered = commit_isolated_join_claim_with_locks_at(root, claim, &claim_lock, join_locks)?;
render_isolated_join_status_batch(&recovered)?;
render_isolated_join_status_batch_with_limit(&recovered, max_payload_chars)?;
Ok(Some(recovered))
}
fn select_isolated_join_claim_batch(
candidates: Vec<JoinDispatch>,
) -> Result<Vec<JoinDispatch>, String> {
select_isolated_join_claim_batch_with_limit(
candidates,
AGENT_RUNTIME_READY_ISOLATED_JOIN_PAYLOAD_MAX_CHARS,
)
}
fn select_isolated_join_claim_batch_with_limit(
candidates: Vec<JoinDispatch>,
max_payload_chars: usize,
) -> Result<Vec<JoinDispatch>, String> {
let mut selected = Vec::new();
for candidate in candidates {
@@ -30753,7 +30976,7 @@ fn select_isolated_join_claim_batch(
}
let mut next = selected.clone();
next.push(candidate.clone());
match render_isolated_join_status_batch(&next) {
match render_isolated_join_status_batch_with_limit(&next, max_payload_chars) {
Ok(_) => selected.push(candidate),
Err(error) if selected.is_empty() => return Err(error),
Err(_) => break,
apps/ai-game-creator-shell/src-tauri/src/delegation.rs
+363 -32
View File
@@ -7,6 +7,7 @@ const STATIC_DELEGATE_DELIVERY_MAX_BYTES: usize = 128 * 1024;
const STATIC_DELEGATE_CLAIM_SCHEMA_VERSION: &str = "game-creator-static-delegate-claim.v1";
const STATIC_DELEGATE_CLAIM_DIR: &str = ".agent/runtime/delegation-claims";
const STATIC_DELEGATE_CLAIM_MAX_BYTES: usize = 128 * 1024;
pub(crate) const STATIC_DELEGATE_READY_RECEIPTS_PAYLOAD_MAX_CHARS: usize = 6_000;
const STATIC_DELEGATE_MAX_ACCEPTANCE_CRITERIA: usize = 8;
const STATIC_DELEGATE_ACCEPTANCE_CRITERION_MAX_CHARS: usize = 240;
const STATIC_DELEGATE_MAX_EXPECTED_ARTIFACTS: usize = 16;
@@ -449,11 +450,51 @@ pub(crate) fn static_delegate_completion_barrier_at(
Ok(barrier)
}
pub(crate) fn static_delegate_run_status_may_include_receipts_at(
root: &Path,
parent_agent_id: &str,
parent_run_id: &str,
action_id: Option<&str>,
) -> Result<bool, String> {
validate_static_delegate_id(parent_agent_id, "parentAgentId", 96)?;
validate_static_delegate_id(parent_run_id, "parentRunId", 160)?;
if let Some(action_id) = action_id {
validate_static_delegate_id(action_id, "actionId", 160)?;
}
Ok(list_static_delegate_deliveries_at(root)?
.into_iter()
.any(|delivery| {
delivery.parent_agent_id == parent_agent_id
&& delivery.parent_run_id == parent_run_id
&& (delivery.status == StaticDelegateDeliveryStatus::Ready
|| (delivery.status == StaticDelegateDeliveryStatus::ClaimedByParent
&& action_id.is_some_and(|action_id| {
delivery.claimed_by_action_id.as_deref() == Some(action_id)
})))
}))
}
pub(crate) fn claim_ready_static_delegate_receipts_at(
root: &Path,
parent_agent_id: &str,
parent_run_id: &str,
action_id: &str,
) -> Result<Vec<StaticDelegateReadyReceipt>, String> {
claim_ready_static_delegate_receipts_with_budget_at(
root,
parent_agent_id,
parent_run_id,
action_id,
STATIC_DELEGATE_READY_RECEIPTS_PAYLOAD_MAX_CHARS,
)
}
pub(crate) fn claim_ready_static_delegate_receipts_with_budget_at(
root: &Path,
parent_agent_id: &str,
parent_run_id: &str,
action_id: &str,
max_payload_chars: usize,
) -> Result<Vec<StaticDelegateReadyReceipt>, String> {
validate_static_delegate_id(parent_agent_id, "parentAgentId", 96)?;
validate_static_delegate_id(parent_run_id, "parentRunId", 160)?;
@@ -463,9 +504,14 @@ pub(crate) fn claim_ready_static_delegate_receipts_at(
if let Some(claim) =
read_static_delegate_claim_at(root, parent_agent_id, parent_run_id, action_id)?
{
return commit_static_delegate_claim_locked_at(root, claim, &claim_lock);
return commit_static_delegate_claim_locked_with_budget_at(
root,
claim,
&claim_lock,
max_payload_chars,
);
}
let delivery_ids = list_static_delegate_deliveries_at(root)?
let deliveries = list_static_delegate_deliveries_at(root)?
.into_iter()
.filter(|delivery| {
delivery.parent_agent_id == parent_agent_id
@@ -476,44 +522,52 @@ pub(crate) fn claim_ready_static_delegate_receipts_at(
| StaticDelegateDeliveryStatus::ClaimedByParent
)
})
.map(|delivery| delivery.delegation_id)
.collect::<Vec<_>>();
let delivery_locks = acquire_static_delegate_delivery_locks_at(root, delivery_ids.clone())?;
let mut required_delegation_ids = std::collections::BTreeSet::new();
let mut receipts = Vec::new();
for delegation_id in delivery_ids {
let Some(delivery) = read_static_delegate_delivery_at(root, &delegation_id)? else {
return Err(format!("静态委派 delivery 在认领前消失：{delegation_id}"));
};
for delivery in deliveries {
if delivery.status == StaticDelegateDeliveryStatus::ClaimedByParent {
if delivery.claimed_by_action_id.as_deref() != Some(action_id) {
continue;
}
required_delegation_ids.insert(delivery.delegation_id.clone());
}
receipts.push(static_delegate_ready_receipt_from_delivery(delivery));
}
receipts.sort_by(|left, right| left.delegation_id.cmp(&right.delegation_id));
if receipts.is_empty() {
return Ok(receipts);
}
receipts = select_static_delegate_receipt_batch(
receipts,
&required_delegation_ids,
max_payload_chars,
)?;
let delivery_locks = acquire_static_delegate_delivery_locks_at(
root,
receipts
.iter()
.map(|receipt| receipt.delegation_id.clone())
.collect(),
)?;
for expected in &receipts {
let delivery = read_static_delegate_delivery_at(root, &expected.delegation_id)?
.ok_or_else(|| format!("静态委派 delivery 在认领前消失：{}", expected.delegation_id))?;
if delivery.parent_agent_id != parent_agent_id
|| delivery.parent_run_id != parent_run_id
|| !matches!(
delivery.status,
StaticDelegateDeliveryStatus::Ready | StaticDelegateDeliveryStatus::ClaimedByParent
)
|| (delivery.status == StaticDelegateDeliveryStatus::ClaimedByParent
&& delivery.claimed_by_action_id.as_deref() != Some(action_id))
|| static_delegate_ready_receipt_from_delivery(delivery) != *expected
{
continue;
return Err(format!(
"静态委派 delivery 在预算选择后发生变化：{}",
expected.delegation_id
));
}
if delivery.status == StaticDelegateDeliveryStatus::ClaimedByParent {
if delivery.claimed_by_action_id.as_deref() != Some(action_id) {
continue;
}
}
receipts.push(StaticDelegateReadyReceipt {
delegation_id: delivery.delegation_id,
target_agent_id: delivery.target_agent_id,
status: delivery
.terminal_status
.unwrap_or_else(|| "unknown".to_string()),
summary: delivery.result_summary.unwrap_or_default(),
acceptance_criteria: delivery.acceptance_criteria,
expected_artifacts: delivery.expected_artifacts,
repair_of_delegation_id: delivery.repair_of_delegation_id,
structured_result: delivery.structured_result,
});
}
receipts.sort_by(|left, right| left.delegation_id.cmp(&right.delegation_id));
if receipts.is_empty() {
return Ok(receipts);
}
let claim = StaticDelegateClaimRecord {
schema_version: STATIC_DELEGATE_CLAIM_SCHEMA_VERSION.to_string(),
@@ -525,7 +579,85 @@ pub(crate) fn claim_ready_static_delegate_receipts_at(
updated_at: unix_timestamp(),
};
write_static_delegate_claim_at(root, &claim)?;
commit_static_delegate_claim_with_locks_at(root, claim, &claim_lock, delivery_locks)
commit_static_delegate_claim_with_locks_with_budget_at(
root,
claim,
&claim_lock,
delivery_locks,
max_payload_chars,
)
}
fn serialize_static_delegate_ready_receipts_payload(
receipts: &[StaticDelegateReadyReceipt],
) -> Result<String, String> {
serde_json::to_string(&serde_json::json!({
"ready": true,
"receipts": receipts,
}))
.map_err(|error| format!("序列化专业 Agent ready receipts 失败：{error}"))
}
fn static_delegate_ready_receipt_from_delivery(
delivery: StaticDelegateDeliveryRecord,
) -> StaticDelegateReadyReceipt {
StaticDelegateReadyReceipt {
delegation_id: delivery.delegation_id,
target_agent_id: delivery.target_agent_id,
status: delivery
.terminal_status
.unwrap_or_else(|| "unknown".to_string()),
summary: delivery.result_summary.unwrap_or_default(),
acceptance_criteria: delivery.acceptance_criteria,
expected_artifacts: delivery.expected_artifacts,
repair_of_delegation_id: delivery.repair_of_delegation_id,
structured_result: delivery.structured_result,
}
}
fn select_static_delegate_receipt_batch(
receipts: Vec<StaticDelegateReadyReceipt>,
required_delegation_ids: &std::collections::BTreeSet<String>,
max_payload_chars: usize,
) -> Result<Vec<StaticDelegateReadyReceipt>, String> {
let mut selected = receipts
.iter()
.filter(|receipt| required_delegation_ids.contains(&receipt.delegation_id))
.cloned()
.collect::<Vec<_>>();
if !selected.is_empty() {
let required_payload_chars = serialize_static_delegate_ready_receipts_payload(&selected)?
.chars()
.count();
if required_payload_chars > max_payload_chars {
return Err(format!(
"专业 Agent 已认领 ready receipts 超过单次完整观察上限：{} > {}",
required_payload_chars, max_payload_chars
));
}
}
for receipt in receipts
.iter()
.filter(|receipt| !required_delegation_ids.contains(&receipt.delegation_id))
.cloned()
{
let mut next = selected.clone();
next.push(receipt);
next.sort_by(|left, right| left.delegation_id.cmp(&right.delegation_id));
let fits = serialize_static_delegate_ready_receipts_payload(&next)?
.chars()
.count()
<= max_payload_chars;
if fits {
selected = next;
continue;
}
break;
}
if selected.is_empty() {
return Err("专业 Agent ready receipts 无法形成完整观察批次".to_string());
}
Ok(selected)
}
pub(crate) fn mark_static_delegate_claim_observed_at(
@@ -565,6 +697,56 @@ pub(crate) fn mark_static_delegate_claim_observed_at(
Ok(true)
}
pub(crate) fn mark_static_delegate_claim_observed_for_receipts_at(
root: &Path,
parent_agent_id: &str,
parent_run_id: &str,
action_id: &str,
observed_delegation_ids: &std::collections::BTreeSet<String>,
) -> Result<bool, String> {
let claim_lock =
acquire_static_delegate_claim_lock_at(root, parent_agent_id, parent_run_id, action_id)?;
let Some(mut claim) =
read_static_delegate_claim_at(root, parent_agent_id, parent_run_id, action_id)?
else {
return Ok(false);
};
let claim_delegation_ids = claim
.receipts
.iter()
.map(|receipt| receipt.delegation_id.clone())
.collect::<std::collections::BTreeSet<_>>();
if &claim_delegation_ids != observed_delegation_ids {
return Err(format!(
"静态委派 observation 未完整包含 claim receipts：expected={} observed={}",
claim_delegation_ids.len(),
observed_delegation_ids.len()
));
}
if claim.status == StaticDelegateClaimStatus::Prepared {
let delivery_locks = acquire_static_delegate_delivery_locks_at(
root,
claim
.receipts
.iter()
.map(|receipt| receipt.delegation_id.clone())
.collect(),
)?;
commit_static_delegate_claim_with_locks_at(root, claim, &claim_lock, delivery_locks)?;
claim = read_static_delegate_claim_at(root, parent_agent_id, parent_run_id, action_id)?
.ok_or_else(|| "静态委派 claim 在标记 observation 前消失".to_string())?;
}
if claim.status != StaticDelegateClaimStatus::Observed {
if claim.status != StaticDelegateClaimStatus::Committed {
return Err("静态委派 claim 尚未完成，不能标记 observation".to_string());
}
claim.status = StaticDelegateClaimStatus::Observed;
claim.updated_at = unix_timestamp();
write_static_delegate_claim_at(root, &claim)?;
}
Ok(true)
}
fn commit_static_delegate_claim_at(
root: &Path,
claim: StaticDelegateClaimRecord,
@@ -583,6 +765,20 @@ fn commit_static_delegate_claim_locked_at(
root: &Path,
claim: StaticDelegateClaimRecord,
claim_lock: &AgentRuntimeTaskLock,
) -> Result<Vec<StaticDelegateReadyReceipt>, String> {
commit_static_delegate_claim_locked_with_budget_at(
root,
claim,
claim_lock,
STATIC_DELEGATE_READY_RECEIPTS_PAYLOAD_MAX_CHARS,
)
}
fn commit_static_delegate_claim_locked_with_budget_at(
root: &Path,
claim: StaticDelegateClaimRecord,
claim_lock: &AgentRuntimeTaskLock,
max_payload_chars: usize,
) -> Result<Vec<StaticDelegateReadyReceipt>, String> {
validate_static_delegate_claim_record(&claim)?;
let latest = read_static_delegate_claim_at(
@@ -601,7 +797,13 @@ fn commit_static_delegate_claim_locked_at(
.map(|receipt| receipt.delegation_id.clone())
.collect(),
)?;
commit_static_delegate_claim_with_locks_at(root, latest, claim_lock, delivery_locks)
commit_static_delegate_claim_with_locks_with_budget_at(
root,
latest,
claim_lock,
delivery_locks,
max_payload_chars,
)
}
fn commit_static_delegate_claim_with_locks_at(
@@ -609,6 +811,22 @@ fn commit_static_delegate_claim_with_locks_at(
claim: StaticDelegateClaimRecord,
_claim_lock: &AgentRuntimeTaskLock,
_delivery_locks: Vec<AgentRuntimeTaskLock>,
) -> Result<Vec<StaticDelegateReadyReceipt>, String> {
commit_static_delegate_claim_with_locks_with_budget_at(
root,
claim,
_claim_lock,
_delivery_locks,
STATIC_DELEGATE_READY_RECEIPTS_PAYLOAD_MAX_CHARS,
)
}
fn commit_static_delegate_claim_with_locks_with_budget_at(
root: &Path,
claim: StaticDelegateClaimRecord,
_claim_lock: &AgentRuntimeTaskLock,
_delivery_locks: Vec<AgentRuntimeTaskLock>,
max_payload_chars: usize,
) -> Result<Vec<StaticDelegateReadyReceipt>, String> {
validate_static_delegate_claim_record(&claim)?;
let mut claim = read_static_delegate_claim_at(
@@ -618,6 +836,14 @@ fn commit_static_delegate_claim_with_locks_at(
&claim.action_id,
)?
.ok_or_else(|| "静态委派 claim 在提交期间消失".to_string())?;
let payload = serialize_static_delegate_ready_receipts_payload(&claim.receipts)?;
if payload.chars().count() > max_payload_chars {
return Err(format!(
"专业 Agent ready receipts 超过单次完整观察上限：{} > {}",
payload.chars().count(),
max_payload_chars
));
}
for receipt in &claim.receipts {
let mut delivery = read_static_delegate_delivery_at(root, &receipt.delegation_id)?
.ok_or_else(|| {
@@ -1690,4 +1916,109 @@ mod tests {
fs::remove_dir_all(root).ok();
}
#[test]
fn missing_claim_journal_recovers_required_receipt_before_new_ready_prefix() {
let root = std::env::temp_dir().join(format!(
"genarrative-static-required-recovery-{}-{}",
std::process::id(),
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.expect("system time after unix epoch")
.as_nanos()
));
init_local_game_project_at(&root, "project-1", "静态委派必选回执恢复测试")
.expect("project init");
let parent_agent_id = GAME_CREATOR_PROJECT_SUPERVISOR_AGENT_ID;
let parent_run_id = "required-recovery-parent-run";
let action_id = "required-recovery-claim-action";
let required = new_static_delegate_delivery(
parent_agent_id,
"required-recovery-parent-session-with-long-identity",
parent_run_id,
"required-recovery-delegate-action-with-long-identity",
"zz-required-recovery-delivery-with-long-identity",
"design-director",
"required-recovery-child-session-with-long-identity",
"required-recovery-child-run-with-long-identity",
);
create_or_read_static_delegate_delivery_at(&root, &required)
.expect("create required delivery");
mark_static_delegate_delivery_ready_at(
&root,
&required.target_agent_id,
&required.target_session_id,
&required.target_run_id,
&required.delegation_id,
"completed",
"必须优先恢复的已认领回执",
)
.expect("mark required delivery ready");
let required_receipts = claim_ready_static_delegate_receipts_at(
&root,
parent_agent_id,
parent_run_id,
action_id,
)
.expect("create original required claim");
assert_eq!(required_receipts.len(), 1);
fs::remove_file(root.join(static_delegate_claim_relative_path(
parent_agent_id,
parent_run_id,
action_id,
)))
.expect("remove claim journal to simulate torn legacy state");
let optional = new_static_delegate_delivery(
parent_agent_id,
"p",
parent_run_id,
"a",
"aa-new-ready",
"art-director",
"s",
"r",
);
create_or_read_static_delegate_delivery_at(&root, &optional)
.expect("create optional ready delivery");
mark_static_delegate_delivery_ready_at(
&root,
&optional.target_agent_id,
&optional.target_session_id,
&optional.target_run_id,
&optional.delegation_id,
"completed",
"新回执",
)
.expect("mark optional delivery ready");
let required_budget = serialize_static_delegate_ready_receipts_payload(&required_receipts)
.expect("serialize required receipt")
.chars()
.count();
let optional_lock = try_acquire_game_creator_agent_delegation_lock_with_wait(
&root,
&optional.delegation_id,
"static-delivery",
)
.expect("acquire deferred optional delivery lock")
.expect("deferred optional delivery lock available");
let recovered = claim_ready_static_delegate_receipts_with_budget_at(
&root,
parent_agent_id,
parent_run_id,
action_id,
required_budget,
)
.expect("recover required receipt without consuming optional prefix");
drop(optional_lock);
assert_eq!(recovered, required_receipts);
let deferred = read_static_delegate_delivery_at(&root, &optional.delegation_id)
.expect("read deferred optional delivery")
.expect("deferred optional delivery exists");
assert_eq!(deferred.status, StaticDelegateDeliveryStatus::Ready);
assert!(deferred.claimed_by_action_id.is_none());
fs::remove_dir_all(root).ok();
}
}
apps/ai-game-creator-shell/src-tauri/src/tests.rs
+511
View File
File diff suppressed because it is too large Load Diff