bkinnightskytw
/
agent-orchestrator
mirror of https://github.com/ComposioHQ/agent-orchestrator.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

fix(codex): classify activity by payload.type for wrapped event_msg 

Real Codex sessions emit records like
`{"type":"event_msg","payload":{"type":"error",...}}` and
`{"type":"event_msg","payload":{"type":"approval_request",...}}`.
readLastJsonlEntry only exposed the top-level `type`, so the codex
plugin's activity switch matched `event_msg` and decayed to ready/idle,
never surfacing `blocked` or `waiting_input`. The approval_request/error
branches were dead code for payload-wrapped sessions, which is the exact
format this PR series is migrating to.

- readLastJsonlEntry now returns payloadType alongside lastType.
- Codex getActivityState prefers payloadType when present and classifies
  task_started/agent_reasoning as active, task_complete as ready, and
  the approval/error variants as waiting_input/blocked.
- New tests cover the payload-wrapped approval_request, exec_approval_request,
  error, task_started, and task_complete cases end-to-end.
- Core utils gains coverage for payloadType extraction and null fallbacks.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Dhruv Sharma 2026-04-13 18:53:50 +05:30
parent 4ab844e47f
commit df9f3c8b5d
4 changed files with 145 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

25
packages/core/src/__tests__/utils.test.ts
View File

@ -90,6 +90,31 @@ describe("readLastJsonlEntry", () => {
const result = await readLastJsonlEntry(path);
expect(result!.modifiedAt).toBeInstanceOf(Date);
});
it("extracts payloadType from nested payload.type", async () => {
// Real Codex writes records like {"type":"event_msg","payload":{"type":"error",...}}
// Consumers need the inner payload.type to classify activity correctly.
const path = setup(
'{"type":"event_msg","payload":{"type":"error","message":"bad"}}\n',
);
const result = await readLastJsonlEntry(path);
expect(result!.lastType).toBe("event_msg");
expect(result!.payloadType).toBe("error");
});
it("returns payloadType null when payload has no type field", async () => {
const path = setup('{"type":"session_meta","payload":{"cwd":"/workspace"}}\n');
const result = await readLastJsonlEntry(path);
expect(result!.lastType).toBe("session_meta");
expect(result!.payloadType).toBeNull();
});
it("returns payloadType null when payload is not an object", async () => {
const path = setup('{"type":"x","payload":"string"}\n');
const result = await readLastJsonlEntry(path);
expect(result!.lastType).toBe("x");
expect(result!.payloadType).toBeNull();
});
});
describe("isGitBranchNameSafe", () => {

11
packages/core/src/utils.ts
View File

@ -138,7 +138,7 @@ async function readLastLine(filePath: string): Promise<string | null> {
*/
export async function readLastJsonlEntry(
filePath: string,
): Promise<{ lastType: string | null; modifiedAt: Date } | null> {
): Promise<{ lastType: string | null; payloadType: string | null; modifiedAt: Date } | null> {
try {
const [line, fileStat] = await Promise.all([readLastLine(filePath), stat(filePath)]);
@ -148,10 +148,15 @@ export async function readLastJsonlEntry(
if (typeof parsed === "object" && parsed !== null && !Array.isArray(parsed)) {
const obj = parsed as Record<string, unknown>;
const lastType = typeof obj.type === "string" ? obj.type : null;
return { lastType, modifiedAt: fileStat.mtime };
let payloadType: string | null = null;
if (typeof obj.payload === "object" && obj.payload !== null && !Array.isArray(obj.payload)) {
const payload = obj.payload as Record<string, unknown>;
if (typeof payload.type === "string") payloadType = payload.type;
}
return { lastType, payloadType, modifiedAt: fileStat.mtime };
}
return { lastType: null, modifiedAt: fileStat.mtime };
return { lastType: null, payloadType: null, modifiedAt: fileStat.mtime };
} catch {
return null;
}

88
packages/plugins/agent-codex/src/index.test.ts
View File

@ -739,6 +739,94 @@ describe("getActivityState", () => {
expect(result?.state).toBe("ready");
});
it("returns waiting_input when payload.type is approval_request on event_msg", async () => {
// Real Codex writes {"type":"event_msg","payload":{"type":"approval_request",...}}
// Without payloadType handling, this decays to ready/idle via the event_msg case.
mockTmuxWithProcess("codex");
const content = '{"type":"session_meta","cwd":"/workspace/test"}\n';
mockReaddir.mockResolvedValue(["sess.jsonl"]);
setupMockOpen(content);
mockStat.mockResolvedValue({ mtimeMs: Date.now(), mtime: new Date() });
mockReadLastJsonlEntry.mockResolvedValue({
lastType: "event_msg",
payloadType: "approval_request",
modifiedAt: new Date(),
});
const session = makeSession({ runtimeHandle: makeTmuxHandle(), workspacePath: "/workspace/test" });
const result = await agent.getActivityState(session);
expect(result?.state).toBe("waiting_input");
});
it("returns waiting_input for exec_approval_request payload type", async () => {
mockTmuxWithProcess("codex");
const content = '{"type":"session_meta","cwd":"/workspace/test"}\n';
mockReaddir.mockResolvedValue(["sess.jsonl"]);
setupMockOpen(content);
mockStat.mockResolvedValue({ mtimeMs: Date.now(), mtime: new Date() });
mockReadLastJsonlEntry.mockResolvedValue({
lastType: "event_msg",
payloadType: "exec_approval_request",
modifiedAt: new Date(),
});
const session = makeSession({ runtimeHandle: makeTmuxHandle(), workspacePath: "/workspace/test" });
const result = await agent.getActivityState(session);
expect(result?.state).toBe("waiting_input");
});
it("returns blocked when payload.type is error on event_msg", async () => {
// Real Codex writes {"type":"event_msg","payload":{"type":"error",...}}
mockTmuxWithProcess("codex");
const content = '{"type":"session_meta","cwd":"/workspace/test"}\n';
mockReaddir.mockResolvedValue(["sess.jsonl"]);
setupMockOpen(content);
mockStat.mockResolvedValue({ mtimeMs: Date.now(), mtime: new Date() });
mockReadLastJsonlEntry.mockResolvedValue({
lastType: "event_msg",
payloadType: "error",
modifiedAt: new Date(),
});
const session = makeSession({ runtimeHandle: makeTmuxHandle(), workspacePath: "/workspace/test" });
const result = await agent.getActivityState(session);
expect(result?.state).toBe("blocked");
});
it("returns active when payload.type is task_started on a recent event_msg", async () => {
mockTmuxWithProcess("codex");
const content = '{"type":"session_meta","cwd":"/workspace/test"}\n';
mockReaddir.mockResolvedValue(["sess.jsonl"]);
setupMockOpen(content);
mockStat.mockResolvedValue({ mtimeMs: Date.now(), mtime: new Date() });
mockReadLastJsonlEntry.mockResolvedValue({
lastType: "event_msg",
payloadType: "task_started",
modifiedAt: new Date(),
});
const session = makeSession({ runtimeHandle: makeTmuxHandle(), workspacePath: "/workspace/test" });
const result = await agent.getActivityState(session);
expect(result?.state).toBe("active");
});
it("returns ready when payload.type is task_complete on a recent event_msg", async () => {
mockTmuxWithProcess("codex");
const content = '{"type":"session_meta","cwd":"/workspace/test"}\n';
mockReaddir.mockResolvedValue(["sess.jsonl"]);
setupMockOpen(content);
mockStat.mockResolvedValue({ mtimeMs: Date.now(), mtime: new Date() });
mockReadLastJsonlEntry.mockResolvedValue({
lastType: "event_msg",
payloadType: "task_complete",
modifiedAt: new Date(),
});
const session = makeSession({ runtimeHandle: makeTmuxHandle(), workspacePath: "/workspace/test" });
const result = await agent.getActivityState(session);
expect(result?.state).toBe("ready");
});
it("detects activity from payload-wrapped Codex session_meta files", async () => {
mockTmuxWithProcess("codex");
const content =

31
packages/plugins/agent-codex/src/index.ts
View File

@ -508,30 +508,47 @@ function createCodexAgent(): Agent {
const ageMs = Date.now() - entry.modifiedAt.getTime();
const timestamp = entry.modifiedAt;
// Real Codex wraps the semantic type in `payload.type` on event_msg
// records (e.g. `{"type":"event_msg","payload":{"type":"error",...}}`).
// Prefer payloadType when present so approval_request/error surface
// correctly instead of decaying to ready/idle via the event_msg case.
const effectiveType = entry.payloadType ?? entry.lastType;
// Map Codex JSONL entry types to activity states.
// Confirmed types: session_meta, event_msg. Others are best-effort.
const activeWindowMs = Math.min(DEFAULT_ACTIVE_WINDOW_MS, threshold);
switch (entry.lastType) {
switch (effectiveType) {
case "approval_request":
case "exec_approval_request":
case "apply_patch_approval_request":
return { state: "waiting_input", timestamp };
case "error":
case "stream_error":
return { state: "blocked", timestamp };
case "task_started":
case "agent_reasoning":
case "response_item":
case "turn_context":
case "user_input":
case "tool_call":
case "exec_command":
case "exec_command_begin":
case "exec_command_end":
if (ageMs <= activeWindowMs) return { state: "active", timestamp };
return { state: ageMs > threshold ? "idle" : "ready", timestamp };
case "task_complete":
case "turn_aborted":
case "agent_message":
case "assistant_message":
case "session_meta":
case "event_msg":
case "compacted":
case "token_count":
return { state: ageMs > threshold ? "idle" : "ready", timestamp };
case "approval_request":
return { state: "waiting_input", timestamp };
case "error":
return { state: "blocked", timestamp };
default:
if (ageMs <= activeWindowMs) return { state: "active", timestamp };
return { state: ageMs > threshold ? "idle" : "ready", timestamp };