Struct Handler 

pub struct Handler {

    uid_to_block: HashMap<String, String>,
    block_to_uid: HashMap<String, String>,
    name_to_blocks: HashMap<String, Vec<String>>,
    block_names: HashMap<String, NameBindings>,
    agent_info: HashMap<String, AgentRegistration>,
    input_sender: Option<Arc<dyn Fn(&str, &[u8]) -> Result<(), String> + Send + Sync>>,
    message_sender: Option<Arc<dyn Fn(&str, &str) -> Result<SenderDelivery, String> + Send + Sync>>,
    agent_identity_confirmer: Option<Arc<dyn Fn(&str) -> Option<String> + Send + Sync>>,
    stable_agent_identity_confirmer: Option<Arc<dyn Fn(&str) -> Option<String> + Send + Sync>>,
    uid_identity_confirmer: Option<Arc<dyn Fn(&str) -> Option<String> + Send + Sync>>,
    block_liveness: Option<Arc<dyn Fn(&str) -> bool + Send + Sync>>,
    audit_log: Vec<AuditLogEntry>,
    delivery_source_uid: String,
    delivery_wan: Option<WanAudit>,
    delivery_lan_claimed_uid: String,
    rate_limiter: RateLimiter,
    include_source_in_message: bool,
    nudge_counters: HashMap<String, NudgeCounterState>,
}

Core reactive messaging handler.

Manages agent registrations, rate limiting, message injection, and audit logging.

Fields

uid_to_block: HashMap<String, String>

The key, when known: the agent’s UID (db_agents.id) → its block. One-to-one — a UID registering on a new block evicts its old one (a respawn). Empty for blocks the server has no row for (quick-launch panes, PTY shells), which are then reachable by name only.

block_to_uid: HashMap<String, String>

name_to_blocks: HashMap<String, Vec<String>>

Lowercased name → every block bound to it. A name may hold several blocks for the first time: two agents whose names collide both stay registered (today one silently evicts the other), and delivery to that name is refused with the candidates instead of guessed (§5.2).

block_names: HashMap<String, NameBindings>

A block’s own names, typed — see NameBindings.

agent_info: HashMap<String, AgentRegistration>

Keyed by block id (was: by lowercased name). agent_id in the record is the block’s display binding.

input_sender: Option<Arc<dyn Fn(&str, &[u8]) -> Result<(), String> + Send + Sync>>

message_sender: Option<Arc<dyn Fn(&str, &str) -> Result<SenderDelivery, String> + Send + Sync>>

Controller-aware delivery for non-PTY agents (persistent stream-json / ACP). When set, it is tried before the PTY keystroke path so messages reach (and steer) agents that have no terminal. See set_message_sender.

agent_identity_confirmer: Option<Arc<dyn Fn(&str) -> Option<String> + Send + Sync>>

Queries a resolved delivery target’s own live identity before delivering, as a check independent of this handler’s own uid_to_block / name_to_blocks registry. See AgentIdentityConfirmer’s doc comment and set_agent_identity_confirmer.

stable_agent_identity_confirmer: Option<Arc<dyn Fn(&str) -> Option<String> + Send + Sync>>

Same shape as agent_identity_confirmer, but backed by Controller::stable_agent_id() (the frozen AGENTMUX_AGENT_ID) instead of Controller::agent_id() (the live, renameable display name). Checked as an alternative match in inject_message_inner’s #2695 identity check — a jekt addressed to the stable name would otherwise always fail once the display binding has moved on post-rename. See NameBindings::stable.

uid_identity_confirmer: Option<Arc<dyn Fn(&str) -> Option<String> + Send + Sync>>

Identity M2: backed by Controller::stable_agent_uid(). Consulted ONLY when the target resolved by UID, and None is “unverifiable” (delivery proceeds) — never combined with the two name confirmers above. A registered-but-unspawned persistent agent has a live name (set_agent_id is called without a spawn) but no UID yet; running a UID target through the name check would reject it as a mismatch before the start-on-delivery fall-through could run (spec §4.4.3, revision 4.0’s P0).

block_liveness: Option<Arc<dyn Fn(&str) -> bool + Send + Sync>>

Identity M2: is this block still alive (has a controller)? The garbage collector that replaces name eviction (spec §4.4.4 Q9): when a name resolves to several blocks, the dead ones are swept before ambiguity is decided. None (tests) treats every block as live.

audit_log: Vec<AuditLogEntry>

delivery_source_uid: String

The UID of the delivery or Supervisor decision in flight (InjectionRequest::audit_source_uid), copied onto every audit entry it writes. Set and restored around one inject_message_inner or record_supervisor_decision call, under this handler’s mutex, so no other delivery can see it.

delivery_wan: Option<WanAudit>

The WAN verdict of the delivery in flight (W3-S audit), copied onto its audit entries. Same set-and-restore discipline as delivery_source_uid.

delivery_lan_claimed_uid: String

The LAN-claimed UID of the delivery in flight (identity M4d-6), same set-and-restore discipline.

rate_limiter: RateLimiter

include_source_in_message: bool

nudge_counters: HashMap<String, NudgeCounterState>

Warden Supervisor consecutive-nudge ceiling state, keyed on the target agent’s lowercased name. In-memory only (same lifecycle as audit_log) — not persisted across a restart.

Implementations

impl Handler

pub fn new() -> Self

Create a new handler without an input sender. Call set_input_sender before injecting messages.

pub fn set_input_sender( &mut self, sender: Arc<dyn Fn(&str, &[u8]) -> Result<(), String> + Send + Sync>, )

Set the input sender function for message injection.

pub fn set_message_sender( &mut self, sender: Arc<dyn Fn(&str, &str) -> Result<SenderDelivery, String> + Send + Sync>, )

Set the controller-aware message sender. When present, inject_message tries it first: persistent stream-json and ACP agents receive a structured message on their live channel (mid-turn steering); PTY-based agents report back so injection falls through to the keystroke path.

pub fn set_agent_identity_confirmer( &mut self, confirmer: Arc<dyn Fn(&str) -> Option<String> + Send + Sync>, )

Set the agent-identity confirmer used by inject_message_inner to double-check a resolved delivery target against its own live, spawn-time-captured identity before delivering. Optional — if never set, delivery proceeds exactly as before this check existed (fail-open).

pub fn set_stable_agent_identity_confirmer( &mut self, confirmer: Arc<dyn Fn(&str) -> Option<String> + Send + Sync>, )

See stable_agent_identity_confirmer’s field doc comment.

pub fn set_uid_identity_confirmer( &mut self, confirmer: Arc<dyn Fn(&str) -> Option<String> + Send + Sync>, )

See uid_identity_confirmer’s field doc comment (identity M2).

pub fn set_block_liveness( &mut self, probe: Arc<dyn Fn(&str) -> bool + Send + Sync>, )

See block_liveness’s field doc comment (identity M2).

pub fn set_include_source(&mut self, include: bool)

Set whether to include source agent prefix in injected messages.

pub fn register_agent( &mut self, agent_id: &str, block_id: &str, tab_id: Option<&str>, ) -> Result<(), String>

Register an agent with a block. Name-only registration. Every production site now goes through [register_agent_full] with its UID and counter site; this is the test-facing form (and the shape the pre-M2 API had).

pub fn register_agent_with_nonce( &mut self, agent_id: &str, block_id: &str, tab_id: Option<&str>, registration_nonce: u64, alias: Option<&str>, ) -> Result<(), String>

[register_agent], recording the registering persistent-controller spawn’s process-wide registration nonce (AgentRegistration::registration_nonce) so its own exit-handler can later compare-and-remove ([unregister_block_if_nonce]) instead of blindly wiping a fallback respawn’s fresh registration (issue #2363).

alias, if given, becomes the block’s stable name binding (NameBindings::stable): kept when a later call replaces the display binding, so a jekt addressed to AGENTMUX_AGENT_ID still resolves after the display name has moved on post-rename.

pub fn register_agent_full( &mut self, display: &str, block_id: &str, tab_id: Option<&str>, registration_nonce: u64, stable: Option<&str>, uid: Option<&str>, no_uid_counter: &'static str, ) -> Result<(), String>

The full registration (identity M2, spec §4.4.2): display is the block’s display binding, stable its stable binding (set or kept), uid its identity when the caller knows it, and no_uid_counter the §9.2 counter to bump when it does not.

Eviction is by identity, never by name:

• uid already registered on another block → that block is evicted entirely (a respawn).
• a name already held by another block is left alone unless neither block has a UID — that case keeps today’s name eviction exactly, and is counted.
• a block that has a UID keeps it when re-registered without one (sticky — a transient row miss on the presence path must not drop a live block out of uid_to_block); a block without one is upgraded in place when a later registration supplies one.

fn bind_name(&mut self, key: &str, block_id: &str)

fn unbind_name(&mut self, key: &str, block_id: &str)

fn remove_block_bindings(&mut self, block_id: &str)

Drop every trace of block_id — names, UID, record. No audit entry; callers log with the context they have.

fn live_blocks_for_key(&self, key: &str) -> Vec<String>

Blocks bound to key (lowercased name) that are still alive, per the liveness probe. Read-only: does not sweep. None probe = all live.

pub fn has_live_name(&self, name: &str) -> bool

Does any live block hold name (case-insensitive, like the cloud subscription’s keys)? An exit’s own nonce check can’t answer this: it reports false both when a newer spawn owns the registration and when nobody holds it any more.

fn is_live(&self, block_id: &str) -> bool

Is block_id alive per the liveness probe? None probe = live.

fn sweep_dead(&mut self, block_id: &str)

Drop a registration whose block has no controller (spec §4.4.4 Q9).

fn read_resolve(&self, target: &str) -> Result<Option<String>, Vec<String>>

Read-only name resolution shared by [get_agent], [lookup_by_name] and [peek_target]: Ok(Some(block)) for a UID hit or a name held by exactly one live block, Ok(None) for nothing, Err(blocks) for a name several live blocks hold. Liveness is consulted only when the name is contested (spec §4.4.4 Q9): an uncontested registration is never second-guessed, so a block that is momentarily between controllers (a resync_controller replace, or a frontend register that beats the resync) is not answered as “not found” — which would forward a purely local miss to other instances.

fn resolve_target(&mut self, target: &str) -> TargetResolution

Resolve a delivery target (spec §4.4.3): UID first, then a name held by exactly one live block — sweeping dead candidates only when the name is contested (§4.4.4 Q9) — refusing an ambiguous name with its candidates rather than guessing. Counts what it did (§9.2).

fn peek_target(&self, target: &str) -> Option<String>

The block a target would resolve to, without counting or sweeping — for callers that resolve before handing off to inject_message_inner, which resolves (and counts) for real. None for unknown or ambiguous.

pub fn lookup_by_name(&self, name: &str) -> LookupOutcome

Name lookup that tells an ambiguous name apart from an unknown one (identity M2) — for callers that must refuse with candidates rather than fall through to cross-instance forwarding (GetAgentTranscript).

pub fn unregister_agent(&mut self, agent_id: &str) -> UnregisterOutcome

Unregister by name (identity M2, spec §4.4.4 Q7). One live block → removed. Several → nothing is removed and the candidates are returned; the caller must not report success. None → NotFound.

pub fn unregister_block(&mut self, block_id: &str) -> Vec<String>

Unregister by block ID. Returns every name the block was bound under (empty if it was not registered) so name-keyed side state can be torn down for each.

pub fn unregister_block_if_nonce( &mut self, block_id: &str, expected_nonce: u64, ) -> bool

Unregister by block ID only if the current registration was written by the spawn with expected_nonce — a compare-and-remove for persistent-controller exit-handlers (issue #2363: the handler’s is_current_generation gate is read once, while a fallback respawn re-registers on a parallel task; an unconditional [unregister_block] here could wipe the NEW spawn’s registration, leaving the live agent invisible to Tier-1 delivery with nothing left to re-register it). The nonce is process-wide unique, so the guard also holds across controller replacement (resync_controller — codex P1 on PR #2500), where a controller-local generation would restart at 1 and collide. Runs atomically under the handler’s own lock (via the outer wrapper). A registration with no recorded nonce (0 — HTTP/PTY paths) is never removed by this variant: leaving a stale entry to the TTL sweep is strictly safer than deleting a live one.

Returns true if the registration was ours and was removed.

pub fn update_last_seen(&mut self, agent_id: &str)

Update the last_seen timestamp for an agent.

pub fn get_agent(&self, agent_id: &str) -> Option<&AgentRegistration>

Get agent registration by agent ID — a UID, or a name held by exactly one live block. A name held by several is None, counted (lookup.ambiguous_name): callers that authorize on it fail closed, and callers that display on it show nothing rather than the wrong one.

pub fn has_uid_registration(&self, uid: &str) -> bool

Whether uid has a live registration — a lookup with no side effect (no counter, no rate-limit token, no audit entry), for the held-jekt replay’s presence check.

pub fn block_for_uid(&self, uid: &str) -> Option<String>

The block uid is registered on, with no side effect — for the held-jekt replay’s quiet sign-in check.

pub fn inject_held(&mut self, req: InjectionRequest) -> InjectionResponse

Deliver a held jekt (durable jekt spec §2.3): the ordinary delivery path, audited as held_delivered (or held_retry on failure).

pub fn get_agent_by_block(&self, block_id: &str) -> Option<&AgentRegistration>

Get agent registration by block ID.

pub fn names_for_block(&self, block_id: &str) -> Vec<String>

Every name block_id is bound under (display first, then stable). Production teardown gets the same list back from [unregister_block]; this read-only form is what registration subscribes for cloud delivery.

pub fn list_agents(&self) -> Vec<AgentRegistration>

List all registered agents.

pub fn inject_message(&mut self, req: InjectionRequest) -> InjectionResponse

Inject a message into an agent’s terminal.

Sends message\r as a single payload (required for text display), then spawns 3 delayed \r sends at 200ms intervals as separate PTY writes to ensure submission. See docs/specs/jekt-inject-timing.md.

fn inject_message_inner( &mut self, req: InjectionRequest, outcome_on_success: Option<&str>, outcome_on_failure: Option<&str>, reason: Option<&str>, ) -> InjectionResponse

Shared delivery path behind both inject_message (ordinary jekt, every outcome/reason param always None) and record_supervisor_decision’s Nudge arm (outcome_on_success: Some("nudge_sent"), outcome_on_failure: Some("nudge_failed"), reason the Supervisor’s stated reasoning) — every audit-log write below carries these through instead of the two call sites duplicating sanitize/deliver logic. Two separate outcome params (rather than one applied uniformly) so a failed delivery is audited distinctly from a successful one (reagentx P2 on PR #2557 — the Supervisor UI’s decision feed must not show “nudged” for a delivery that actually failed).

fn deliver_audited( &mut self, req: InjectionRequest, outcome_on_success: Option<&str>, outcome_on_failure: Option<&str>, reason: Option<&str>, ) -> InjectionResponse

The body of Self::inject_message_inner, with the sender’s UID in delivery_source_uid for every audit entry it writes.

pub fn deliver_system_note( &mut self, target_agent: &str, text: &str, ) -> Result<(), String>

Deliver one of srv’s own [AgentMux] system notes to target_agent (muxbus::delivery_status). Not a jekt: no marker block, tier or trust label, because nothing but srv wrote the text, and no jekt or forwarded text can carry the marker (sanitize::neutralize_markers). Routed like a jekt: the controller’s structured channel, else keystrokes. Audited as system_note from agentmux.

fn send_to_block(&self, block_id: &str, text: &str) -> Result<(), String>

text to block_id the way a jekt goes: structured when the controller takes it, else typed and submitted.

pub fn record_supervisor_decision( &mut self, target_agent: &str, action: SupervisorAction, reason: &str, request_id: &str, source_agent: Option<&str>, source_uid: &str, ) -> Result<InjectionResponse, String>

Record a Warden Supervisor watcher agent’s decision about target_agent. A Nudge is delivered through the same path inject_message uses (inject_message_inner) and audited with outcome: "nudge_sent"; a Decline sends nothing and is audited directly with outcome: "nudge_declined".

Enforces the consecutive-nudge ceiling (MAX_CONSECUTIVE_AUTO_CONTINUES): if this nudge would exceed it, the decision is forced to a decline (audited with outcome: "nudge_declined", reason: "consecutive-nudge ceiling reached") and Err is returned so the calling agent’s MCP tool-call result surfaces the refusal directly — a signal for it to stop nudging and escalate to a human via an ordinary jekt instead of retrying. The counter resets when the target’s registration_nonce or block_id changes (a respawn — “consecutive” only makes sense within one continuous run; see NudgeCounterState’s field docs for why both signals are needed) or after NUDGE_COOLDOWN_RESET_MS of inactivity.

Does NOT check the target’s auto_continue_enabled opt-in itself — Handler has no Store access by design (this module doesn’t depend on backend::storage). That gate lives at the HTTP boundary, in handle_reactive_supervisor_decision (server/reactive.rs), which has AppState::mstore. Any other caller of this method directly is responsible for its own entitlement check first.

fn decide_audited( &mut self, target_agent: &str, action: SupervisorAction, reason: &str, request_id: &str, source_agent: Option<&str>, source_uid: &str, ) -> Result<InjectionResponse, String>

The body of Self::record_supervisor_decision, with the Supervisor’s UID in delivery_source_uid.

pub fn get_audit_log(&self, limit: usize) -> Vec<AuditLogEntry>

Get audit log entries, most recent first.

pub(super) fn log_audit( &mut self, source_agent: Option<&str>, target_agent: &str, block_id: &str, message: &str, success: bool, error_message: Option<&str>, request_id: &str, outcome: Option<&str>, reason: Option<&str>, )

Add an entry to the audit ring buffer. outcome/reason are None for every ordinary jekt injection (all current call sites) — only Warden Supervisor decisions (see record_supervisor_decision) ever set them.

pub(super) fn log_audit_registration( &mut self, event_kind: &str, agent_id: &str, block_id: &str, evicted_block: Option<&str>, evicted_agent: Option<&str>, )

Record a registration/eviction event in the audit log — the counterpart to [log_audit] (which only ever recorded delivery attempts). evicted_block/evicted_agent capture whichever prior mapping a “register” call displaced, so a same-host identity collision (two panes racing to register the same agent_id) is reconstructable after the fact instead of leaving no trace.

Trait Implementations

impl Default for Handler

fn default() -> Self

Returns the “default value” for a type.