fix(workbench): manage child conversations and read archived sessions

plugins/agent/native/src/owner.rs

@@ -993,7 +993,7 @@ impl Owner {
             .get_session::<StoredSessionConfiguration>(&input.session_id)
             .await
             .map_err(|_| "agent_store_unavailable")?;
-        if session.is_none_or(|session| session.archived) {
+        if session.is_none() {
             return Ok(MarkReadResult::Rejected {
                 reason: Rejection::SessionUnavailable,
             });

plugins/agent/native/src/owner/tests/session_availability.rs

@@ -43,6 +43,8 @@ async fn session_read_guards_reject_foreign_missing_and_logically_removed_sessio
         ];
         for id in ["foreign_session", missing.as_str(), session.as_str()] {
             assert_eq!(fixture.owner.require_session(id).await.unwrap_err(), "agent_session_unavailable");
+            assert_eq!(fixture.mutation(2, "markRead", json!({"sessionId":id,"messageId":"unavailable-tail"}), Uuid::new_v4()).await.unwrap(),
+                json!({"kind":"rejected","reason":"session_unavailable"}));
             for (method, input) in &requests {
                 let mut input = input.clone();
                 input["sessionId"] = json!(id);
@@ -127,3 +129,99 @@ async fn history_visibility_does_not_decode_configuration_but_source_still_requi
     .await
     .unwrap();
 }
+
+#[tokio::test]
+async fn archived_session_mark_read_acknowledges_only_the_exact_tail_without_restoring() {
+    tokio::time::timeout(Duration::from_secs(30), async {
+        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let fixture = Fixture::new(
+            &format!("http://{}/v1", listener.local_addr().unwrap()),
+            None,
+        )
+        .await;
+        let session = fixture.create("Archived unread conversation").await;
+        let mut tails = Vec::new();
+        for _ in 0..2 {
+            fixture.send(&session).await.unwrap();
+            let (mut socket, _) = listener.accept().await.unwrap();
+            request(&mut socket).await;
+            response(&mut socket).await;
+            fixture.drained().await;
+            let observed = fixture
+                .query("observe", json!({"sessionId":session}))
+                .await
+                .unwrap();
+            assert_eq!(observed["session"]["hasUnread"], true);
+            tails.push(
+                observed["session"]["tailMessageId"]
+                    .as_str()
+                    .unwrap()
+                    .to_owned(),
+            );
+        }
+        assert_ne!(tails[0], tails[1]);
+        let revision = fixture.owner.session(&session).await.unwrap().revision;
+        let archived = fixture
+            .mutation(
+                2,
+                "changeSession",
+                json!({
+                    "sessionId":session,"expectedRevision":revision,"action":"archive"
+                }),
+                Uuid::new_v4(),
+            )
+            .await
+            .unwrap();
+        assert_eq!(archived["kind"], "changed");
+        assert_eq!(archived["session"]["archived"], true);
+        assert_eq!(archived["session"]["hasUnread"], true);
+
+        let stale = fixture
+            .mutation(
+                2,
+                "markRead",
+                json!({
+                    "sessionId":session,"messageId":tails[0]
+                }),
+                Uuid::new_v4(),
+            )
+            .await
+            .unwrap();
+        assert_eq!(stale["kind"], "observed");
+        assert_eq!(stale["session"]["hasUnread"], true);
+        assert_eq!(
+            stale["session"]["revision"],
+            archived["session"]["revision"]
+        );
+
+        let read = fixture
+            .mutation(
+                2,
+                "markRead",
+                json!({
+                    "sessionId":session,"messageId":tails[1]
+                }),
+                Uuid::new_v4(),
+            )
+            .await
+            .unwrap();
+        assert_eq!(read["kind"], "observed");
+        assert_eq!(read["session"]["archived"], true);
+        assert_eq!(read["session"]["hasUnread"], false);
+        let observed = fixture
+            .query("observe", json!({"sessionId":session}))
+            .await
+            .unwrap();
+        assert_eq!(observed["session"]["archived"], true);
+        assert_eq!(observed["session"]["hasUnread"], false);
+        let stored = fixture.owner.session(&session).await.unwrap();
+        assert_eq!(
+            stored.read_state.last_read_message_id.as_deref(),
+            Some(tails[1].as_str())
+        );
+        fixture.app.shutdown().await.unwrap();
+        fixture.config.shutdown().await.unwrap();
+    })
+    .await
+    .unwrap();
+}

products/maka/terminal-shell/tests/real_host/desktop_new_session.rs

@@ -2264,7 +2264,7 @@ async fn desktop_parent_reviews_child_access_and_opens_attributed_sender_beside(
 }
 
 #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
-async fn desktop_child_conversations_expand_archived_ancestors_and_open_a_grandchild() {
+async fn desktop_child_conversations_manage_nested_and_archived_workers() {
     use maka_agent_api::session::StoredSessionConfiguration;
     let fixture = Fixture::open().await;
     let original = fixture
@@ -2280,9 +2280,11 @@ async fn desktop_child_conversations_expand_archived_ancestors_and_open_a_grandc
     let prefix = fixture.session.rsplit_once('_').unwrap().0;
     let child = format!("{prefix}_{}", Uuid::new_v4().simple());
     let grandchild = format!("{prefix}_{}", Uuid::new_v4().simple());
+    let disposable = format!("{prefix}_{}", Uuid::new_v4().simple());
     for (id, parent, title) in [
         (&child, &fixture.session, "Archived worker"),
         (&grandchild, &child, "Verification worker"),
+        (&disposable, &fixture.session, "Disposable worker"),
     ] {
         let mut configuration = original.clone();
         configuration.parent_session_id = Some(parent.clone());
@@ -2334,6 +2336,87 @@ async fn desktop_child_conversations_expand_archived_ancestors_and_open_a_grandc
         "deep rows stay inside the dialog's viewport"
     );
     desktop.artifact(&format!("child-conversations-{theme}-{layout}"));
+    for (id, title, action, archived) in [
+        (
+            &grandchild,
+            "Verification worker",
+            "Archive conversation",
+            true,
+        ),
+        (
+            &grandchild,
+            "Verification worker",
+            "Restore conversation",
+            false,
+        ),
+        (&child, "Archived worker", "Restore conversation", false),
+        (
+            &disposable,
+            "Disposable worker",
+            "Archive conversation",
+            true,
+        ),
+    ] {
+        desktop.right_click(&format!("Open {title}")).await;
+        desktop.control(action).await;
+        desktop.artifact(&format!(
+            "child-menu-{}-{theme}-{layout}",
+            if archived { "archive" } else { "restore" }
+        ));
+        desktop.press(action).await;
+        desktop
+            .wait("tree action did not change the targeted worker", async {
+                loop {
+                    let observed = desktop
+                        .client
+                        .query("agent", "observe", json!({"sessionId":id}))
+                        .await
+                        .unwrap();
+                    if observed["session"]["archived"] == archived {
+                        break;
+                    }
+                    tokio::task::yield_now().await;
+                }
+            })
+            .await;
+        desktop
+            .until("tree action menu did not settle", || {
+                (desktop.controls(action).is_empty()
+                    && desktop.controls("Close child conversations").len() == 1)
+                    .then_some(())
+            })
+            .await;
+        desktop.control("Open Verification worker").await;
+    }
+    desktop.press("Actions for Disposable worker").await;
+    desktop.press("Delete conversation").await;
+    desktop.control("Confirm delete conversation").await;
+    assert!(
+        fixture
+            .store
+            .get_session::<StoredSessionConfiguration>(&disposable)
+            .await
+            .unwrap()
+            .is_some(),
+        "opening delete confirmation must not delete the worker"
+    );
+    desktop.artifact(&format!("child-delete-confirmation-{theme}-{layout}"));
+    desktop.press("Confirm delete conversation").await;
+    desktop
+        .until("deleted worker remained in the tree", || {
+            (desktop.controls("Open Disposable worker").is_empty()
+                && desktop.controls("Close child conversations").len() == 1)
+                .then_some(())
+        })
+        .await;
+    assert!(
+        fixture
+            .store
+            .get_session::<StoredSessionConfiguration>(&disposable)
+            .await
+            .unwrap()
+            .is_none()
+    );
     let previous = desktop.platform.text_input_target();
     desktop.press("Open Verification worker").await;
     desktop.target_after(previous).await;

products/maka/workbench/app/SessionActions.tsx

@@ -360,6 +360,16 @@ export function SessionActions(props: {
           session={session()}
           current={() => active() && page() === 'children'}
           close={props.close}
+          renderActions={(target) => (
+            <SessionActions
+              session={target.session}
+              anchor={target.anchor}
+              current={target.current}
+              close={target.close}
+              notice={props.notice}
+              change={(action) => target.run(() => props.change(action))}
+            />
+          )}
           openSession={(id) => {
             props.close();
             props.openSession?.(id);