diff --git a/extensions/codex/src/app-server/event-projector-events.ts b/extensions/codex/src/app-server/event-projector-events.ts index 72bd1b87ab98..9e98aa8188f2 100644 --- a/extensions/codex/src/app-server/event-projector-events.ts +++ b/extensions/codex/src/app-server/event-projector-events.ts @@ -1,4 +1,7 @@ -import type { EmbeddedRunAttemptParamsV2 as EmbeddedRunAttemptParams } from "openclaw/plugin-sdk/agent-harness-runtime"; +import type { + EmbeddedRunAttemptParamsV2 as EmbeddedRunAttemptParams, + ToolProgressDetailMode, +} from "openclaw/plugin-sdk/agent-harness-runtime"; import { asFiniteNumber, readStringField as readString, @@ -28,6 +31,55 @@ import { isJsonObject, type CodexThreadItem, type JsonObject } from "./protocol. type AgentEvent = Parameters>[0]; +type NormalizedToolItemProjection = { + name: string; + status: ReturnType; + args: Record | undefined; + meta: string | undefined; + event: AgentEvent | undefined; +}; + +export function projectNormalizedToolItem(params: { + phase: "start" | "result"; + item: CodexThreadItem | undefined; + detailMode?: ToolProgressDetailMode; +}): NormalizedToolItemProjection | undefined { + const { item } = params; + if (!item || !shouldSynthesizeToolProgressForItem(item)) { + return undefined; + } + const name = itemName(item); + if (!name) { + return undefined; + } + const status = params.phase === "result" ? itemStatus(item) : "running"; + const args = itemToolArgs(item); + const commandBearing = isCommandBearingToolItem(item, args); + const meta = itemMeta(item, params.detailMode); + const event = shouldEmitTranscriptToolProgress(name, args) + ? { + stream: "tool", + data: { + phase: params.phase, + name, + itemId: item.id, + toolCallId: item.id, + ...(meta ? { meta } : {}), + ...(commandBearing ? { commandBearing: true as const } : {}), + ...(params.phase === "start" && args ? { args } : {}), + ...(params.phase === "result" + ? { + status, + isError: isNonSuccessItemStatus(status), + ...itemToolResult(item), + } + : {}), + }, + } + : undefined; + return { name, status, args, meta, event }; +} + export class CodexEventProjection { private reviewCount = 0; @@ -138,48 +190,27 @@ export class CodexEventProjection { phase: "start" | "result"; item: CodexThreadItem | undefined; }): Promise { + const projection = projectNormalizedToolItem({ + ...params, + detailMode: this.toolProgress.toolProgressDetailMode(), + }); + if (!projection || !params.item) { + return; + } const { item } = params; - if (!item || !shouldSynthesizeToolProgressForItem(item)) { - return; - } - const name = itemName(item); - if (!name) { - return; - } - const status = params.phase === "result" ? itemStatus(item) : "running"; - const args = itemToolArgs(item); - const commandBearing = isCommandBearingToolItem(item, args); - const meta = itemMeta(item, this.toolProgress.toolProgressDetailMode()); + const { name, status, args, meta, event } = projection; this.toolTranscript.recordTrajectoryEvent({ phase: params.phase, item, name, args, status }); if (params.phase === "result") { this.toolProgress.recordNativeToolError({ item, name, meta, status }); } - if (!shouldEmitTranscriptToolProgress(name, args)) { + if (!event) { if (params.phase === "result") { this.toolTranscript.emitAfterToolCallObservation(item); await this.onNativeToolResultRecorded?.(); } return; } - this.emitAgentEvent({ - stream: "tool", - data: { - phase: params.phase, - name, - itemId: item.id, - toolCallId: item.id, - ...(meta ? { meta } : {}), - ...(commandBearing ? { commandBearing: true } : {}), - ...(params.phase === "start" && args ? { args } : {}), - ...(params.phase === "result" - ? { - status, - isError: isNonSuccessItemStatus(status), - ...itemToolResult(item), - } - : {}), - }, - }); + this.emitAgentEvent(event); if (params.phase === "result") { this.toolTranscript.emitAfterToolCallObservation(item); await this.onNativeToolResultRecorded?.(); diff --git a/extensions/codex/src/app-server/native-subagent-monitor.test.ts b/extensions/codex/src/app-server/native-subagent-monitor.test.ts index 482fb5392772..65a40ce259a9 100644 --- a/extensions/codex/src/app-server/native-subagent-monitor.test.ts +++ b/extensions/codex/src/app-server/native-subagent-monitor.test.ts @@ -1,3 +1,4 @@ +import { onAgentEvent } from "openclaw/plugin-sdk/agent-harness-runtime"; // Codex tests cover native subagent monitor plugin behavior. import type { AgentHarnessScopedSetDeliveryStatusParams, @@ -715,6 +716,78 @@ describe("CodexNativeSubagentMonitor", () => { client.close(); }); + it("publishes child assistant and tool activity under the mirrored thread run id", async () => { + const events: Array<{ runId: string; stream: string; data: Record }> = []; + const unsubscribe = onAgentEvent((event) => events.push(event)); + const client = createClient(); + const runtime = createRuntime(); + const monitor = new CodexNativeSubagentMonitor(client as never, runtime); + try { + registerParent(monitor); + await notifyChildStarted(client); + await client.notify({ + method: "item/agentMessage/delta", + params: { + threadId: "child-thread", + turnId: "child-turn", + itemId: "assistant-1", + delta: "Inspecting the registry", + }, + }); + await client.notify({ + method: "item/reasoning/summaryTextDelta", + params: { + threadId: "child-thread", + turnId: "child-turn", + itemId: "reasoning-1", + summaryIndex: 0, + delta: "Planning the fix", + }, + }); + await client.notify({ + method: "item/started", + params: { + threadId: "child-thread", + turnId: "child-turn", + item: { + type: "commandExecution", + id: "command-1", + command: "pnpm test", + cwd: "/workspace", + status: "inProgress", + }, + }, + }); + + expect(events).toEqual( + expect.arrayContaining([ + expect.objectContaining({ + runId: "codex-thread:child-thread", + stream: "assistant", + data: expect.objectContaining({ delta: "Inspecting the registry" }), + }), + expect.objectContaining({ + runId: "codex-thread:child-thread", + stream: "thinking", + data: expect.objectContaining({ delta: "Planning the fix" }), + }), + expect.objectContaining({ + runId: "codex-thread:child-thread", + stream: "tool", + data: expect.objectContaining({ + phase: "start", + name: "bash", + toolCallId: "command-1", + }), + }), + ]), + ); + } finally { + unsubscribe(); + client.close(); + } + }); + it("delivers a completed child turn from its terminal snapshot", async () => { const client = createClient(); const runtime = createRuntime(); diff --git a/extensions/codex/src/app-server/native-subagent-monitor.ts b/extensions/codex/src/app-server/native-subagent-monitor.ts index fee7dde601af..50d200b86071 100644 --- a/extensions/codex/src/app-server/native-subagent-monitor.ts +++ b/extensions/codex/src/app-server/native-subagent-monitor.ts @@ -2,7 +2,11 @@ * Mirrors Codex native subagent lifecycle and completion into OpenClaw task * runtime records, with app-server history as the recovery source. */ -import { embeddedAgentLog, formatErrorMessage } from "openclaw/plugin-sdk/agent-harness-runtime"; +import { + embeddedAgentLog, + emitAgentEvent, + formatErrorMessage, +} from "openclaw/plugin-sdk/agent-harness-runtime"; import { createAgentHarnessTaskRuntime, deliverAgentHarnessTaskCompletion, @@ -24,6 +28,8 @@ import { type CodexAppServerLiveThreadOwnership, } from "./client-runtime.js"; import type { CodexAppServerClient } from "./client.js"; +import { projectNormalizedToolItem } from "./event-projector-events.js"; +import { readItem } from "./event-projector-values.js"; import { codexNativeSubagentNotifications as nativeSubagentNotifications, type CodexNativeSubagentCompletion, @@ -140,6 +146,7 @@ const NATIVE_SUBAGENT_NOTIFICATION_METHODS = new Set([ "turn/started", "turn/completed", "item/agentMessage/delta", + "item/reasoning/summaryTextDelta", "item/started", "item/completed", // App-server exposes no typed terminal subagent result. Keep this one raw @@ -470,6 +477,9 @@ class Monitor { if (notification.method === "turn/started" && childState) { this.resumeChild(childState); } + if (childState && !childState.terminal) { + this.emitChildTaskActivity(notification, childState.childThreadId); + } this.captureChildAssistantMessage(notification); await this.handleChildTurnCompletion(notification); if (notification.method === "thread/status/changed" && threadId && threadStatus) { @@ -501,6 +511,45 @@ class Monitor { await this.handleCompletionNotification(notification); } + private emitChildTaskActivity( + notification: CodexServerNotification, + childThreadId: string, + ): void { + const params = isJsonObject(notification.params) ? notification.params : undefined; + if (!params) { + return; + } + const runId = codexNativeSubagentRunId(childThreadId); + if (notification.method === "item/agentMessage/delta") { + const delta = readString(params, "delta"); + if (delta) { + emitAgentEvent({ runId, stream: "assistant", data: { delta } }); + } + return; + } + if (notification.method === "item/reasoning/summaryTextDelta") { + const delta = readString(params, "delta"); + if (delta) { + emitAgentEvent({ runId, stream: "thinking", data: { delta } }); + } + return; + } + if (notification.method !== "item/started" && notification.method !== "item/completed") { + return; + } + const item = readItem(params.item); + if (item?.type === "agentMessage" && notification.method === "item/completed" && item.text) { + emitAgentEvent({ runId, stream: "assistant", data: { text: item.text } }); + } + const projection = projectNormalizedToolItem({ + phase: notification.method === "item/started" ? "start" : "result", + item, + }); + if (projection?.event) { + emitAgentEvent({ runId, ...projection.event }); + } + } + private resumeChild(childState: ChildState, options: { scheduleRecovery?: boolean } = {}): void { if (childState.terminal) { return; diff --git a/src/gateway/agent-turn/agent-run-admission-phase.ts b/src/gateway/agent-turn/agent-run-admission-phase.ts index cfd03aafd43f..dd19c7a3620e 100644 --- a/src/gateway/agent-turn/agent-run-admission-phase.ts +++ b/src/gateway/agent-turn/agent-run-admission-phase.ts @@ -328,6 +328,7 @@ export async function prepareAgentRunDispatch(params: { logGateway: params.context.logGateway, }), modelRun: params.isOneShotModelRun, + runId: params.runId, }); const dispatchTaskTrackingMode: PreparedAgentRunDispatch["dispatchTaskTrackingMode"] = taskTrackingMode === "cli" ? "cli" : "none"; diff --git a/src/gateway/server-methods/agent-task-tracking.ts b/src/gateway/server-methods/agent-task-tracking.ts index 96f0f7071cb6..e6abefe0eb3d 100644 --- a/src/gateway/server-methods/agent-task-tracking.ts +++ b/src/gateway/server-methods/agent-task-tracking.ts @@ -14,6 +14,7 @@ import { parseThreadSessionSuffix, } from "../../sessions/session-key-utils.js"; import { finalizeTaskRunByRunId } from "../../tasks/detached-task-runtime.js"; +import { findTaskByRunId } from "../../tasks/runtime-internal.js"; import type { TaskStatus } from "../../tasks/task-registry.types.js"; import { formatForLog } from "../ws-log.js"; import type { GatewayRequestContext, GatewayRequestHandlerOptions } from "./types.js"; @@ -88,6 +89,7 @@ export function resolveGatewayAgentTaskTrackingMode(params: { inputProvenance?: InputProvenance; confirmedAcpManualSpawn?: boolean; modelRun?: boolean; + runId?: string; }): GatewayAgentTaskTrackingMode { // Model probes are stateless one-shot work. A terminal CLI task row would // outlive the probe even when its session/transcript effects are internal. @@ -100,6 +102,15 @@ export function resolveGatewayAgentTaskTrackingMode(params: { if (params.client?.internal?.agentRunTracking === "plugin_subagent") { return "plugin_subagent"; } + // The subagent registry created the authoritative row before its host-owned + // gateway dispatch. A CLI row here would represent the same run twice. + const existingTask = params.runId ? findTaskByRunId(params.runId) : undefined; + if ( + existingTask?.runtime === "subagent" && + existingTask.childSessionKey === params.sessionKey?.trim() + ) { + return "none"; + } // A confirmed ACP manual-spawn child turn already owns its requester-visible // `acp` task row from the spawn control plane (src/agents/subagents/spawn/acp-spawn.ts). The // Gateway CLI path runs that same childRunId, so tracking it here would emit a diff --git a/src/gateway/server-methods/agent.sessions-and-models.test-utils.ts b/src/gateway/server-methods/agent.sessions-and-models.test-utils.ts index a02578af5b52..9184bcb388fc 100644 --- a/src/gateway/server-methods/agent.sessions-and-models.test-utils.ts +++ b/src/gateway/server-methods/agent.sessions-and-models.test-utils.ts @@ -2572,6 +2572,38 @@ describe("gateway agent handler", () => { }); }); + it("keeps a host-owned subagent run to its pre-registered task row", async () => { + await withTempDir({ prefix: "openclaw-gateway-subagent-owner-" }, async (root) => { + useTestStateDir(root); + resetAgentTaskRegistryForTests(); + const childSessionKey = "agent:main:subagent:owned"; + const runId = "host-owned-subagent-run"; + mockAcpChildSessionEntry(childSessionKey); + getDetachedTaskLifecycleRuntime().createRunningTaskRun({ + runtime: "subagent", + requesterSessionKey: "agent:main:main", + ownerKey: "agent:main:main", + scopeKind: "session", + childSessionKey, + runId, + task: "Run one owned subagent", + deliveryStatus: "pending", + }); + const createRunningTaskRunSpy = spyDetachedCreateRunningTaskRun(); + + await invokeAgent( + { message: "host-owned child turn", sessionKey: childSessionKey, idempotencyKey: runId }, + { reqId: runId, client: backendGatewayClient() }, + ); + await waitForAgentCommandCall(); + + expect(createRunningTaskRunSpy).not.toHaveBeenCalled(); + expect(listTaskRecords().filter((task) => task.runId === runId)).toEqual([ + expect.objectContaining({ runtime: "subagent", childSessionKey }), + ]); + }); + }); + it("keeps CLI tracking when a non-backend operator-write caller sets acpTurnSource", async () => { await withTempDir({ prefix: "openclaw-gateway-acp-operator-write-" }, async (root) => { useTestStateDir(root); diff --git a/src/gateway/server-runtime-subscriptions.test.ts b/src/gateway/server-runtime-subscriptions.test.ts index 376518a08a6b..cff940ff28bf 100644 --- a/src/gateway/server-runtime-subscriptions.test.ts +++ b/src/gateway/server-runtime-subscriptions.test.ts @@ -22,6 +22,7 @@ import { createTaskRecord, markTaskLostById, markTaskTerminalById, + recordTaskProgressByRunId, } from "../tasks/task-registry.js"; import { getTaskRegistryObservers } from "../tasks/task-registry.store.js"; import { resetTaskRegistryForTests } from "../tasks/task-runtime.test-helpers.js"; @@ -527,6 +528,50 @@ describe("startGatewayEventSubscriptions", () => { expect(broadcast).not.toHaveBeenCalled(); }); + it("suppresses identical task summaries without delaying status transitions", async () => { + const broadcast = vi.fn(); + unsubs = startGatewayEventSubscriptions({ ...createParams(), broadcast }); + await waitForFast(() => expect(getTaskRegistryObservers()).not.toBeNull()); + const runId = "run-identical-task-summary"; + const task = createTaskRecord({ + runtime: "subagent", + requesterSessionKey: "agent:main:main", + ownerKey: "agent:main:main", + scopeKind: "session", + childSessionKey: "agent:main:subagent:summary", + runId, + task: "Avoid duplicate broadcasts", + status: "running", + deliveryStatus: "not_applicable", + notifyPolicy: "silent", + startedAt: 100, + lastEventAt: 100, + }); + if (!task) { + throw new Error("expected task record"); + } + broadcast.mockClear(); + + for (let index = 0; index < 2; index += 1) { + recordTaskProgressByRunId({ + runId, + runtime: "subagent", + lastEventAt: 200, + progressSummary: "Working", + }); + } + markTaskTerminalById({ taskId: task.taskId, status: "succeeded", endedAt: 300 }); + + const taskEvents = broadcast.mock.calls + .filter(([event]) => event === "task") + .map(([, payload]) => payload as TaskEventPayload) + .filter( + (payload): payload is Extract => + payload.action === "upserted", + ); + expect(taskEvents.map((event) => event.task.status)).toEqual(["running", "completed"]); + }); + it.each(["succeeded", "failed", "cancelled", "timed_out", "lost"] as const)( "closes task-run terminals exactly once for a %s transition", async (status) => { diff --git a/src/gateway/server-runtime-subscriptions.ts b/src/gateway/server-runtime-subscriptions.ts index c7359d8d841b..e21236a09400 100644 --- a/src/gateway/server-runtime-subscriptions.ts +++ b/src/gateway/server-runtime-subscriptions.ts @@ -378,17 +378,27 @@ export function startGatewayEventSubscriptions(params: { }); let taskObserverDisposed = false; + const lastTaskSummaryById = new Map(); const taskObservers = { onEvent: (event: TaskRegistryObserverEvent) => { let payload: TaskEventPayload; switch (event.kind) { - case "upserted": - payload = { action: "upserted", task: mapTaskSummary(event.task) }; + case "upserted": { + const task = mapTaskSummary(event.task); + const summary = JSON.stringify(task); + if (lastTaskSummaryById.get(task.id) === summary) { + return; + } + lastTaskSummaryById.set(task.id, summary); + payload = { action: "upserted", task }; break; + } case "deleted": + lastTaskSummaryById.delete(event.taskId); payload = { action: "deleted", taskId: event.taskId }; break; case "restored": + lastTaskSummaryById.clear(); payload = { action: "restored" }; break; } diff --git a/src/tasks/task-registry-activity.ts b/src/tasks/task-registry-activity.ts index bfd24ef872a4..7bc03afda472 100644 --- a/src/tasks/task-registry-activity.ts +++ b/src/tasks/task-registry-activity.ts @@ -117,7 +117,27 @@ function readEditPairs(args: Record): EditPair[] { function readPatchDelta(args: Record): DiffDelta | undefined { if (typeof args.input !== "string") { - return undefined; + let added = 0; + let removed = 0; + const files: string[] = []; + for (const candidate of Array.isArray(args.changes) ? args.changes : []) { + const change = asOptionalObjectRecord(candidate); + const target = change ? readTarget(change) : undefined; + if (!change || !target) { + continue; + } + files.push(target); + const stat = asOptionalObjectRecord(change.stat); + added += + typeof stat?.added === "number" && Number.isFinite(stat.added) + ? Math.max(0, stat.added) + : 0; + removed += + typeof stat?.removed === "number" && Number.isFinite(stat.removed) + ? Math.max(0, stat.removed) + : 0; + } + return files.length > 0 ? { files, added, removed } : undefined; } const files = extractApplyPatchTargetPaths(args); if (files.length === 0) { diff --git a/src/tasks/task-registry.test.ts b/src/tasks/task-registry.test.ts index 08951b592298..f5ae737c0d37 100644 --- a/src/tasks/task-registry.test.ts +++ b/src/tasks/task-registry.test.ts @@ -33,6 +33,7 @@ import { requestFlowCancel, } from "./task-flow-registry.js"; import type { TaskFlowRecord } from "./task-flow-registry.types.js"; +import { getTaskActivitySnapshot } from "./task-registry-activity.js"; import { cancelTaskById, deleteTaskRecordById, @@ -792,6 +793,67 @@ describe("task-registry", () => { }); }); + it("folds Codex native child activity under its canonical thread run id", async () => { + await withTaskRegistryTempDir(async () => { + resetTaskRegistryMemoryForTest(); + const runId = "codex-thread:019fef4-native-child"; + const task = createTaskFixture("subagent", { + childSessionKey: runId, + runId, + task: "Inspect the ACP runtime", + startedAt: 100, + }); + + emitAgentEvent({ + runId, + stream: "assistant", + data: { delta: "Editing the native child path" }, + }); + emitAgentEvent({ + runId, + stream: "tool", + data: { phase: "start", name: "bash", toolCallId: "cmd-1" }, + }); + emitAgentEvent({ + runId, + stream: "tool", + data: { + phase: "start", + name: "apply_patch", + toolCallId: "patch-1", + args: { + changes: [ + { + path: "src/tasks/task-registry.ts", + kind: "update", + stat: { added: 5, removed: 2 }, + }, + { + path: "src/tasks/task-registry.test.ts", + kind: "update", + stat: { added: 8, removed: 0 }, + }, + ], + }, + }, + }); + emitAgentEvent({ + runId, + stream: "tool", + data: { phase: "result", name: "apply_patch", toolCallId: "patch-1", isError: false }, + }); + + expectRecordFields(requireTaskByRunId(runId), { + toolUseCount: 2, + lastToolName: "apply_patch", + }); + expect(getTaskActivitySnapshot(task.taskId)).toEqual({ + lastActivity: "Editing the native child path", + diffStat: { files: 2, added: 13, removed: 2 }, + }); + }); + }); + it("keeps subagent abort lifecycle projections provisional", async () => { await withTaskRegistryTempDir(async () => { resetTaskRegistryMemoryForTest();