agentmux_srv/server/agent_handlers/
instance.rs

1// Copyright 2025-2026, AgentMux Corp.
2// SPDX-License-Identifier: Apache-2.0
3
4
5use std::sync::Arc;
6
7use crate::backend::rpc::engine::WshRpcEngine;
8use crate::backend::rpc_types::{
9    COMMAND_LIST_AGENT_INSTANCES, COMMAND_GET_AGENT_INSTANCE,
10    COMMAND_CREATE_AGENT_INSTANCE, COMMAND_UPDATE_AGENT_INSTANCE,
11    COMMAND_DELETE_AGENT_INSTANCE, COMMAND_AGENT_LAST_RUNTIME,
12    AgentLastRuntime, CommandAgentLastRuntimeData,
13    CommandListAgentInstancesData, CommandGetAgentInstanceData,
14    CommandCreateAgentInstanceData, CommandUpdateAgentInstanceData,
15    CommandDeleteAgentInstanceData, DeleteAgentInstanceResult,
16};
17use crate::backend::storage::store::{
18    AgentInstance, InstanceStatus,
19};
20
21use super::super::AppState;
22
23pub fn register(engine: &Arc<WshRpcEngine>, state: &AppState) {
24    // ---- Agent instance CRUD ----
25
26    let mstore = state.mstore.clone();
27    engine.register_typed(
28        COMMAND_LIST_AGENT_INSTANCES,
29        // `Option<_>` rather than the bare struct, which is what the previous
30        // `unwrap_or_default()` was really buying: both encodings of "no
31        // filter" -- the `{}` the stub sends and the `null` a client that
32        // omits `data` sends -- still mean "no filter". What it no longer
33        // buys is swallowing a MALFORMED filter: `{"status": 5}` used to
34        // return the unfiltered list, which is the wrong answer to give
35        // silently, so it is an error now.
36        move |cmd: Option<CommandListAgentInstancesData>, _ctx| {
37            let mstore = mstore.clone();
38            async move {
39                let cmd = cmd.unwrap_or_default();
40                let rows = mstore
41                    .instance_list(cmd.definition_id.as_deref(), cmd.status.as_deref())
42                    .map_err(|e| format!("listagentinstances: {e}"))?;
43                Ok(rows)
44            }
45        },
46    );
47
48    let mstore = state.mstore.clone();
49    engine.register_typed(
50        COMMAND_GET_AGENT_INSTANCE,
51        move |cmd: CommandGetAgentInstanceData, _ctx| {
52            let mstore = mstore.clone();
53            async move {
54                // Not-found stays an Err rather than becoming `Option<..>`:
55                // the stub types this `Promise<AgentInstance>`, so answering
56                // null would be a shape change, not just a typing change.
57                mstore
58                    .instance_get(&cmd.id)
59                    .map_err(|e| format!("getagentinstance: {e}"))?
60                    .ok_or_else(|| format!("getagentinstance: not found id={}", cmd.id))
61            }
62        },
63    );
64
65    let mstore = state.mstore.clone();
66    let broker = state.broker.clone();
67    engine.register_typed(
68        COMMAND_CREATE_AGENT_INSTANCE,
69        move |cmd: CommandCreateAgentInstanceData, _ctx| {
70            let mstore = mstore.clone();
71            let broker = broker.clone();
72            async move {
73                let now = agentmux_common::time::now_ms();
74                let inst = AgentInstance {
75                    id: uuid::Uuid::new_v4().to_string(),
76                    definition_id: cmd.definition_id,
77                    parent_instance_id: cmd.parent_instance_id,
78                    block_id: cmd.block_id,
79                    session_id: String::new(),
80                    status: InstanceStatus::Running.as_str().to_string(),
81                    github_context: String::new(),
82                    started_at: now,
83                    ended_at: 0,
84                    created_at: now,
85                    // PR-F.3: launch modal passes through Identity +
86                    // Bundle picks. Empty string = blank
87                    // singleton (no override; the resolver returns
88                    // immediately on either "" or "blank").
89                    identity_id: cmd.identity_id,
90                    memory_id: cmd.memory_id,
91                    // v8: named-agent continuation. instance_name +
92                    // working_directory come from the launch-modal
93                    // overrides via CommandCreateAgentInstanceData
94                    // (added in the same spec). Empty string for
95                    // legacy/ambient launches.
96                    instance_name: cmd.instance_name.clone(),
97                    working_directory: cmd.working_directory.clone(),
98                    display_hidden: false,
99                };
100                // The row the launch landed on — its id can differ from
101                // `inst.id` (a launch of a user agent folds into that agent's
102                // row; a continuation folds into the row it continues). The
103                // frontend stores THIS id as the block's `agentInstanceId`,
104                // so the response must carry the canonical row. The session
105                // zone stamp and the change event below deliberately keep
106                // using the REQUESTED definition id: Option E anchors the
107                // zone on the definition the launch modal opened
108                // (`agent:<defId>:current`), and the frontend subscribes to
109                // `agentinstances:changed:<defId>` by that same id.
110                let created = mstore
111                    .instance_create(&inst)
112                    .map_err(|e| format!("createagentinstance: {e}"))?;
113
114                // Option E (PR 1 of 2) — stamp the agent-anchored
115                // session zone reference onto the block meta. Every
116                // block of this agent definition reads/writes through
117                // `agent:<defId>:current`. Continuation is now
118                // structural (same zone, different block) rather than
119                // parametric (per-block snapshot copy + --continue).
120                if !inst.block_id.is_empty()
121                    && crate::backend::agent_session::is_valid_definition_id(&inst.definition_id)
122                {
123                    let zone = crate::backend::agent_session::agent_current_zone(
124                        &inst.definition_id,
125                    );
126                    let mut meta_update = crate::backend::obj::MetaMapType::new();
127                    meta_update.insert(
128                        "agent:sessionZone".to_string(),
129                        serde_json::json!(zone),
130                    );
131                    let oref_str = format!("block:{}", inst.block_id);
132                    if let Err(e) = crate::server::service::update_object_meta(
133                        &mstore, &oref_str, &meta_update,
134                    ) {
135                        // Non-fatal — the instance row is the source
136                        // of truth, the meta stamp is a frontend
137                        // convenience. Log + continue so the launch
138                        // doesn't abort mid-flow.
139                        tracing::warn!(
140                            block_id = %inst.block_id,
141                            definition_id = %inst.definition_id,
142                            error = %e,
143                            "createagentinstance: failed to stamp agent:sessionZone"
144                        );
145                    }
146                }
147
148                broker.publish_event_for::<crate::backend::mps_events::AgentInstancesChanged>(&inst.definition_id, &(), vec![], 0);
149                Ok(created)
150            }
151        },
152    );
153
154    let mstore = state.mstore.clone();
155    let broker = state.broker.clone();
156    engine.register_typed(
157        COMMAND_UPDATE_AGENT_INSTANCE,
158        move |cmd: CommandUpdateAgentInstanceData, _ctx| {
159            let mstore = mstore.clone();
160            let broker = broker.clone();
161            async move {
162                // Partial write — only the fields the command provided.
163                // No fetch-and-merge: this used to `instance_get` the full
164                // row to fill the unspecified fields, which was the sole
165                // production caller needing `instance_get`'s transient
166                // per-launch columns. The store builds a dynamic UPDATE
167                // and returns the post-write row (for the event scope +
168                // response) from the reload it already runs.
169                // SPEC_UPDATEAGENTINSTANCE_PARTIAL_UPDATE_2026_05_29.md.
170                let upd = crate::backend::storage::InstanceUpdate {
171                    block_id: cmd.block_id,
172                    session_id: cmd.session_id,
173                    status: cmd.status,
174                    github_context: cmd.github_context,
175                    ended_at: cmd.ended_at,
176                };
177                let fresh = mstore
178                    .instance_update_partial(&cmd.id, &upd)
179                    .map_err(|e| format!("updateagentinstance: {e}"))?
180                    .ok_or_else(|| format!("updateagentinstance: not found id={}", cmd.id))?;
181                broker.publish_event_for::<crate::backend::mps_events::AgentInstancesChanged>(&fresh.definition_id, &(), vec![], 0);
182                Ok(fresh)
183            }
184        },
185    );
186
187    let mstore = state.mstore.clone();
188    engine.register_typed(
189        COMMAND_AGENT_LAST_RUNTIME,
190        move |cmd: CommandAgentLastRuntimeData, _ctx| {
191            let mstore = mstore.clone();
192            async move {
193                if let Some(raw) = &cmd.runtime {
194                    let normalized = crate::backend::storage::agents::normalize_last_runtime(raw)
195                        .map_err(|e| format!("agentlastruntime: {e}"))?;
196                    mstore
197                        .agent_last_runtime_set(&cmd.id, &normalized)
198                        .map_err(|e| format!("agentlastruntime: {e}"))?;
199                }
200                let runtime = mstore
201                    .agent_last_runtime_get(&cmd.id)
202                    .map_err(|e| format!("agentlastruntime: {e}"))?;
203                Ok(AgentLastRuntime { runtime })
204            }
205        },
206    );
207
208    let mstore = state.mstore.clone();
209    let broker = state.broker.clone();
210    let identity_store_del = state.identity_store.clone();
211    let id_store_del = state.id_store.clone();
212    engine.register_typed(
213        COMMAND_DELETE_AGENT_INSTANCE,
214        move |cmd: CommandDeleteAgentInstanceData, _ctx| {
215            let mstore = mstore.clone();
216            let broker = broker.clone();
217            let identity_store = identity_store_del.clone();
218            let id_store = id_store_del.clone();
219            async move {
220                // Read the row first so we can emit a scoped event after.
221                let definition_id = mstore
222                    .instance_get(&cmd.id)
223                    .map_err(|e| format!("deleteagentinstance: {e}"))?
224                    .map(|i| i.definition_id);
225                let deleted = mstore
226                    .instance_delete(&cmd.id)
227                    .map_err(|e| format!("deleteagentinstance: {e}"))?;
228                // Unconditional, matching the sibling `deleteagent` handler
229                // (ReAgent P1 round 2 on PR #3262). Gating on `deleted`
230                // skipped the identity-store purge on a RETRY — where the
231                // local `db_agents` row is already gone but an earlier
232                // attempt's credential/identity-link cleanup failed — which
233                // is the "the confirm dialog promises a purge it doesn't
234                // deliver" bug, reproduced on this entry point. The purge
235                // is a no-op for an id with no rows, so there is nothing to
236                // gate on.
237                super::purge_identity_store_rows(&identity_store, &id_store, &cmd.id, "deleteagentinstance");
238                if let Some(def_id) = definition_id.filter(|_| deleted) {
239                    broker.publish_event_for::<crate::backend::mps_events::AgentInstancesChanged>(&def_id, &(), vec![], 0);
240                }
241                Ok(DeleteAgentInstanceResult { deleted })
242            }
243        },
244    );
245
246}
247
248#[cfg(test)]
249mod tests {
250    use super::*;
251    use crate::backend::rpc::engine::WshRpcEngine;
252    use crate::server::tests::test_state;
253
254    /// `deleteagentinstance` is recorded in the engine's schema with its
255    /// exact type names — the property the RPC codegen plan depends on.
256    /// `DeleteAgentInstanceResult` didn't exist before this migration; the
257    /// The four CRUD commands record their request and response types, so the
258    /// whole of instance.rs is now in the registry rather than one command of
259    /// five. A command that falls back to `register_handler` disappears from
260    /// here rather than being recorded wrong.
261    #[tokio::test]
262    async fn register_typed_records_the_four_instance_crud_commands() {
263        let state = crate::server::tests::test_state();
264        let (engine, _rx) = WshRpcEngine::new();
265        register(&engine, &state);
266        let schema = engine.schema_json();
267        let rows = schema.as_array().unwrap();
268        let find = |cmd: &str| {
269            rows.iter()
270                .find(|r| r["command"] == cmd)
271                .unwrap_or_else(|| panic!("{cmd} missing from the schema"))
272        };
273
274        for (cmd, resp) in [
275            (crate::backend::rpc_types::COMMAND_GET_AGENT_INSTANCE, "AgentInstance"),
276            (crate::backend::rpc_types::COMMAND_CREATE_AGENT_INSTANCE, "AgentInstance"),
277            (crate::backend::rpc_types::COMMAND_UPDATE_AGENT_INSTANCE, "AgentInstance"),
278        ] {
279            assert_eq!(find(cmd)["responseName"], resp, "{cmd} response");
280        }
281        let list = find(crate::backend::rpc_types::COMMAND_LIST_AGENT_INSTANCES);
282        let list_resp = list["responseName"].as_str().unwrap();
283        assert!(
284            list_resp.starts_with("alloc::vec::Vec<") && list_resp.contains("AgentInstance"),
285            "listagentinstances should answer Vec<AgentInstance>, got {list_resp:?}",
286        );
287    }
288
289    /// `listagentinstances` used to `unwrap_or_default()` its payload. That
290    /// bought two different things, and only one of them was wanted.
291    ///
292    /// Wanted: both encodings of "no filter" work -- the `{}` the stub sends
293    /// and the `null` a client that omits `data` sends. `Option<Req>` keeps
294    /// that.
295    ///
296    /// Not wanted: a MALFORMED filter silently became "no filter", so
297    /// `{"status": 5}` returned every row instead of erroring. Answering an
298    /// unfiltered list to a request whose filter failed to parse is the wrong
299    /// answer to give quietly, so it is an error now. This test pins both
300    /// halves so the change is visible rather than incidental.
301    #[tokio::test]
302    async fn listagentinstances_accepts_no_filter_but_rejects_a_malformed_one() {
303        let state = crate::server::tests::test_state();
304        let (engine, mut rx) = WshRpcEngine::new();
305        register(&engine, &state);
306
307        let call = |payload: serde_json::Value, reqid: &str| {
308            engine.handle_message(crate::backend::rpc_types::RpcMessage {
309                command: crate::backend::rpc_types::COMMAND_LIST_AGENT_INSTANCES.to_string(),
310                reqid: reqid.to_string(),
311                data: Some(payload),
312                ..Default::default()
313            });
314        };
315
316        for (i, payload) in [serde_json::json!({}), serde_json::Value::Null]
317            .into_iter()
318            .enumerate()
319        {
320            call(payload.clone(), &format!("ok-{i}"));
321            let resp = tokio::time::timeout(std::time::Duration::from_secs(2), rx.recv())
322                .await
323                .unwrap()
324                .unwrap();
325            assert!(
326                resp.error.is_empty(),
327                "{payload} means no filter and must be accepted, got {:?}",
328                resp.error,
329            );
330        }
331
332        call(serde_json::json!({ "status": 5 }), "bad");
333        let resp = tokio::time::timeout(std::time::Duration::from_secs(2), rx.recv())
334            .await
335            .unwrap()
336            .unwrap();
337        assert!(
338            !resp.error.is_empty(),
339            "a malformed filter must not be silently answered with the unfiltered list",
340        );
341    }
342
343    /// handler used to answer with an anonymous `json!({"deleted": ..})`.
344    #[tokio::test]
345    async fn register_typed_records_deleteagentinstance() {
346        let state = test_state();
347        let (engine, _rx) = WshRpcEngine::new();
348        register(&engine, &state);
349        let schema = engine.schema_json();
350        let rows = schema.as_array().unwrap();
351        let row = rows
352            .iter()
353            .find(|r| r["command"] == COMMAND_DELETE_AGENT_INSTANCE)
354            .expect("deleteagentinstance missing from the schema");
355        assert_eq!(row["requestName"], "CommandDeleteAgentInstanceData");
356        assert_eq!(row["responseName"], "DeleteAgentInstanceResult");
357        // The other four instance commands waited for the storage-level
358        // `AgentInstance` entity to be generated, which it now is, so they are
359        // present rather than absent. The property being asserted is the same
360        // one, with the opposite sign --
361        // `register_typed_records_the_four_instance_crud_commands` covers
362        // their exact types; here just assert instance.rs has nothing left
363        // outside the registry.
364        for cmd in [
365            COMMAND_LIST_AGENT_INSTANCES,
366            COMMAND_GET_AGENT_INSTANCE,
367            COMMAND_CREATE_AGENT_INSTANCE,
368            COMMAND_UPDATE_AGENT_INSTANCE,
369        ] {
370            assert!(
371                rows.iter().any(|r| r["command"] == cmd),
372                "{cmd} is migrated and must appear in the schema",
373            );
374        }
375    }
376
377    /// `agentlastruntime`: read / replace / forget the remembered runtime, and
378    /// refuse a value the launch path could not read.
379    #[tokio::test]
380    async fn agentlastruntime_reads_replaces_forgets_and_refuses_bad_values() {
381        let state = crate::server::tests::test_state();
382        let mut agent: crate::backend::storage::AgentDefinition = serde_json::from_value(serde_json::json!({
383            "id": "rt-agent",
384            "name": "rt-agent",
385            "icon": "wand-magic-sparkles",
386            "provider": "claude",
387            "description": "",
388            "created_at": 1,
389        }))
390        .unwrap();
391        state.mstore.agent_def_insert(&mut agent).unwrap();
392        let (engine, mut rx) = WshRpcEngine::new();
393        register(&engine, &state);
394
395        let mut n = 0;
396        let mut call = |payload: serde_json::Value| {
397            n += 1;
398            engine.handle_message(crate::backend::rpc_types::RpcMessage {
399                command: COMMAND_AGENT_LAST_RUNTIME.to_string(),
400                reqid: format!("r{n}"),
401                data: Some(payload),
402                ..Default::default()
403            });
404        };
405        macro_rules! next {
406            () => {
407                tokio::time::timeout(std::time::Duration::from_secs(2), rx.recv())
408                    .await
409                    .unwrap()
410                    .unwrap()
411            };
412        }
413
414        call(serde_json::json!({ "id": "rt-agent" }));
415        let r = next!();
416        assert!(r.error.is_empty(), "{:?}", r.error);
417        assert_eq!(r.data.unwrap()["runtime"], "");
418
419        call(serde_json::json!({ "id": "rt-agent", "runtime": r#"{"model":"opus","effort":"high"}"# }));
420        let r = next!();
421        assert!(r.error.is_empty(), "{:?}", r.error);
422        assert_eq!(r.data.unwrap()["runtime"], r#"{"effort":"high","model":"opus"}"#);
423
424        call(serde_json::json!({ "id": "rt-agent", "runtime": r#"{"colour":"red"}"# }));
425        let r = next!();
426        assert!(!r.error.is_empty(), "an unknown key is refused");
427
428        call(serde_json::json!({ "id": "rt-agent" }));
429        assert_eq!(next!().data.unwrap()["runtime"], r#"{"effort":"high","model":"opus"}"#, "a refused write changes nothing");
430
431        call(serde_json::json!({ "id": "rt-agent", "runtime": "" }));
432        assert_eq!(next!().data.unwrap()["runtime"], "");
433    }
434}
agentmux_srv/server/agent_handlers/instance.rs

agentmux_srv/server/agent_handlers/
instance.rs
