Skip to content
Closed
Show file tree
Hide file tree
Changes from 3 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
35 changes: 35 additions & 0 deletions apps/server/src/provider/Layers/CodexAdapter.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -555,6 +555,41 @@ lifecycleLayer("CodexAdapterLive lifecycle", (it) => {
}),
);

it.effect("renames an agent without overwriting its existing identity metadata", () =>
Effect.gen(function* () {
const { adapter, runtime } = yield* startLifecycleRuntime();
const firstEventFiber = yield* Stream.runHead(adapter.streamEvents).pipe(Effect.forkChild);

yield* runtime.emit({
id: asEventId("evt-agent-renamed"),
kind: "notification",
provider: ProviderDriverKind.make("codex"),
createdAt: "2026-01-01T00:00:00.000Z",
method: "collabAgent/renamed",
threadId: asThreadId("thread-1"),
payload: {
agentThreadId: "child-rename",
nickname: "Alpha",
},
} satisfies ProviderEvent);

const firstEvent = yield* Fiber.join(firstEventFiber);
NodeAssert.equal(firstEvent._tag, "Some");
if (firstEvent._tag !== "Some") {
return;
}
NodeAssert.equal(firstEvent.value.type, "task.updated");
if (firstEvent.value.type !== "task.updated") {
return;
}
NodeAssert.deepStrictEqual(firstEvent.value.payload, {
taskId: "child-rename",
title: "Alpha",
timelineBypass: true,
});
}),
);

it.effect("labels MCP lifecycle entries with server and tool names", () =>
Effect.gen(function* () {
const { adapter, runtime } = yield* startLifecycleRuntime();
Expand Down
12 changes: 12 additions & 0 deletions apps/server/src/provider/Layers/CodexAdapter.ts
Original file line number Diff line number Diff line change
Expand Up @@ -605,6 +605,18 @@ function mapCollabAgentEvent(
payload: { taskId, status: "running", ...statusLinkage },
},
];
case "collabAgent/renamed":
return [
{
Comment thread
macroscopeapp[bot] marked this conversation as resolved.
...base,
type: "task.updated",
payload: {
taskId,
...(nickname ? { title: nickname } : {}),
timelineBypass: true,
},
},
];
case "collabAgent/turnCompleted": {
// Idle, not terminal: the identity is resumable via sendInput/resume.
const turn =
Expand Down
291 changes: 291 additions & 0 deletions apps/server/src/provider/Layers/CodexCollabRuntime.integration.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,8 @@ import { makeCodexSessionRuntime } from "./CodexSessionRuntime.ts";

const ROOT = wireFixture.rootThreadId;
const [CHILD_A, CHILD_B] = wireFixture.childThreadIds as [string, string];
const CHILD_C = "019fcfd6-1806-7de1-8564-de69fd55bff3";
const CHILD_D = "019fcfd6-1806-7de1-8564-de69fd55bff4";

/**
* The captured sequence, extended with the shapes the live capture didn't
Expand Down Expand Up @@ -70,10 +72,299 @@ function buildScript() {
};
}

function buildDirectChildScript() {
const rootTurnId = "019fcfd6-1806-7de1-8564-de69fd55bffb";
const childTurnId = `${CHILD_A}-direct-turn`;
const childItemId = `${CHILD_A}-direct-message`;
return {
rootThreadId: ROOT,
notifications: [
{
method: "item/completed",
params: {
threadId: ROOT,
turnId: rootTurnId,
item: {
type: "collabAgentToolCall",
id: "call_direct_spawn",
tool: "spawnAgent",
status: "completed",
senderThreadId: ROOT,
receiverThreadIds: [CHILD_A],
prompt: "Return one concise result.",
agentsStates: {
[CHILD_A]: { status: "pendingInit", message: null },
},
},
completedAtMs: 1785898350000,
},
},
{
method: "item/started",
params: {
threadId: CHILD_A,
turnId: childTurnId,
item: { type: "agentMessage", id: childItemId, text: "" },
},
},
{
method: "item/agentMessage/delta",
params: {
threadId: CHILD_A,
turnId: childTurnId,
itemId: childItemId,
delta: "child narration must not enter the parent transcript",
},
},
{
method: "item/completed",
params: {
threadId: CHILD_A,
turnId: childTurnId,
completedAtMs: 1785898350000,
item: {
type: "agentMessage",
id: childItemId,
phase: "final_answer",
text: "child result is consumed by the parent model",
},
},
},
{
method: "item/completed",
params: {
threadId: ROOT,
turnId: rootTurnId,
item: {
type: "agentMessage",
id: "root-summary",
phase: "final_answer",
text: "The agent completed:\n- Direct Child Researcher: returned one concise result.",
},
completedAtMs: 1785898350000,
},
},
],
};
}

function buildOutOfOrderNamingScript() {
const rootTurnId = "019fcfd6-1806-7de1-8564-de69fd55bffb";
return {
rootThreadId: ROOT,
notifications: [
// Child B speaks before the parent-side fan-out arrives. This makes
// provisional registration order differ from receiverThreadIds order.
{
method: "turn/started",
params: {
threadId: CHILD_B,
turn: {
id: `${CHILD_B}-early-turn`,
items: [],
itemsView: "notLoaded",
status: "inProgress",
error: null,
startedAt: 1785898342,
completedAt: null,
durationMs: null,
},
},
},
{
method: "item/completed",
params: {
threadId: ROOT,
turnId: rootTurnId,
item: {
type: "collabAgentToolCall",
id: "call_direct_spawn_ordered",
tool: "spawnAgent",
status: "completed",
senderThreadId: ROOT,
receiverThreadIds: [CHILD_A, CHILD_B],
prompt: "Return one concise result.",
agentsStates: {
[CHILD_A]: { status: "pendingInit", message: null },
[CHILD_B]: { status: "pendingInit", message: null },
},
},
completedAtMs: 1785898350000,
},
},
{
method: "item/completed",
params: {
threadId: ROOT,
turnId: rootTurnId,
item: {
type: "collabAgentToolCall",
id: "call_direct_spawn_ordered_second",
tool: "spawnAgent",
status: "completed",
senderThreadId: ROOT,
receiverThreadIds: [CHILD_C, CHILD_D],
prompt: "Return one concise result.",
agentsStates: {
[CHILD_C]: { status: "pendingInit", message: null },
[CHILD_D]: { status: "pendingInit", message: null },
},
},
completedAtMs: 1785898350000,
},
},
{
method: "item/completed",
params: {
threadId: ROOT,
turnId: rootTurnId,
item: {
type: "collabAgentToolCall",
id: "call_direct_wait_reordered",
tool: "wait",
status: "completed",
senderThreadId: ROOT,
// Non-spawn calls can list receivers in any order, but they must
// not change the original spawn positions.
receiverThreadIds: [CHILD_D, CHILD_C],
agentsStates: {
[CHILD_C]: { status: "running", message: null },
[CHILD_D]: { status: "running", message: null },
},
},
completedAtMs: 1785898350000,
},
},
{
method: "item/completed",
params: {
threadId: ROOT,
turnId: rootTurnId,
item: {
type: "agentMessage",
id: "root-order-summary",
phase: "final_answer",
text: "Agents:\n- Alpha\n- Beta\n- Gamma\n- Delta",
},
completedAtMs: 1785898350000,
},
},
],
};
}

const scriptPath = NodePath.join(import.meta.dirname, "../testFixtures/.collab-script.json");
const peerPath = NodePath.join(import.meta.dirname, "../testFixtures/codexCollabMockPeer.sh");

describe("CodexSessionRuntime collab integration", () => {
it.effect("registers receiver ids and keeps child narration out of the parent stream", () =>
Effect.gen(function* () {
// @effect-diagnostics-next-line preferSchemaOverJson:off
NodeFS.writeFileSync(scriptPath, JSON.stringify(buildDirectChildScript()), "utf8");
yield* Effect.addFinalizer(() =>
Effect.sync(() => NodeFS.rmSync(scriptPath, { force: true })),
);

const runtime = yield* makeCodexSessionRuntime({
threadId: ThreadId.make("thread-collab-direct-child"),
binaryPath: peerPath,
cwd: "/tmp",
runtimeMode: "full-access",
environment: { ...process.env, T3_CODEX_COLLAB_SCRIPT: scriptPath },
});

const eventsFiber = yield* runtime.events.pipe(
Stream.takeUntil((event) => event.method === "turn/completed"),
Stream.runCollect,
Effect.forkScoped,
);

yield* runtime.start();
yield* runtime.sendTurn({ input: "direct child" });

const events = Array.from(yield* Fiber.join(eventsFiber));
const methods = events.map((event) => event.method);
assert.include(methods, "collabAgent/started");
assert.include(methods, "collabAgent/item");
assert.include(methods, "collabAgent/statusChanged");
assert.include(methods, "collabAgent/renamed");
assert.include(methods, "turn/completed");
assert.notInclude(methods, "item/agentMessage/delta");
const started = events.find((event) => event.method === "collabAgent/started");
assert.isUndefined((started?.payload as { nickname?: string } | undefined)?.nickname);
const renamed = events.find((event) => event.method === "collabAgent/renamed");
assert.equal(
(renamed?.payload as { nickname?: string } | undefined)?.nickname,
"Direct Child Researcher",
);
const statusChanged = events.find(
(event) =>
event.method === "collabAgent/statusChanged" &&
(event.payload as { status?: { type?: string } } | undefined)?.status?.type === "idle",
);
assert.deepEqual((statusChanged?.payload as { status?: unknown } | undefined)?.status, {
type: "idle",
});

const leakedChildEvents = events.filter((event) => {
if (event.method === "item/agentMessage/delta") return true;
const payload = event.payload as { threadId?: string } | undefined;
return payload?.threadId === CHILD_A;
});
assert.deepEqual(
leakedChildEvents.map((event) => event.method),
[],
"child notifications must not be emitted as parent-timeline events",
);

yield* runtime.close;
}).pipe(Effect.scoped, Effect.provide(NodeServices.layer)),
);

it.effect("assigns names in global spawn order and ignores later non-spawn receiver order", () =>
Effect.gen(function* () {
// @effect-diagnostics-next-line preferSchemaOverJson:off
NodeFS.writeFileSync(scriptPath, JSON.stringify(buildOutOfOrderNamingScript()), "utf8");
yield* Effect.addFinalizer(() =>
Effect.sync(() => NodeFS.rmSync(scriptPath, { force: true })),
);

const runtime = yield* makeCodexSessionRuntime({
threadId: ThreadId.make("thread-collab-name-order"),
binaryPath: peerPath,
cwd: "/tmp",
runtimeMode: "full-access",
environment: { ...process.env, T3_CODEX_COLLAB_SCRIPT: scriptPath },
});

const eventsFiber = yield* runtime.events.pipe(
Stream.takeUntil((event) => event.method === "turn/completed"),
Stream.runCollect,
Effect.forkScoped,
);

yield* runtime.start();
yield* runtime.sendTurn({ input: "name the children" });

const events = Array.from(yield* Fiber.join(eventsFiber));
const renamed = events.filter((event) => event.method === "collabAgent/renamed");
assert.deepEqual(
renamed.map((event) => {
const payload = event.payload as { agentThreadId?: string; nickname?: string };
return [payload.agentThreadId, payload.nickname];
}),
[
[CHILD_A, "Alpha"],
[CHILD_B, "Beta"],
[CHILD_C, "Gamma"],
[CHILD_D, "Delta"],
],
);

yield* runtime.close;
}).pipe(Effect.scoped, Effect.provide(NodeServices.layer)),
);

it.effect("replays the captured fan-out into synthetic agent events without child leaks", () =>
Effect.gen(function* () {
// @effect-diagnostics-next-line preferSchemaOverJson:off
Expand Down
Loading
Loading