From 761d09c5ed0ab1993609434b7f1ff422e5f5bda4 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E2=80=9CM4n5ter=E2=80=9D?= Date: Sun, 15 Nov 2026 00:16:26 +0800 Subject: [PATCH] fix(agent): index external assistant replies for unread tracking Generated-by: OpenAI Codex --- plugins/agent/facts/src/executor.rs | 8 +++++ plugins/agent/native/src/service/delivery.rs | 6 +--- .../native/src/service/history/projection.rs | 2 +- .../0039_external_catalog_messages.sql | 30 +++++++++++++++++++ plugins/agent/state/src/lib.rs | 5 ++++ .../agent/state/src/session_catalog/index.rs | 8 +++++ products/maka/cli/tests/external_jobs.rs | 17 +++++++++++ .../cli/tests/support/external_codex_cli.mjs | 24 +++++++++++---- 8 files changed, 89 insertions(+), 11 deletions(-) create mode 100644 plugins/agent/state/migrations/0039_external_catalog_messages.sql diff --git a/plugins/agent/facts/src/executor.rs b/plugins/agent/facts/src/executor.rs index 65b3b4a895..7c57848f4a 100644 --- a/plugins/agent/facts/src/executor.rs +++ b/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)] diff --git a/plugins/agent/native/src/service/delivery.rs b/plugins/agent/native/src/service/delivery.rs index a4ded38397..251f44d9b6 100644 --- a/plugins/agent/native/src/service/delivery.rs +++ b/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 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 { diff --git a/plugins/agent/native/src/service/history/projection.rs b/plugins/agent/native/src/service/history/projection.rs index 098f012dcc..c02249e811 100644 --- a/plugins/agent/native/src/service/history/projection.rs +++ b/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, diff --git a/plugins/agent/state/migrations/0039_external_catalog_messages.sql b/plugins/agent/state/migrations/0039_external_catalog_messages.sql new file mode 100644 index 0000000000..f0d4e2e66c --- /dev/null +++ b/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')<>''; diff --git a/plugins/agent/state/src/lib.rs b/plugins/agent/state/src/lib.rs index 13edf41729..8300927fdc 100644 --- a/plugins/agent/state/src/lib.rs +++ b/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"), + }, ]; diff --git a/plugins/agent/state/src/session_catalog/index.rs b/plugins/agent/state/src/session_catalog/index.rs index 24a16977dc..748e9939cc 100644 --- a/plugins/agent/state/src/session_catalog/index.rs +++ b/plugins/agent/state/src/session_catalog/index.rs @@ -98,6 +98,14 @@ pub(crate) fn fact_message( .collect::(), ), )), + 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, } } diff --git a/products/maka/cli/tests/external_jobs.rs b/products/maka/cli/tests/external_jobs.rs index b140dc29d5..92607155b8 100644 --- a/products/maka/cli/tests/external_jobs.rs +++ b/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?; diff --git a/products/maka/cli/tests/support/external_codex_cli.mjs b/products/maka/cli/tests/support/external_codex_cli.mjs index a3ad0be6bd..eabe1d4a14 100644 --- a/products/maka/cli/tests/support/external_codex_cli.mjs +++ b/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;