mirror of
https://github.com/tinyhumansai/openhuman.git
synced 2026-07-27 21:08:00 +00:00
feat(orchestration): auto-accept contact requests from linked agents (O2) (#4709)
This commit is contained in:
sanil-23
@@ -291,6 +291,7 @@ End-to-end coverage of the agent harness via the web-chat RPC surface against an
|
||||
| 6.3.7 | Full-stack trigger pipeline with mocked LLM | RI | `tests/subconscious_fullstack_e2e.rs` (feature `e2e-test-support`) | ✅ | Real `GatePass`+`LongLivedSession`+`Agent`+sub-agent run against a provider-layer mock (no network); promote/drop, persistence, real `spawn_subagent`. |
|
||||
| 6.3.8 | Subconscious Triggers debug/manage panel (Brain) | WD | `app/test/playwright/specs/subconscious-triggers.spec.ts` | ✅ | Brain→Subconscious panel: renders disabled baseline + hint + reserved thread ids; enable toggle → Pipeline Enabled + event_driven + orchestrator running; disable; refresh re-fetches. |
|
||||
| 6.3.9 | Vision sub-agent reads attached images | RU | `src/openhuman/agent_registry/agents/loader.rs::vision_agent_loads_on_vision_hint`, `src/openhuman/inference/provider/factory_tests.rs::vision_tier_is_vision_capable`, `src/openhuman/agent/harness/engine/core.rs::gate_tests`, `src/openhuman/agent/multimodal_tests.rs::extract_image_placeholders_pulls_att_tokens_in_order` | ✅ | Orchestrator (non-vision `chat-v1`) keeps the image as a placeholder, delegates to `vision_agent` on the `vision-v1` tier, which rehydrates the on-disk attachment and reads it. Engine gate prefers per-tier `current_model_vision`; turn placeholders forwarded into the sub-agent prompt. |
|
||||
| 6.3.10 | Auto-accept contact requests from linked agents | RU | `src/openhuman/agent_orchestration/pairing.rs::{auto_accept_gate_accepts_linked_but_leaves_others_pending,auto_accept_gate_unifies_base58_and_base64_of_same_key,auto_accept_gate_accepts_nothing_with_empty_linked_set,auto_accept_fails_closed_on_unreadable_pairing_store,incoming_pending_requesters_filters_and_resolves_requester}` | ✅ | On an inbound tiny.place `contact_request`, OpenHuman auto-accepts iff the requester is in `linked_agent_ids()` (its own paired agents) and otherwise leaves it pending for the human — the e2e gate that stops the relay dropping a linked agent's `session_info` intro. Fail-closed: a pairing-store read error yields an empty linked set → nothing auto-accepted. base58/base64 encodings of the same key unify via the shared DM-ingest matcher. |
|
||||
|
||||
### 6.4 Managed Cloud File Storage
|
||||
|
||||
|
||||
@@ -4,7 +4,7 @@
|
||||
//! local consent record for orchestration sessions that are allowed to exchange
|
||||
//! 1:1 encrypted envelopes.
|
||||
|
||||
use std::collections::HashMap;
|
||||
use std::collections::{HashMap, HashSet};
|
||||
use std::fmt::Write as _;
|
||||
use std::path::{Path, PathBuf};
|
||||
use std::sync::{Arc, LazyLock};
|
||||
@@ -15,6 +15,7 @@ use tokio::sync::Mutex;
|
||||
|
||||
use crate::core::event_bus::{publish_global, DomainEvent};
|
||||
use crate::openhuman::config::Config;
|
||||
use crate::openhuman::orchestration::ingest::resolve_linked_id;
|
||||
use crate::openhuman::tinyplace::ops::{global_state as tinyplace_state, map_err};
|
||||
|
||||
const LOG_TARGET: &str = "orchestration_pairing";
|
||||
@@ -327,11 +328,131 @@ pub(crate) async fn linked_agent_ids(workspace_dir: &Path) -> std::collections::
|
||||
.collect(),
|
||||
Err(e) => {
|
||||
log::warn!(target: LOG_TARGET, "[orchestration_pairing] linked_agent_ids read failed: {e}");
|
||||
std::collections::HashSet::new()
|
||||
HashSet::new()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Poll the tiny.place incoming contact-request queue and auto-accept every
|
||||
/// request whose requester is already a linked (paired) agent — and ONLY those.
|
||||
/// A request from an agent that is not in the local linked set is deliberately
|
||||
/// left **pending** for the human to decide: generic auto-accept stays off,
|
||||
/// because accepting a contact is a trust decision (the relay's own rule is
|
||||
/// "never auto-accept"). The one exception is the user's *own* already-paired
|
||||
/// agents.
|
||||
///
|
||||
/// Why this is the delivery gate: tiny.place's relay DROPS any DM to a peer that
|
||||
/// has not accepted a contact request, so a wrapped agent that re-establishes
|
||||
/// contact (a fresh daemon or reconnect that re-sends its one `contact_add`, or a
|
||||
/// relay-side contact reset) is otherwise blocked behind a pending request — its
|
||||
/// session intro (`session_info`) and entire session stream never arrive.
|
||||
/// Auto-accepting linked agents opens that gate for them only. (A *rotated* key is
|
||||
/// a different Ed25519 identity, so it is NOT in the linked set and its request is
|
||||
/// correctly left pending for the human.)
|
||||
///
|
||||
/// No per-cycle churn: accepting moves the relay edge from `pending` → `accepted`,
|
||||
/// and `/contacts/requests` only lists *pending* requests, so an accepted
|
||||
/// requester drops out of `incoming` and is not re-selected next pass. We also
|
||||
/// accept under the **canonical** linked id (see [`requesters_to_auto_accept`]),
|
||||
/// so a base64-form request can't slip past a base58 stored id and re-fire.
|
||||
///
|
||||
/// Fail-closed: [`linked_agent_ids`] returns an **empty** set on any pairing-store
|
||||
/// read error, so a read failure auto-accepts NOTHING (every request is left
|
||||
/// pending) rather than opening the gate. Returns the number of requests accepted
|
||||
/// this pass.
|
||||
pub async fn auto_accept_linked_contact_requests(config: &Config) -> Result<usize, String> {
|
||||
let linked = linked_agent_ids(&config.workspace_dir).await;
|
||||
// An empty linked set can match nothing — this is also the fail-closed
|
||||
// read-error case — so skip the network round-trip entirely.
|
||||
if linked.is_empty() {
|
||||
return Ok(0);
|
||||
}
|
||||
let client = tinyplace_state().client().await?;
|
||||
// `limit=100` caps one scan (consistent with `list()`); a linked requester
|
||||
// beyond the 100th *pending* request waits until earlier ones clear. Fine at
|
||||
// expected volumes — a paired fleet is small — revisit with pagination if not.
|
||||
let requests: Value = client
|
||||
.http()
|
||||
.get_agent_auth::<Value>(
|
||||
"/contacts/requests",
|
||||
&[("limit".to_string(), "100".to_string())],
|
||||
)
|
||||
.await
|
||||
.map_err(map_err)?;
|
||||
let to_accept = requesters_to_auto_accept(&incoming_pending_requesters(&requests), &linked);
|
||||
log::debug!(
|
||||
target: LOG_TARGET,
|
||||
"[orchestration_pairing] auto_accept.scan linked={} accept={}",
|
||||
linked.len(),
|
||||
to_accept.len()
|
||||
);
|
||||
let mut accepted = 0usize;
|
||||
for agent_id in to_accept {
|
||||
match accept_request(config, &agent_id).await {
|
||||
Ok(_) => {
|
||||
accepted += 1;
|
||||
log::info!(
|
||||
target: LOG_TARGET,
|
||||
"[orchestration_pairing] auto_accept.linked agent_id={agent_id}"
|
||||
);
|
||||
}
|
||||
Err(e) => log::warn!(
|
||||
target: LOG_TARGET,
|
||||
"[orchestration_pairing] auto_accept.failed agent_id={agent_id}: {e}"
|
||||
),
|
||||
}
|
||||
}
|
||||
Ok(accepted)
|
||||
}
|
||||
|
||||
/// Requester agent id of every *pending incoming* contact request in a raw
|
||||
/// `/contacts/requests` response. For an incoming request the counterpart is the
|
||||
/// requester: prefer the top-level `agentId`, falling back to `contact.requester`
|
||||
/// (the relay does not always populate `agentId`). Mirrors the frontend
|
||||
/// `contactAddress` resolution in `orchestrationTabHelpers.ts`. Pure — no IO — so
|
||||
/// the parse is unit-testable without a live client.
|
||||
fn incoming_pending_requesters(requests: &Value) -> Vec<String> {
|
||||
let Some(incoming) = requests.get("incoming").and_then(Value::as_array) else {
|
||||
return Vec::new();
|
||||
};
|
||||
incoming
|
||||
.iter()
|
||||
.filter(|view| view.get("status").and_then(Value::as_str) == Some("pending"))
|
||||
.filter_map(request_view_requester)
|
||||
.collect()
|
||||
}
|
||||
|
||||
/// The counterpart (requester) address of a single incoming contact-request view.
|
||||
fn request_view_requester(view: &Value) -> Option<String> {
|
||||
if let Some(id) = view.get("agentId").and_then(Value::as_str) {
|
||||
let trimmed = id.trim();
|
||||
if !trimmed.is_empty() {
|
||||
return Some(trimmed.to_string());
|
||||
}
|
||||
}
|
||||
view.get("contact")
|
||||
.and_then(|contact| contact.get("requester"))
|
||||
.and_then(Value::as_str)
|
||||
.map(str::trim)
|
||||
.filter(|requester| !requester.is_empty())
|
||||
.map(str::to_string)
|
||||
}
|
||||
|
||||
/// Decide which incoming requesters to auto-accept: exactly those already in the
|
||||
/// linked-agent set. Requesters not linked are intentionally left for the human.
|
||||
///
|
||||
/// Returns each match's **canonical linked id** (via [`resolve_linked_id`]), NOT
|
||||
/// the raw wire id — so a request that carries the base64 form of a key stored as
|
||||
/// base58 is accepted/persisted under the existing base58 record rather than
|
||||
/// spawning a duplicate `Linked` record for the same identity. Pure — the trust
|
||||
/// gate is unit-testable without any network or store IO.
|
||||
fn requesters_to_auto_accept(incoming: &[String], linked: &HashSet<String>) -> Vec<String> {
|
||||
incoming
|
||||
.iter()
|
||||
.filter_map(|id| resolve_linked_id(id, linked))
|
||||
.collect()
|
||||
}
|
||||
|
||||
async fn save_store(workspace_dir: &Path, store: &PairingStore) -> Result<(), String> {
|
||||
let path = store_path(workspace_dir);
|
||||
let parent = path
|
||||
@@ -483,6 +604,129 @@ mod tests {
|
||||
assert_eq!(store.records[0].source, PairingSource::ApprovedRequest);
|
||||
}
|
||||
|
||||
// ── Auto-accept gate (O2) ────────────────────────────────────────────────
|
||||
|
||||
// A real base58 pairing-store address and the base64 Ed25519 key of the SAME
|
||||
// 32-byte identity (as seen on the wire), reused from the ingest matcher test.
|
||||
const LINKED_BASE58: &str = "7jr5FKYETssD6T1MCzsR4aT4dnjjyJCE2SANYYX1R5vm";
|
||||
const LINKED_BASE64: &str = "ZCAAuA+2GVoRrT08Gt8JUVnxnISTelSxnDuyScze334=";
|
||||
const UNLINKED_BASE58: &str = "De6RHrMj6eDqX1WBTXk11sks4WXHMaqEX9A6oQ3ZEmsg";
|
||||
|
||||
#[test]
|
||||
fn auto_accept_gate_accepts_linked_but_leaves_others_pending() {
|
||||
let linked: HashSet<String> = [LINKED_BASE58.to_string()].into_iter().collect();
|
||||
|
||||
// (1) A linked requester is selected for auto-accept.
|
||||
assert_eq!(
|
||||
requesters_to_auto_accept(&[LINKED_BASE58.to_string()], &linked),
|
||||
vec![LINKED_BASE58.to_string()]
|
||||
);
|
||||
|
||||
// (2) A non-linked requester is left pending (never selected).
|
||||
assert!(
|
||||
requesters_to_auto_accept(&[UNLINKED_BASE58.to_string()], &linked).is_empty(),
|
||||
"an unlinked requester must be left pending for the human"
|
||||
);
|
||||
|
||||
// Mixed batch: only the linked id is accepted, order preserved.
|
||||
assert_eq!(
|
||||
requesters_to_auto_accept(
|
||||
&[
|
||||
UNLINKED_BASE58.to_string(),
|
||||
LINKED_BASE58.to_string(),
|
||||
UNLINKED_BASE58.to_string(),
|
||||
],
|
||||
&linked,
|
||||
),
|
||||
vec![LINKED_BASE58.to_string()]
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn auto_accept_gate_unifies_and_canonicalizes_base58_and_base64() {
|
||||
// The pairing store keeps the base58 address; a contact request may carry
|
||||
// the base64 Ed25519 key. Both are the same identity, so the linked
|
||||
// agent's request must still be accepted (the shared matcher unifies the
|
||||
// two encodings) — otherwise the e2e intro gate stays shut. Crucially the
|
||||
// gate returns the CANONICAL base58 stored id, not the raw base64 wire id,
|
||||
// so accepting under it reuses the existing record instead of persisting a
|
||||
// duplicate `Linked` row for the same identity.
|
||||
let linked: HashSet<String> = [LINKED_BASE58.to_string()].into_iter().collect();
|
||||
assert_eq!(
|
||||
requesters_to_auto_accept(&[LINKED_BASE64.to_string()], &linked),
|
||||
vec![LINKED_BASE58.to_string()],
|
||||
"must canonicalize the base64 wire id to the stored base58 id"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn auto_accept_gate_accepts_nothing_with_empty_linked_set() {
|
||||
// linked_agent_ids() fails closed to an empty set on any store read error;
|
||||
// with no linked agents the gate must accept nothing (every request stays
|
||||
// pending) rather than opening the contact gate wide.
|
||||
let empty = HashSet::new();
|
||||
assert!(
|
||||
requesters_to_auto_accept(&[LINKED_BASE58.to_string()], &empty).is_empty(),
|
||||
"an empty (fail-closed) linked set must auto-accept nothing"
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn auto_accept_fails_closed_on_unreadable_pairing_store() {
|
||||
// The read-error path end-to-end (minus network): a corrupt pairing store
|
||||
// makes linked_agent_ids() fail closed to an empty set, so the gate cannot
|
||||
// auto-accept even a would-be-linked requester.
|
||||
let tmp = tempfile::tempdir().unwrap();
|
||||
let path = store_path(tmp.path());
|
||||
tokio::fs::create_dir_all(path.parent().unwrap())
|
||||
.await
|
||||
.unwrap();
|
||||
tokio::fs::write(&path, b"{ this is not valid json")
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
let linked = linked_agent_ids(tmp.path()).await;
|
||||
assert!(
|
||||
linked.is_empty(),
|
||||
"a corrupt pairing store must fail closed to an empty linked set"
|
||||
);
|
||||
assert!(
|
||||
requesters_to_auto_accept(&[LINKED_BASE58.to_string()], &linked).is_empty(),
|
||||
"no auto-accept when the linked set is fail-closed empty"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn incoming_pending_requesters_filters_and_resolves_requester() {
|
||||
// Pending incoming only; `agentId` preferred, else `contact.requester`;
|
||||
// accepted requests and outgoing requests are ignored.
|
||||
let requests = serde_json::json!({
|
||||
"incoming": [
|
||||
{ "agentId": "AAA", "status": "pending", "direction": "incoming",
|
||||
"contact": { "requester": "AAA", "addressee": "me", "status": "pending" } },
|
||||
// agentId blank → fall back to contact.requester.
|
||||
{ "agentId": "", "status": "pending", "direction": "incoming",
|
||||
"contact": { "requester": "BBB", "addressee": "me", "status": "pending" } },
|
||||
// already accepted → skipped.
|
||||
{ "agentId": "CCC", "status": "accepted", "direction": "incoming",
|
||||
"contact": { "requester": "CCC", "addressee": "me", "status": "accepted" } },
|
||||
],
|
||||
"outgoing": [
|
||||
// outgoing is never a request to accept.
|
||||
{ "agentId": "DDD", "status": "pending", "direction": "outgoing",
|
||||
"contact": { "requester": "me", "addressee": "DDD", "status": "pending" } },
|
||||
],
|
||||
});
|
||||
assert_eq!(
|
||||
incoming_pending_requesters(&requests),
|
||||
vec!["AAA".to_string(), "BBB".to_string()]
|
||||
);
|
||||
|
||||
// Missing/empty `incoming` → nothing, no panic.
|
||||
assert!(incoming_pending_requesters(&serde_json::json!({})).is_empty());
|
||||
assert!(incoming_pending_requesters(&serde_json::json!({ "incoming": [] })).is_empty());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn pairing_store_serializes_concurrent_mutations() {
|
||||
let tmp = tempfile::tempdir().unwrap();
|
||||
|
||||
@@ -21,7 +21,7 @@ use super::types::{
|
||||
|
||||
const LOG: &str = "orchestration";
|
||||
|
||||
/// True when the DM sender `from` is one of the linked (paired) agents.
|
||||
/// True when a tiny.place agent id `from` is one of the linked (paired) agents.
|
||||
///
|
||||
/// tiny.place identifies the *same* Ed25519 key two ways: the orchestration
|
||||
/// pairing store keeps the **base58** Solana address (that's what the
|
||||
@@ -32,16 +32,34 @@ const LOG: &str = "orchestration";
|
||||
/// so the message never lands in the orchestration view. Compare by the decoded
|
||||
/// 32-byte key so the two encodings unify. Fall back to the exact-string check
|
||||
/// first (cheap, and covers ids that aren't 32-byte keys, e.g. a handle).
|
||||
fn agent_id_in_linked_set(from: &str, linked: &HashSet<String>) -> bool {
|
||||
if linked.contains(from) {
|
||||
return true;
|
||||
///
|
||||
/// Shared with the contact-request auto-accept gate
|
||||
/// (`agent_orchestration::pairing`): "is this id one of my linked agents?" is the
|
||||
/// same trust question for an inbound DM and an inbound contact request, so both
|
||||
/// resolve it through this single encoding-unifying matcher.
|
||||
pub(crate) fn agent_id_in_linked_set(from: &str, linked: &HashSet<String>) -> bool {
|
||||
resolve_linked_id(from, linked).is_some()
|
||||
}
|
||||
|
||||
/// Resolve `from` to the **canonical** linked-set id it matches — the exact
|
||||
/// stored string if present, otherwise the stored id whose decoded 32-byte key
|
||||
/// equals `from`'s (unifying the base58 pairing-store form and the base64 wire
|
||||
/// form of one Ed25519 key). Returns `None` when `from` is not a linked agent.
|
||||
///
|
||||
/// A caller that then accepts/persists against the match MUST use this canonical
|
||||
/// id, not the raw wire id: `PairingStore` dedupes records by exact-string
|
||||
/// `agent_id`, so accepting under a base64 `from` while the linked record is
|
||||
/// base58 would persist a *second* `Linked` record for the same identity — and
|
||||
/// unlinking one encoding would leave the other still authorizing the peer.
|
||||
pub(crate) fn resolve_linked_id(from: &str, linked: &HashSet<String>) -> Option<String> {
|
||||
if let Some(exact) = linked.get(from) {
|
||||
return Some(exact.clone());
|
||||
}
|
||||
let Some(from_key) = decode_agent_key(from) else {
|
||||
return false;
|
||||
};
|
||||
let from_key = decode_agent_key(from)?;
|
||||
linked
|
||||
.iter()
|
||||
.any(|id| decode_agent_key(id) == Some(from_key))
|
||||
.find(|id| decode_agent_key(id) == Some(from_key))
|
||||
.cloned()
|
||||
}
|
||||
|
||||
/// Decode a tiny.place agent identifier to its raw 32-byte Ed25519 public key,
|
||||
@@ -654,6 +672,17 @@ mod tests {
|
||||
// An unrelated key is still rejected.
|
||||
let other = "De6RHrMj6eDqX1WBTXk11sks4WXHMaqEX9A6oQ3ZEmsg";
|
||||
assert!(!agent_id_in_linked_set(other, &linked));
|
||||
|
||||
// `resolve_linked_id` unifies encodings AND returns the CANONICAL stored
|
||||
// id (base58), never the raw base64 wire form — so a caller accepting
|
||||
// against the match reuses the one linked record instead of duplicating it.
|
||||
assert_eq!(
|
||||
resolve_linked_id(base64, &linked).as_deref(),
|
||||
Some(base58),
|
||||
"a base64 `from` must canonicalize to the stored base58 id"
|
||||
);
|
||||
assert_eq!(resolve_linked_id(base58, &linked).as_deref(), Some(base58));
|
||||
assert_eq!(resolve_linked_id(other, &linked), None);
|
||||
}
|
||||
|
||||
#[test]
|
||||
|
||||
@@ -211,6 +211,21 @@ pub fn start_message_drain_supervisor() {
|
||||
),
|
||||
}
|
||||
}
|
||||
// Auto-accept contact requests from already-linked agents FIRST, so a
|
||||
// paired wrapped-agent that re-established contact is unblocked before
|
||||
// this same cycle drains its mailbox. Non-linked requesters are left
|
||||
// pending for the human (accepting a contact is a trust decision).
|
||||
match crate::openhuman::agent_orchestration::pairing::auto_accept_linked_contact_requests(
|
||||
&config,
|
||||
)
|
||||
.await
|
||||
{
|
||||
Ok(n) if n > 0 => {
|
||||
log::info!(target: LOG, "[orchestration] auto-accepted {n} linked contact request(s)")
|
||||
}
|
||||
Ok(_) => {}
|
||||
Err(e) => log::debug!(target: LOG, "[orchestration] auto-accept error: {e}"),
|
||||
}
|
||||
match super::ingest::drain_mailbox_once(&config).await {
|
||||
Ok(n) if n > 0 => {
|
||||
log::debug!(target: LOG, "[orchestration] drain: examined {n} envelope(s)")
|
||||
|
||||
Reference in New Issue
Block a user