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

Fix orchestrator activity event instrumentation

This commit is contained in:
whoisasx 2026-05-18 15:18:07 +05:30
parent d05560565d
commit 24ce5d77a4
2 changed files with 163 additions and 14 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

113
packages/core/src/__tests__/session-manager-instrumentation.test.ts
View File

@ -166,6 +166,80 @@ describe("session.kill_started (MUST)", () => {
});
describe("session.spawn_failed — orchestrator path (MUST)", () => {
it("emits session.spawned after a successful orchestrator spawn", async () => {
const sm = createSessionManager({ config, registry: mockRegistry });
const session = await sm.spawnOrchestrator({
projectId: "my-app",
systemPrompt: "be helpful",
});
const events = findAllEvents("session.spawned");
const orchestratorSpawned = events.find(
(e) => e.data && (e.data as Record<string, unknown>)["role"] === "orchestrator",
);
expect(session.id).toBe("app-orchestrator");
expect(orchestratorSpawned).toMatchObject({
projectId: "my-app",
sessionId: "app-orchestrator",
source: "session-manager",
kind: "session.spawned",
summary: "spawned: app-orchestrator",
data: {
agent: "mock-agent",
branch: "orchestrator/app-orchestrator",
role: "orchestrator",
},
});
});
it("does not emit terminal spawn_failed when ensure recovers a fixed reservation conflict", async () => {
let releaseWorkspace: () => void = () => {};
const blockingWorkspace = new Promise<void>((resolve) => {
releaseWorkspace = resolve;
});
vi.mocked(ctx.mockWorkspace.create).mockImplementationOnce(async (cfg) => {
await blockingWorkspace;
return {
path: join(ctx.tmpDir, "ws-orchestrator"),
branch: cfg.branch,
sessionId: cfg.sessionId,
projectId: cfg.projectId,
};
});
const firstManager = createSessionManager({ config, registry: mockRegistry });
const secondManager = createSessionManager({ config, registry: mockRegistry });
const firstEnsure = firstManager.ensureOrchestrator({
projectId: "my-app",
systemPrompt: "be helpful",
});
await vi.waitFor(() => {
expect(ctx.mockWorkspace.create).toHaveBeenCalledTimes(1);
});
const secondEnsure = secondManager.ensureOrchestrator({
projectId: "my-app",
systemPrompt: "be helpful",
});
await vi.waitFor(() => {
expect(findEvent("session.orchestrator_conflict")).toBeDefined();
});
releaseWorkspace();
const [created, recovered] = await Promise.all([firstEnsure, secondEnsure]);
expect(created.id).toBe("app-orchestrator");
expect(recovered.id).toBe("app-orchestrator");
expect(findAllEvents("session.orchestrator_conflict")).toHaveLength(1);
expect(
findAllEvents("session.spawn_failed").filter(
(e) => e.data && (e.data as Record<string, unknown>)["role"] === "orchestrator",
),
).toHaveLength(0);
});
it("emits one terminal failure plus one stage failure when workspace.create throws", async () => {
vi.mocked(ctx.mockWorkspace.create).mockRejectedValue(new Error("disk full"));
@ -359,6 +433,45 @@ describe("session.send_failed (MUST)", () => {
expect(restoreFailed[0]!.data).toMatchObject({ stage: "workspace_restore" });
expect(findEvent("session.send_failed")).toBeDefined();
});
it("tags restore-for-delivery timeout restore_failed with ready_timeout stage", async () => {
vi.useFakeTimers();
try {
const wsPath = join(ctx.tmpDir, "send-restore-ready-timeout");
mkdirSync(wsPath, { recursive: true });
writeTerminatedSession("app-send-timeout", { worktree: wsPath, branch: "feat/send" });
vi.mocked(ctx.mockRuntime.isAlive).mockImplementation(async (handle) => {
return handle.id !== "rt-restored";
});
vi.mocked(ctx.mockAgent.isProcessRunning).mockImplementation(async (handle) => {
return handle.id !== "rt-restored";
});
vi.mocked(ctx.mockRuntime.create).mockResolvedValue(makeHandle("rt-restored"));
vi.mocked(ctx.mockRuntime.getOutput).mockResolvedValue("");
vi.mocked(ctx.mockAgent.detectActivity).mockReturnValue("idle");
const sm = createSessionManager({ config, registry: mockRegistry });
const sendPromise = sm.send("app-send-timeout", "hi");
const rejection = expect(sendPromise).rejects.toThrow(
"restored session did not become ready for delivery",
);
await vi.runAllTimersAsync();
await rejection;
const restoreFailed = findAllEvents("session.restore_failed");
expect(restoreFailed).toHaveLength(1);
expect(restoreFailed[0]!.data).toMatchObject({
stage: "ready_timeout",
reason: "restored session did not become ready for delivery",
trigger: "send",
});
expect(findEvent("session.send_failed")).toBeDefined();
} finally {
vi.useRealTimers();
}
});
});
describe("session.restore_failed (MUST)", () => {

64
packages/core/src/session-manager.ts
View File

@ -1571,7 +1571,29 @@ export function createSessionManager(deps: SessionManagerDeps): OpenCodeSessionM
}
}
async function spawnOrchestrator(orchestratorConfig: OrchestratorSpawnConfig): Promise<Session> {
function recordOrchestratorSpawnFailed(
orchestratorConfig: OrchestratorSpawnConfig,
err: unknown,
sessionId?: string,
): void {
recordActivityEvent({
projectId: orchestratorConfig.projectId,
...(sessionId ? { sessionId } : {}),
source: "session-manager",
kind: "session.spawn_failed",
level: "error",
summary: "orchestrator spawn failed",
data: {
role: "orchestrator",
reason: err instanceof Error ? err.message : String(err),
},
});
}
async function spawnOrchestrator(
orchestratorConfig: OrchestratorSpawnConfig,
options?: { suppressFixedReservationFailure?: boolean },
): Promise<Session> {
recordActivityEvent({
projectId: orchestratorConfig.projectId,
source: "session-manager",
@ -1582,17 +1604,15 @@ export function createSessionManager(deps: SessionManagerDeps): OpenCodeSessionM
try {
return await _spawnOrchestratorInner(orchestratorConfig);
} catch (err) {
recordActivityEvent({
projectId: orchestratorConfig.projectId,
source: "session-manager",
kind: "session.spawn_failed",
level: "error",
summary: "orchestrator spawn failed",
data: {
role: "orchestrator",
reason: err instanceof Error ? err.message : String(err),
},
});
const project = config.projects[orchestratorConfig.projectId];
const sessionId = project ? getOrchestratorSessionId(project) : undefined;
const shouldSuppressRecoverableConflict =
options?.suppressFixedReservationFailure === true &&
sessionId !== undefined &&
isFixedOrchestratorReservationError(err, sessionId);
if (!shouldSuppressRecoverableConflict) {
recordOrchestratorSpawnFailed(orchestratorConfig, err, sessionId);
}
throw err;
}
}
@ -2001,6 +2021,19 @@ export function createSessionManager(deps: SessionManagerDeps): OpenCodeSessionM
throw err;
}
recordActivityEvent({
projectId: orchestratorConfig.projectId,
sessionId,
source: "session-manager",
kind: "session.spawned",
summary: `spawned: ${sessionId}`,
data: {
agent: plugins.agent.name,
branch: session.branch ?? undefined,
role: "orchestrator",
},
});
return session;
}
@ -2069,7 +2102,9 @@ export function createSessionManager(deps: SessionManagerDeps): OpenCodeSessionM
}
try {
return await spawnOrchestrator(orchestratorConfig);
return await spawnOrchestrator(orchestratorConfig, {
suppressFixedReservationFailure: true,
});
} catch (err) {
if (!isFixedOrchestratorReservationError(err, sessionId)) {
throw err;
@ -2087,6 +2122,7 @@ export function createSessionManager(deps: SessionManagerDeps): OpenCodeSessionM
const concurrent = await waitForConcurrentOrchestrator(sessionId);
if (concurrent) return concurrent;
recordOrchestratorSpawnFailed(orchestratorConfig, err, sessionId);
throw err;
}
}
@ -2900,7 +2936,7 @@ export function createSessionManager(deps: SessionManagerDeps): OpenCodeSessionM
kind: "session.restore_failed",
level: "error",
summary: `restore for delivery failed: ${sessionId}`,
data: { reason: detail, trigger: "send" },
data: { stage: "ready_timeout", reason: detail, trigger: "send" },
});
throw new Error(`Cannot send to session ${sessionId}: ${reason} (${detail})`);
}