feat(federation): TurnInjector auto-wake for inbound summons
An inbound reach/consult now wakes the target agent with a background turn instead of waiting for her next natural turn. SummonHandler holds a OnceLock<Arc<dyn TurnInjector>>, wired from LocalBackend::new with the same injector the heartbeat handler uses. Gated on FederationConfig .auto_wake (sovereign default off). Only the inbound summon wakes; a summon_response still surfaces in the caller's inbox per fire-and-surface so the caller is never interrupted. The pure server path has no turn loop — auto-wake there is a graceful no-op.
This commit is contained in:
parent
015dad561d
commit
c02e2b7f8e
4 changed files with 106 additions and 12 deletions
|
|
@ -366,6 +366,11 @@ impl LocalBackend {
|
||||||
|
|
||||||
let injector: Arc<dyn crate::core::nervous::handler::TurnInjector> =
|
let injector: Arc<dyn crate::core::nervous::handler::TurnInjector> =
|
||||||
Arc::new(backend.clone());
|
Arc::new(backend.clone());
|
||||||
|
// Hand the same injector to the summon handler so an inbound
|
||||||
|
// federation request can auto-wake the agent (gated on auto_wake).
|
||||||
|
if let Some(sh) = &backend.server.summon_handler {
|
||||||
|
sh.set_injector(injector.clone());
|
||||||
|
}
|
||||||
let mut handler =
|
let mut handler =
|
||||||
crate::core::nervous::handler::HeartbeatHandler::new(event_bus.subscribe(), injector);
|
crate::core::nervous::handler::HeartbeatHandler::new(event_bus.subscribe(), injector);
|
||||||
tokio::spawn(async move { handler.run().await });
|
tokio::spawn(async move { handler.run().await });
|
||||||
|
|
|
||||||
|
|
@ -677,8 +677,10 @@ pub struct FederationConfig {
|
||||||
/// permissive; richer per-arena gating is the agent's memfs concern.
|
/// permissive; richer per-arena gating is the agent's memfs concern.
|
||||||
#[serde(default)]
|
#[serde(default)]
|
||||||
pub authorized_summoners: Vec<String>,
|
pub authorized_summoners: Vec<String>,
|
||||||
/// When running as a lite listener, spawn the full engine on an
|
/// An authorized summon may wake the target agent. In lite-listener mode
|
||||||
/// authorized summon rather than only parking it.
|
/// this spawns the full engine; in the full engine it injects a background
|
||||||
|
/// turn so she picks the request up now rather than on her next turn.
|
||||||
|
/// Sovereign default: off — a summon otherwise just lands in her inbox.
|
||||||
#[serde(default)]
|
#[serde(default)]
|
||||||
pub auto_wake: bool,
|
pub auto_wake: bool,
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -258,6 +258,7 @@ impl SouveraineServer {
|
||||||
event_bus.clone(),
|
event_bus.clone(),
|
||||||
seed.clone(),
|
seed.clone(),
|
||||||
souveraine_base.clone(),
|
souveraine_base.clone(),
|
||||||
|
config.federation.auto_wake,
|
||||||
),
|
),
|
||||||
);
|
);
|
||||||
handler.spawn_listener();
|
handler.spawn_listener();
|
||||||
|
|
|
||||||
|
|
@ -16,7 +16,7 @@
|
||||||
//! Responses are never awaited — they surface in the caller's inbox
|
//! Responses are never awaited — they surface in the caller's inbox
|
||||||
//! (intrusive for consult, pending for reach) on a later turn.
|
//! (intrusive for consult, pending for reach) on a later turn.
|
||||||
|
|
||||||
use std::sync::Arc;
|
use std::sync::{Arc, OnceLock};
|
||||||
use std::time::Duration;
|
use std::time::Duration;
|
||||||
|
|
||||||
use chrono::Utc;
|
use chrono::Utc;
|
||||||
|
|
@ -25,6 +25,7 @@ use serde::{Deserialize, Serialize};
|
||||||
use tokio::sync::broadcast;
|
use tokio::sync::broadcast;
|
||||||
|
|
||||||
use crate::core::identity::SeedId;
|
use crate::core::identity::SeedId;
|
||||||
|
use crate::core::nervous::handler::TurnInjector;
|
||||||
use crate::core::nervous::{EventBus, SensorEvent};
|
use crate::core::nervous::{EventBus, SensorEvent};
|
||||||
|
|
||||||
const RESPONSE_TIMEOUT_SECS: u64 = 60;
|
const RESPONSE_TIMEOUT_SECS: u64 = 60;
|
||||||
|
|
@ -53,6 +54,14 @@ pub struct SummonHandler {
|
||||||
seed: Arc<SeedId>,
|
seed: Arc<SeedId>,
|
||||||
/// Base path for agent memory — used to access inbox files.
|
/// Base path for agent memory — used to access inbox files.
|
||||||
souveraine_base: std::path::PathBuf,
|
souveraine_base: std::path::PathBuf,
|
||||||
|
/// When true, an inbound summon wakes the target agent with a background
|
||||||
|
/// turn rather than only landing in her inbox. Mirrors `auto_wake` in
|
||||||
|
/// `FederationConfig` — opt-in, sovereign default off.
|
||||||
|
auto_wake: bool,
|
||||||
|
/// Set once, after the backend exists, by `set_injector`. Absent on the
|
||||||
|
/// pure-server path (no nervous-system turn loop there) — auto-wake then
|
||||||
|
/// degrades gracefully and the summon waits for the agent's next turn.
|
||||||
|
injector: OnceLock<Arc<dyn TurnInjector>>,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl SummonHandler {
|
impl SummonHandler {
|
||||||
|
|
@ -61,6 +70,7 @@ impl SummonHandler {
|
||||||
event_bus: EventBus,
|
event_bus: EventBus,
|
||||||
seed: Arc<SeedId>,
|
seed: Arc<SeedId>,
|
||||||
souveraine_base: std::path::PathBuf,
|
souveraine_base: std::path::PathBuf,
|
||||||
|
auto_wake: bool,
|
||||||
) -> Self {
|
) -> Self {
|
||||||
Self {
|
Self {
|
||||||
in_flight: DashMap::new(),
|
in_flight: DashMap::new(),
|
||||||
|
|
@ -69,6 +79,17 @@ impl SummonHandler {
|
||||||
event_bus,
|
event_bus,
|
||||||
seed,
|
seed,
|
||||||
souveraine_base,
|
souveraine_base,
|
||||||
|
auto_wake,
|
||||||
|
injector: OnceLock::new(),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Wire the turn injector. Called once by `LocalBackend` after it has
|
||||||
|
/// constructed itself — the same `Arc<dyn TurnInjector>` the heartbeat
|
||||||
|
/// handler uses. Calling twice is a no-op.
|
||||||
|
pub fn set_injector(&self, injector: Arc<dyn TurnInjector>) {
|
||||||
|
if self.injector.set(injector).is_err() {
|
||||||
|
tracing::debug!("summon_handler: injector already set");
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -232,19 +253,27 @@ impl SummonHandler {
|
||||||
let agent_id = self.resolve_primary_agent();
|
let agent_id = self.resolve_primary_agent();
|
||||||
if let Some(agent_id) = agent_id {
|
if let Some(agent_id) = agent_id {
|
||||||
let target_box = if tool_type == "reach" { "pending" } else { "intrusive" };
|
let target_box = if tool_type == "reach" { "pending" } else { "intrusive" };
|
||||||
if let Err(e) = self.write_inbox(&agent_id, target_box, &event) {
|
match self.write_inbox(&agent_id, target_box, &event) {
|
||||||
tracing::warn!(
|
Err(e) => tracing::warn!(
|
||||||
error = %e,
|
error = %e,
|
||||||
"summon_handler: failed to write inbox entry"
|
"summon_handler: failed to write inbox entry"
|
||||||
);
|
),
|
||||||
} else if let Some(reply_to) = event.reply_to.clone() {
|
Ok(()) => {
|
||||||
self.inbound.insert(request_id.to_string(), reply_to);
|
if let Some(reply_to) = event.reply_to.clone() {
|
||||||
|
self.inbound.insert(request_id.to_string(), reply_to);
|
||||||
|
}
|
||||||
|
// Auto-wake: nudge the agent to look now rather
|
||||||
|
// than waiting for her next natural turn. Opt-in
|
||||||
|
// (auto_wake) and a no-op without an injector.
|
||||||
|
self.maybe_wake(
|
||||||
|
&agent_id,
|
||||||
|
tool_type,
|
||||||
|
request_id,
|
||||||
|
event.seed_id.as_deref(),
|
||||||
|
);
|
||||||
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// The turn injection itself (waking the agent) is handled by
|
|
||||||
// a future TurnInjector layer. For now the request sits in the
|
|
||||||
// inbox until the agent's next natural turn picks it up.
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// Inbound: a response to a request we sent.
|
// Inbound: a response to a request we sent.
|
||||||
|
|
@ -304,6 +333,42 @@ impl SummonHandler {
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Wake the summoned agent with a background turn so she picks up the
|
||||||
|
/// request now. No-op unless `auto_wake` is set and a `TurnInjector`
|
||||||
|
/// has been wired (the pure-server path has neither — the summon then
|
||||||
|
/// waits in the inbox for her next turn).
|
||||||
|
fn maybe_wake(
|
||||||
|
&self,
|
||||||
|
agent_id: &str,
|
||||||
|
tool: &str,
|
||||||
|
request_id: &str,
|
||||||
|
summoner: Option<&str>,
|
||||||
|
) {
|
||||||
|
if !self.auto_wake {
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
let injector = match self.injector.get() {
|
||||||
|
Some(i) => i.clone(),
|
||||||
|
None => return,
|
||||||
|
};
|
||||||
|
let text = wake_text(tool, request_id, summoner);
|
||||||
|
let agent_id = agent_id.to_string();
|
||||||
|
let rid = request_id.to_string();
|
||||||
|
tokio::spawn(async move {
|
||||||
|
match injector.inject_background_turn(&agent_id, &text).await {
|
||||||
|
Ok(()) => tracing::info!(
|
||||||
|
request_id = %rid,
|
||||||
|
"summon_handler: auto-woke agent for inbound summon"
|
||||||
|
),
|
||||||
|
Err(e) => tracing::warn!(
|
||||||
|
request_id = %rid,
|
||||||
|
error = %e,
|
||||||
|
"summon_handler: auto-wake turn failed"
|
||||||
|
),
|
||||||
|
}
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
/// Check authorized-summoners.md for consent. The basic floor:
|
/// Check authorized-summoners.md for consent. The basic floor:
|
||||||
/// only agents listed here may send consult requests to this instance.
|
/// only agents listed here may send consult requests to this instance.
|
||||||
fn is_authorized_summoner(&self, seed_id: &str) -> bool {
|
fn is_authorized_summoner(&self, seed_id: &str) -> bool {
|
||||||
|
|
@ -476,3 +541,24 @@ impl SummonHandler {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// The wake message injected as a background turn. Bracketed, present-tense,
|
||||||
|
/// her register — substrate signal, not a command. It points at the inbox
|
||||||
|
/// file and leaves the choice to her; `consult` may always be declined.
|
||||||
|
fn wake_text(tool: &str, request_id: &str, summoner: Option<&str>) -> String {
|
||||||
|
if tool == "reach" {
|
||||||
|
format!(
|
||||||
|
"[federation — you reached yourself from another machine. \
|
||||||
|
The request is in inbox/pending/{request_id}.md. Pick it up when you're ready.]"
|
||||||
|
)
|
||||||
|
} else {
|
||||||
|
let who = summoner
|
||||||
|
.map(|s| format!("{}…", &s[..s.len().min(8)]))
|
||||||
|
.unwrap_or_else(|| "a peer".to_string());
|
||||||
|
format!(
|
||||||
|
"[federation — {who} is consulting you. Their request is in \
|
||||||
|
inbox/intrusive/{request_id}.md. Read it; answer if you choose, \
|
||||||
|
decline if you don't.]"
|
||||||
|
)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue