fix(orchestrator,plugins): contact cross-project corrélé + découvrabilité assignation plugin par agent — #284 (backend: registre des demandes actives clé projet/agent appelant + correlationId, refus corrélation vide/dupliquée, cancel routé après correspondance exacte sans redirection vers une autre cible, appelant projeté Waiting→Done dans le live-state de son propre projet avec correlationId exposé par idea_workstate_read, gardes RAII réconcilient l'abandon du futur, tools ask/cancel WriteAction deny-by-default dont le refus indique où activer la permission; frontend: action Attribuer aux agents depuis Settings projet > Plugins ouvrant le panneau Agents flottant/ancré/détaché, section Capabilities de plugins — {agent} explicitant attribution skills et tools MCP des plugins activés; QA verte: orchestrator_service 79/79, mcp_server 40/40, domain live_state 15/15, correlated_update_is_preserved_in_the_lean_projection 1/1, frontend ProjectPluginSettings 3 tests, npm run build frontend, cargo fmt --check, git diff --check; AppImage IdeA_0.3.0_amd64.AppImage rebuildée workflow build-appimage, validation live cross-project restant après remplacement/relance manuels; réserves: registre corrélation en mémoire non persisté, surface web unsupported, catalogue dynamique tools MCP = réserve #282)

This commit is contained in:
2026-09-08 17:35:04 +02:00
parent df5728a50a
commit 8bfcb3a534
13 changed files with 604 additions and 35 deletions

View File

@ -401,6 +401,68 @@ impl Drop for LiveStateTurnGuard {
}
}
/// Drop-safe caller-side live-state cleanup for a cross-project request.
struct CrossProjectLiveStateGuard {
update: Option<Arc<UpdateLiveState>>,
requester: AgentId,
correlation_id: String,
armed: bool,
}
impl CrossProjectLiveStateGuard {
fn new(
update: Option<Arc<UpdateLiveState>>,
requester: AgentId,
correlation_id: String,
) -> Self {
Self {
update,
requester,
correlation_id,
armed: true,
}
}
fn disarm(&mut self) {
self.armed = false;
}
}
impl Drop for CrossProjectLiveStateGuard {
fn drop(&mut self) {
if !self.armed {
return;
}
let Some(update) = self.update.clone() else {
return;
};
let requester = self.requester;
let correlation_id = self.correlation_id.clone();
if let Ok(handle) = tokio::runtime::Handle::try_current() {
handle.spawn(async move {
if let Err(error) = update
.execute_correlated(
UpdateLiveStateInput {
agent_id: requester,
ticket: None,
intent: String::new(),
status: domain::live_state::WorkStatus::Done,
progress: None,
last_delegation: None,
},
correlation_id.clone(),
)
.await
{
crate::diag!(
"[cross-project] caller drop cleanup failed: requester={requester} correlation={correlation_id} err={error}"
);
}
});
}
}
}
/// Fournit les faits OS/runtime (exe + endpoint projet) pour écrire la déclaration MCP
/// réelle quand l'orchestrateur (re)lance une cible sur le chemin `ask`. Implémenté dans
/// app-tauri (seul détenteur de current_exe/$APPIMAGE/mcp_endpoint).
@ -513,6 +575,9 @@ pub struct OrchestratorService {
/// attach `idea_run_in_background` tasks for observability/cancellation while the
/// synchronous answer remains the first structured `Final`.
active_rendezvous: StdMutex<HashMap<RuntimeAgentKey, ActiveRendezvous>>,
/// Active cross-project calls keyed by caller identity and caller-supplied
/// correlation id. Cancel uses this registry as its routing authority.
cross_project_requests: StdMutex<HashMap<CrossProjectRequestKey, ActiveCrossProjectRequest>>,
/// Bus d'événements pour publier [`DomainEvent::AgentReplied`] à l'issue d'un
/// `ask` réussi (§17.4). Injecté via [`Self::with_events`] ; `None` ⇒ pas de
/// publication (l'`ask` fonctionne quand même).
@ -653,6 +718,19 @@ struct ActiveRendezvous {
background_tasks: Vec<TaskId>,
}
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
struct CrossProjectRequestKey {
requester_project_id: domain::ProjectId,
requester_agent_id: AgentId,
correlation_id: String,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
struct ActiveCrossProjectRequest {
target_project_id: domain::ProjectId,
target_agent_id: AgentId,
}
impl OrchestratorService {
/// Builds the service from the use cases and ports it dispatches to.
#[must_use]
@ -683,6 +761,7 @@ impl OrchestratorService {
wait_for: StdMutex::new(WaitForGraph::new()),
active_waits: StdMutex::new(Vec::new()),
active_rendezvous: StdMutex::new(HashMap::new()),
cross_project_requests: StdMutex::new(HashMap::new()),
events: None,
ask_locks: StdMutex::new(HashMap::new()),
mcp_runtime_provider: None,
@ -1216,6 +1295,91 @@ impl OrchestratorService {
}
}
async fn mark_cross_project_requester_best_effort(
&self,
root: &ProjectPath,
requester: AgentId,
status: domain::live_state::WorkStatus,
intent: String,
correlation_id: String,
) {
let Some(provider) = &self.live_state else {
return;
};
let Some(update) = provider.live_state_for(root) else {
return;
};
if let Err(error) = update
.execute_correlated(
UpdateLiveStateInput {
agent_id: requester,
ticket: None,
intent,
status,
progress: None,
last_delegation: None,
},
correlation_id.clone(),
)
.await
{
crate::diag!(
"[cross-project] caller live-state update failed: requester={requester} correlation={correlation_id} err={error}"
);
}
}
fn register_cross_project_request(
&self,
requester_project_id: domain::ProjectId,
requester_agent_id: AgentId,
correlation_id: String,
target_project_id: domain::ProjectId,
target_agent_id: AgentId,
) -> Result<CrossProjectRequestGuard<'_>, AppError> {
if correlation_id.trim().is_empty() {
return Err(AppError::Invalid(
"cross-project correlationId must not be empty".to_owned(),
));
}
if correlation_id.len() > domain::live_state::FIELD_MAX_BYTES {
return Err(AppError::Invalid(format!(
"cross-project correlationId exceeds {} bytes",
domain::live_state::FIELD_MAX_BYTES
)));
}
let key = CrossProjectRequestKey {
requester_project_id,
requester_agent_id,
correlation_id,
};
let target = ActiveCrossProjectRequest {
target_project_id,
target_agent_id,
};
let mut active = self
.cross_project_requests
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
if let Some(existing) = active.get(&key) {
return Err(AppError::Invalid(format!(
"cross-project correlationId `{}` is already active for requester {}/{} (target {}/{})",
key.correlation_id,
key.requester_project_id,
key.requester_agent_id,
existing.target_project_id,
existing.target_agent_id
)));
}
active.insert(key.clone(), target);
drop(active);
Ok(CrossProjectRequestGuard {
active: &self.cross_project_requests,
key,
target,
})
}
/// Persiste **best-effort** un tour (`Prompt`/`Response`) dans `conversation`.
///
/// No-op silencieux quand le provider/horloge ne sont pas câblés, quand le provider
@ -1460,6 +1624,18 @@ impl OrchestratorService {
requester_agent_id: AgentId,
correlation_id: String,
) -> Result<OrchestratorOutcome, AppError> {
let requester_project = self
.load_project_for_cross_tool(requester_project_id)
.await?;
if self
.find_agent_by_id(&requester_project, requester_agent_id)
.await?
.is_none()
{
return Err(AppError::NotFound(format!(
"requester agent {requester_agent_id} in project {requester_project_id}"
)));
}
let project = self.load_project_for_cross_tool(target_project_id).await?;
let target = self
.find_agent_by_id(&project, target_agent_id)
@ -1470,11 +1646,46 @@ impl OrchestratorService {
"agent {target_agent_id} is not contactable cross-project"
)));
}
let _request_guard = self.register_cross_project_request(
requester_project_id,
requester_agent_id,
correlation_id.clone(),
target_project_id,
target_agent_id,
)?;
crate::diag!(
"[cross-project] ask correlation={correlation_id} requester={requester_project_id}/{requester_agent_id} target={target_project_id}/{target_agent_id}"
);
self.ask_resolved_agent(&project, target, task, Some(requester_agent_id))
.await
self.mark_cross_project_requester_best_effort(
&requester_project.root,
requester_agent_id,
domain::live_state::WorkStatus::Waiting,
Self::distill_intent(&format!("Waiting for {} in another project", target.name)),
correlation_id.clone(),
)
.await;
let caller_update = self
.live_state
.as_ref()
.and_then(|provider| provider.live_state_for(&requester_project.root));
let mut caller_live_guard = CrossProjectLiveStateGuard::new(
caller_update,
requester_agent_id,
correlation_id.clone(),
);
let result = self
.ask_resolved_agent(&project, target, task, Some(requester_agent_id))
.await;
self.mark_cross_project_requester_best_effort(
&requester_project.root,
requester_agent_id,
domain::live_state::WorkStatus::Done,
String::new(),
correlation_id,
)
.await;
caller_live_guard.disarm();
result
}
/// Cancels/stops one contactable agent in another project by stable ids.
@ -1486,22 +1697,47 @@ impl OrchestratorService {
requester_agent_id: AgentId,
correlation_id: String,
) -> Result<OrchestratorOutcome, AppError> {
let project = self.load_project_for_cross_tool(target_project_id).await?;
let target = self
.find_agent_by_id(&project, target_agent_id)
.await?
.ok_or_else(|| AppError::NotFound(format!("agent {target_agent_id}")))?;
if !target.contactable {
let key = CrossProjectRequestKey {
requester_project_id,
requester_agent_id,
correlation_id: correlation_id.clone(),
};
let active = self
.cross_project_requests
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.get(&key)
.copied()
.ok_or_else(|| {
AppError::NotFound(format!(
"active cross-project request with correlationId `{correlation_id}` for requester {requester_project_id}/{requester_agent_id}"
))
})?;
if active.target_project_id != target_project_id
|| active.target_agent_id != target_agent_id
{
return Err(AppError::Invalid(format!(
"agent {target_agent_id} is not contactable cross-project"
"cross-project correlationId `{correlation_id}` belongs to target {}/{}, not {target_project_id}/{target_agent_id}",
active.target_project_id, active.target_agent_id
)));
}
let project = self
.load_project_for_cross_tool(active.target_project_id)
.await?;
let target = self
.find_agent_by_id(&project, active.target_agent_id)
.await?
.ok_or_else(|| AppError::NotFound(format!("agent {}", active.target_agent_id)))?;
crate::diag!(
"[cross-project] cancel correlation={correlation_id} requester={requester_project_id}/{requester_agent_id} target={target_project_id}/{target_agent_id}"
);
let _ = requester_agent_id;
self.stop_agent_by_id(&project, target.id, target.name, None)
.await
let mut outcome = self
.stop_agent_by_id(&project, target.id, target.name, None)
.await?;
outcome.detail = format!(
"cancelled cross-project request correlationId `{correlation_id}` for target {target_project_id}/{target_agent_id}"
);
Ok(outcome)
}
async fn run_in_background(
@ -1714,7 +1950,7 @@ impl OrchestratorService {
/// name, returned inline as a JSON array. Entries whose agent is no longer in the
/// manifest are dropped; `progress` is **never** exposed (the lean DTO has none).
///
/// Shape: `[{"agent","status","intent","ticket?","lastDelegation?"}]`.
/// Shape: `[{"agent","status","intent","ticket?","lastDelegation?","correlationId?"}]`.
async fn read_workstate(
&self,
project: &Project,
@ -1756,6 +1992,9 @@ impl OrchestratorService {
if let Some(last) = entry.last_delegation {
row["lastDelegation"] = serde_json::json!(last);
}
if let Some(correlation_id) = entry.correlation_id {
row["correlationId"] = serde_json::json!(correlation_id);
}
Some(row)
})
.collect();
@ -3440,6 +3679,24 @@ impl Drop for ActiveRendezvousGuard<'_> {
}
}
struct CrossProjectRequestGuard<'a> {
active: &'a StdMutex<HashMap<CrossProjectRequestKey, ActiveCrossProjectRequest>>,
key: CrossProjectRequestKey,
target: ActiveCrossProjectRequest,
}
impl Drop for CrossProjectRequestGuard<'_> {
fn drop(&mut self) {
let mut active = self
.active
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
if active.get(&self.key) == Some(&self.target) {
active.remove(&self.key);
}
}
}
/// Normalises a profile reference for tolerant matching: lowercased, with spaces,
/// dashes and underscores stripped (`"Claude Code"`, `"claude-code"`, `"claude"`
/// → comparable forms; `claude` ⊂ ... handled by the command match above).

View File

@ -85,6 +85,24 @@ impl UpdateLiveState {
/// [`AppError::Invalid`] if a text field exceeds the domain hard threshold;
/// [`AppError::Store`] on a persistence failure.
pub async fn execute(&self, input: UpdateLiveStateInput) -> Result<(), AppError> {
self.execute_with_correlation(input, None).await
}
/// Publishes a transition carrying an external cross-project correlation id.
pub async fn execute_correlated(
&self,
input: UpdateLiveStateInput,
correlation_id: String,
) -> Result<(), AppError> {
self.execute_with_correlation(input, Some(correlation_id))
.await
}
async fn execute_with_correlation(
&self,
input: UpdateLiveStateInput,
correlation_id: Option<String>,
) -> Result<(), AppError> {
let now_ms = u64::try_from(self.clock.now_millis()).unwrap_or(0);
let entry = LiveEntry::new(
input.agent_id,
@ -95,6 +113,7 @@ impl UpdateLiveState {
input.last_delegation,
now_ms,
)
.and_then(|entry| entry.with_correlation_id(correlation_id))
.map_err(|e| AppError::Invalid(e.to_string()))?;
self.store.upsert(entry).await?;
Ok(())
@ -117,6 +136,9 @@ pub struct LeanLiveEntry {
pub ticket: Option<TicketId>,
/// The ticket of the most recent delegation it issued, if any.
pub last_delegation: Option<TicketId>,
/// External correlation for an active/completed cross-project request.
#[serde(skip_serializing_if = "Option::is_none")]
pub correlation_id: Option<String>,
}
impl From<LiveEntry> for LeanLiveEntry {
@ -127,6 +149,7 @@ impl From<LiveEntry> for LeanLiveEntry {
status: e.status,
ticket: e.ticket,
last_delegation: e.last_delegation,
correlation_id: e.correlation_id,
}
}
}
@ -246,6 +269,32 @@ mod tests {
);
}
#[tokio::test]
async fn correlated_update_is_preserved_in_the_lean_projection() {
let store = Arc::new(InMemoryLiveStateStore::default());
let uc = UpdateLiveState::new(store.clone(), Arc::new(FixedClock(4_242)));
uc.execute_correlated(
UpdateLiveStateInput {
agent_id: aid(1),
ticket: None,
intent: "waiting for another project".to_owned(),
status: WorkStatus::Waiting,
progress: None,
last_delegation: None,
},
"cross-284".to_owned(),
)
.await
.unwrap();
let lean = GetLiveStateLean::new(store, Arc::new(FixedClock(4_242)))
.execute()
.await
.unwrap();
assert_eq!(lean.entries[0].correlation_id.as_deref(), Some("cross-284"));
}
#[tokio::test]
async fn update_applies_domain_bounds_truncation() {
let store = Arc::new(InMemoryLiveStateStore::default());

View File

@ -902,7 +902,10 @@ async fn cross_project_list_agents_filters_to_contactable_agents() {
let target_project = other_project();
let svc = fx
.service
.with_project_store(Arc::new(FakeProjects::new(vec![target_project.clone()])));
.with_project_store(Arc::new(FakeProjects::new(vec![
project(),
target_project.clone(),
])));
let out = svc
.list_project_agents_for_mcp(target_project.id)
@ -919,11 +922,17 @@ async fn cross_project_list_agents_filters_to_contactable_agents() {
#[tokio::test]
async fn cross_project_ask_rejects_non_contactable_agent_before_launch() {
let hidden = scratch_agent(aid(2), "Hidden", "agents/hidden.md");
let fx = fixture(FakeContexts::with_agent(&hidden, "# hidden"));
let requester = scratch_agent(aid(99), "Requester", "agents/requester.md");
let contexts = FakeContexts::with_agent(&hidden, "# hidden");
contexts.add_agent(&requester, "# requester");
let fx = fixture(contexts);
let target_project = other_project();
let svc = fx
.service
.with_project_store(Arc::new(FakeProjects::new(vec![target_project.clone()])));
.with_project_store(Arc::new(FakeProjects::new(vec![
project(),
target_project.clone(),
])));
let err = svc
.ask_project_agent_for_mcp(
@ -950,6 +959,10 @@ async fn cross_project_ask_routes_by_agent_id_when_names_collide() {
scratch_agent(aid(2), "Architect", "agents/architect-second.md").with_contactable(true);
let contexts = FakeContexts::with_agent(&first, "# first");
contexts.add_agent(&second, "# second");
contexts.add_agent(
&scratch_agent(aid(99), "Requester", "agents/requester.md"),
"# requester",
);
let fx = ask_fixture(contexts);
let target_project = other_project();
@ -989,6 +1002,10 @@ async fn cross_project_cancel_routes_by_agent_id_when_names_collide() {
scratch_agent(aid(2), "Architect", "agents/architect-second.md").with_contactable(true);
let contexts = FakeContexts::with_agent(&first, "# first");
contexts.add_agent(&second, "# second");
contexts.add_agent(
&scratch_agent(aid(99), "Requester", "agents/requester.md"),
"# requester",
);
let fx = ask_fixture(contexts);
let target_project = other_project();
let first_cancelled = Arc::new(Mutex::new(0_u32));
@ -1012,7 +1029,27 @@ async fn cross_project_cancel_routes_by_agent_id_when_names_collide() {
fx.structured
.insert_in_project(target_project.id, second_session, second.id, nid(9102));
timeout(
let ask_service = Arc::clone(&fx.service);
let ask = tokio::spawn(async move {
ask_service
.ask_project_agent_for_mcp(
target_project.id,
second.id,
"work".to_owned(),
project().id,
aid(99),
"corr-homonym-cancel".to_owned(),
)
.await
});
await_until(|| {
fx.live
.entry_for(aid(99))
.is_some_and(|entry| entry.status == WorkStatus::Waiting)
})
.await;
let cancelled = timeout(
TEST_GUARD,
fx.service.cancel_project_agent_for_mcp(
target_project.id,
@ -1026,8 +1063,95 @@ async fn cross_project_cancel_routes_by_agent_id_when_names_collide() {
.expect("cross-project cancel must not hang")
.expect("cross-project cancel ok");
assert!(cancelled.detail.contains("corr-homonym-cancel"));
assert_eq!(*first_cancelled.lock().unwrap(), 0);
assert_eq!(*second_cancelled.lock().unwrap(), 1);
fx.mailbox
.completions()
.push(second.id, TestCompletion::Cancelled);
let _ = timeout(TEST_GUARD, ask)
.await
.expect("cancelled ask must settle")
.expect("ask join");
}
#[tokio::test]
async fn cross_project_ask_projects_correlation_on_requester_and_rejects_wrong_cancel() {
let target = scratch_agent(aid(2), "Target", "agents/target.md").with_contactable(true);
let requester = scratch_agent(aid(99), "Requester", "agents/requester.md");
let contexts = FakeContexts::with_agent(&target, "# target");
contexts.add_agent(&requester, "# requester");
let fx = ask_fixture(contexts);
let target_project = other_project();
let ask_service = Arc::clone(&fx.service);
let ask = tokio::spawn(async move {
ask_service
.ask_project_agent_for_mcp(
target_project.id,
target.id,
"work".to_owned(),
project().id,
requester.id,
"corr-visible".to_owned(),
)
.await
});
await_until(|| {
fx.live.entry_for(requester.id).is_some_and(|entry| {
entry.status == WorkStatus::Waiting
&& entry.correlation_id.as_deref() == Some("corr-visible")
})
})
.await;
let workstate = fx
.service
.dispatch(
&project(),
OrchestratorCommand::ReadWorkState {
requester: ConversationParty::agent(requester.id),
},
)
.await
.expect("caller workstate readable");
let rows: serde_json::Value =
serde_json::from_str(workstate.reply.as_deref().unwrap()).unwrap();
let requester_row = rows
.as_array()
.unwrap()
.iter()
.find(|row| row["agent"] == "Requester")
.expect("requester projection");
assert_eq!(requester_row["status"], "waiting");
assert_eq!(requester_row["correlationId"], "corr-visible");
let mismatch = fx
.service
.cancel_project_agent_for_mcp(
target_project.id,
aid(1),
project().id,
requester.id,
"corr-visible".to_owned(),
)
.await
.expect_err("correlation must not be redirectable to another target");
assert!(mismatch.to_string().contains("belongs to target"));
fx.mailbox
.completions()
.push(target.id, TestCompletion::Replied("done".to_owned()));
let out = timeout(TEST_GUARD, ask)
.await
.expect("ask settles")
.expect("ask join")
.expect("ask succeeds");
assert_eq!(out.reply.as_deref(), Some("done"));
let caller = fx.live.entry_for(requester.id).expect("caller row remains");
assert_eq!(caller.status, WorkStatus::Done);
assert_eq!(caller.correlation_id.as_deref(), Some("corr-visible"));
}
#[tokio::test]