fix(agent): index external assistant replies for unread tracking

plugins/agent/facts/src/executor.rs

@@ -21,6 +21,14 @@ use serde::{Deserialize, Serialize};
 mod output;
 pub use output::{Output, ToolResultDetails, TurnUsage};
 
+pub fn message_id(
+    run: crate::durable::RunId,
+    task: crate::durable::ExternalTaskId,
+    id: &str,
+) -> String {
+    format!("executor/{run}/{task}/{id}")
+}
+
 /// Executor-owned model selection, independent of Host model connections.
 #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
 #[serde(rename_all = "camelCase", deny_unknown_fields)]

plugins/agent/native/src/service/delivery.rs

@@ -33,10 +33,6 @@ pub(super) fn executor_tool_id(run: RunId, task: ExternalTaskId, id: &str) -> St
     maka_agent_facts::tool_call::tool_use_id(run, &format!("{task}:{id}"))
 }
 
-pub(super) fn executor_message_id(run: RunId, task: ExternalTaskId, id: &str) -> String {
-    format!("executor/{run}/{task}/{id}")
-}
-
 pub(super) fn executor_tool_display(
     name: &str,
     input: &serde_json::Value,
@@ -243,7 +239,7 @@ pub(super) fn project(item: &HistoryItem, turn_id: String) -> Result<Delivery, S
             match output {
                 Output::OutputDelta { message_id, .. }
                 | Output::ThinkingDelta { message_id, .. } => DeliveryFact::ExecutorTextChanged {
-                    message_id: executor_message_id(run, *task, message_id),
+                    message_id: maka_agent_facts::executor::message_id(run, *task, message_id),
                     thinking: matches!(output, Output::ThinkingDelta { .. }),
                 },
                 Output::ToolStart {

plugins/agent/native/src/service/history/projection.rs

@@ -729,7 +729,7 @@ pub(super) async fn project(
                     let message = executor
                         .and_then(|transcript| transcript.text(message_id, thinking))
                         .ok_or("agent_history_executor_text_missing")?;
-                    row.chunk.message_id = Some(super::super::delivery::executor_message_id(
+                    row.chunk.message_id = Some(maka_agent_facts::executor::message_id(
                         item.recorded.run,
                         *task,
                         message_id,

plugins/agent/state/migrations/0039_external_catalog_messages.sql

@@ -0,0 +1,30 @@
+-- Licensed to the Apache Software Foundation (ASF) under one
+-- or more contributor license agreements. See the NOTICE file
+-- distributed with this work for additional information
+-- regarding copyright ownership. The ASF licenses this file
+-- to you under the Apache License, Version 2.0 (the
+-- "License"); you may not use this file except in compliance
+-- with the License. You may obtain a copy of the License at
+--
+--     http://www.apache.org/licenses/LICENSE-2.0
+--
+-- Unless required by applicable law or agreed to in writing,
+-- software distributed under the License is distributed on an
+-- "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+-- KIND, either express or implied. See the License for the
+-- specific language governing permissions and limitations
+-- under the License.
+
+-- Use the same message identity as executor history rows.
+INSERT INTO agent_catalog_messages(owner,session,sequence,message,at,preview)
+SELECT r.owner,r.session,f.sequence,
+    'executor/'||f.run||'/'||json_extract(b.payload,'$.fact.task')||'/'||
+        json_extract(b.payload,'$.fact.output.messageId'),
+    f.at,substr(json_extract(b.payload,'$.fact.output.text'),1,160)
+FROM agent_runs r
+CROSS JOIN agent_facts f ON f.run=r.run
+CROSS JOIN contents b ON b.id=f.body_content_id
+WHERE r.external IS NOT NULL
+  AND json_extract(b.payload,'$.fact.kind')='external_task_observed'
+  AND json_extract(b.payload,'$.fact.output.type')='output_delta'
+  AND json_extract(b.payload,'$.fact.output.text')<>'';

plugins/agent/state/src/lib.rs

@@ -215,4 +215,9 @@ pub const MIGRATIONS: &[maka_event_log::migration::Migration] = &[
         description: "subagent presets",
         sql: include_str!("../migrations/0038_subagent_presets.sql"),
     },
+    maka_event_log::migration::Migration {
+        version: 39,
+        description: "index external assistant messages in the session catalog",
+        sql: include_str!("../migrations/0039_external_catalog_messages.sql"),
+    },
 ];

plugins/agent/state/src/session_catalog/index.rs

@@ -98,6 +98,14 @@ pub(crate) fn fact_message(
                     .collect::<String>(),
             ),
         )),
+        Fact::ExternalTaskObserved {
+            task,
+            output: maka_agent_facts::executor::Output::OutputDelta { message_id, text },
+            ..
+        } if !text.is_empty() => Some((
+            maka_agent_facts::executor::message_id(fact.run, *task, message_id),
+            Some(text.chars().take(160).collect()),
+        )),
         _ => None,
     }
 }

products/maka/cli/tests/external_jobs.rs

@@ -1255,6 +1255,23 @@ async fn antigravity_can_continue_after_host_restarts_during_a_prompt() {
         assert_eq!(events.iter().filter(|e| e["kind"] == "hello").count(), 2, "{events:?}");
         assert!(events.iter().all(|e| e["kind"] != "failure"), "{events:?}");
         phase("resumed Run settles before Host quit", readiness(&recovery, &session, &next)).await?;
+        let observed = query(&recovery, "agent", "observe", json!({"sessionId":session})).await?;
+        assert_eq!(observed["session"]["hasUnread"], true);
+        let history = query(&recovery, "agent", "history", json!({
+            "sessionId":session,"position":{"kind":"tail"},"includeThinking":false,"view":"chat"
+        })).await?;
+        let final_reply = history["chunks"].as_array().unwrap().iter().rev()
+            .find(|chunk| chunk["role"] == "assistant" && chunk["text"] == "ACP follow-up final reply")
+            .ok_or("external final reply missing from history")?;
+        let message = final_reply["messageId"].as_str().ok_or("final reply identity missing")?;
+        assert_eq!(observed["session"]["tailMessageId"], message);
+        let read = recovery.control(Uuid::new_v4(), "markRead", json!({
+            "sessionId":session,"messageId":message
+        })).await?;
+        read.prepare().await?;
+        receipt_value(&read.confirm().await?)?;
+        let observed = query(&recovery, "agent", "observe", json!({"sessionId":session})).await?;
+        assert_eq!(observed["session"]["hasUnread"], false);
         let external = query(&recovery, "agent", "externalAgents", json!({"inspect":true})).await?;
         let target = &external["target"];
         let source = query(&recovery, "management", "read", json!({})).await?;

products/maka/cli/tests/support/external_codex_cli.mjs

@@ -305,10 +305,15 @@ async function acp(id, method, params) {
     if (job) throw Error('Duplicate ACP task prompt');
     job = JSON.parse(text.slice(start + 'MAKA_EXTERNAL_CASE '.length));
     const checkpoint = `ACP checkpoint ${job.label}`;
-    await event('session/update', {
-      sessionId: thread,
-      update: { sessionUpdate: 'agent_message_chunk', content: { type: 'text', text: checkpoint } },
-    });
+    if (job.label !== 'ACP-followup') {
+      await event('session/update', {
+        sessionId: thread,
+        update: {
+          sessionUpdate: 'agent_message_chunk',
+          content: { type: 'text', text: checkpoint },
+        },
+      });
+    }
     await record('ready', { thread, checkpoint, promptId: id });
     if (job.label === 'ACP-schedule') {
       const task = await codeResult(
@@ -328,7 +333,16 @@ async function acp(id, method, params) {
       await record('scheduled', { thread, taskId: task.taskId, promptId: id });
       return;
     }
-    if (job.label === 'ACP-followup') await readArchivedResult();
+    if (job.label === 'ACP-followup') {
+      await readArchivedResult();
+      await event('session/update', {
+        sessionId: thread,
+        update: {
+          sessionUpdate: 'agent_message_chunk',
+          content: { type: 'text', text: 'ACP follow-up final reply' },
+        },
+      });
+    }
     if (job.label === 'ACP-followup' || job.label === 'ACP-wake')
       return reply(id, { stopReason: 'end_turn' });
     return;