ewilliams28/openclaw
1
0
Fork 0
mirror of https://github.com/openclaw/openclaw.git synced 2026-08-12 21:53:00 -06:00
Code Issues Packages Projects Releases Wiki Activity

fix(tasks): stream Codex subagent activity (#121899)

Browse Source 
* fix(tasks): stream Codex subagent activity

* fix(tasks): honor registry access boundary
This commit is contained in:
Peter Steinberger
2026-08-11 00:32:08 -07:00
committed by GitHub
parent 2b8dbc3a7b
commit e6b356f35a
10 changed files with 370 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files
extensions/codex/src/app-server/event-projector-events.ts
+63 -32
View File
@@ -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<NonNullable<EmbeddedRunAttemptParams["onAgentEvent"]>>[0];
type NormalizedToolItemProjection = {
name: string;
status: ReturnType<typeof itemStatus>;
args: Record<string, unknown> | 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<void> {
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?.();
extensions/codex/src/app-server/native-subagent-monitor.test.ts
+73
View File
@@ -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<string, unknown> }> = [];
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();
extensions/codex/src/app-server/native-subagent-monitor.ts
+50 -1
View File
@@ -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;
src/gateway/agent-turn/agent-run-admission-phase.ts
+1
View File
@@ -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";
src/gateway/server-methods/agent-task-tracking.ts
+11
View File
@@ -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
src/gateway/server-methods/agent.sessions-and-models.test-utils.ts
+32
View File
@@ -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);
src/gateway/server-runtime-subscriptions.test.ts
+45
View File
@@ -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<SubscriptionParams["broadcast"]>();
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<TaskEventPayload, { action: "upserted" }> =>
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) => {
src/gateway/server-runtime-subscriptions.ts
+12 -2
View File
@@ -378,17 +378,27 @@ export function startGatewayEventSubscriptions(params: {
});
let taskObserverDisposed = false;
const lastTaskSummaryById = new Map<string, string>();
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;
}
src/tasks/task-registry-activity.ts
+21 -1
View File
@@ -117,7 +117,27 @@ function readEditPairs(args: Record<string, unknown>): EditPair[] {
function readPatchDelta(args: Record<string, unknown>): 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) {
src/tasks/task-registry.test.ts
+62
View File
@@ -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();