refactor: remove EventBus, simplify lifecycle manager
The event bus was over-engineered for a system with ~50 sessions. The dashboard can read metadata files + poll GitHub directly. Remove the EventBus interface, event-bus.ts module, and all emit/subscribe wiring from session-manager and lifecycle-manager. The reaction engine (the valuable part) stays — executeReaction now takes sessionId/projectId directly instead of going through OrchestratorEvent. Also fix listMetadata() to filter out directories (not just the "archive" dir by name) so readMetadataRaw() can't crash on non-file entries. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Prateek
parent
4c47a42b6e
commit
95fd47ef2a
|
|
@ -1,324 +0,0 @@
|
|||
import { describe, it, expect, beforeEach, afterEach } from "vitest";
|
||||
import { mkdirSync, rmSync, readFileSync, existsSync } from "node:fs";
|
||||
import { join } from "node:path";
|
||||
import { tmpdir } from "node:os";
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { createEventBus, createEvent } from "../event-bus.js";
|
||||
import type { OrchestratorEvent } from "../types.js";
|
||||
|
||||
let tmpDir: string;
|
||||
|
||||
beforeEach(() => {
|
||||
tmpDir = join(tmpdir(), `ao-test-eventbus-${randomUUID()}`);
|
||||
mkdirSync(tmpDir, { recursive: true });
|
||||
});
|
||||
|
||||
afterEach(() => {
|
||||
rmSync(tmpDir, { recursive: true, force: true });
|
||||
});
|
||||
|
||||
describe("createEvent", () => {
|
||||
it("creates event with all fields", () => {
|
||||
const event = createEvent("session.spawned", {
|
||||
sessionId: "app-1",
|
||||
projectId: "my-app",
|
||||
message: "Session spawned",
|
||||
data: { branch: "main" },
|
||||
});
|
||||
|
||||
expect(event.id).toBeTruthy();
|
||||
expect(event.type).toBe("session.spawned");
|
||||
expect(event.sessionId).toBe("app-1");
|
||||
expect(event.projectId).toBe("my-app");
|
||||
expect(event.message).toBe("Session spawned");
|
||||
expect(event.data).toEqual({ branch: "main" });
|
||||
expect(event.timestamp).toBeInstanceOf(Date);
|
||||
});
|
||||
|
||||
it("infers priority from event type", () => {
|
||||
expect(
|
||||
createEvent("session.stuck", { sessionId: "a", projectId: "p", message: "" }).priority,
|
||||
).toBe("urgent");
|
||||
expect(
|
||||
createEvent("session.needs_input", { sessionId: "a", projectId: "p", message: "" }).priority,
|
||||
).toBe("urgent");
|
||||
expect(
|
||||
createEvent("session.errored", { sessionId: "a", projectId: "p", message: "" }).priority,
|
||||
).toBe("urgent");
|
||||
expect(
|
||||
createEvent("review.approved", { sessionId: "a", projectId: "p", message: "" }).priority,
|
||||
).toBe("action");
|
||||
expect(
|
||||
createEvent("merge.ready", { sessionId: "a", projectId: "p", message: "" }).priority,
|
||||
).toBe("action");
|
||||
expect(
|
||||
createEvent("merge.completed", { sessionId: "a", projectId: "p", message: "" }).priority,
|
||||
).toBe("action");
|
||||
expect(
|
||||
createEvent("ci.failing", { sessionId: "a", projectId: "p", message: "" }).priority,
|
||||
).toBe("warning");
|
||||
expect(
|
||||
createEvent("review.changes_requested", { sessionId: "a", projectId: "p", message: "" })
|
||||
.priority,
|
||||
).toBe("warning");
|
||||
expect(
|
||||
createEvent("session.spawned", { sessionId: "a", projectId: "p", message: "" }).priority,
|
||||
).toBe("info");
|
||||
expect(
|
||||
createEvent("pr.created", { sessionId: "a", projectId: "p", message: "" }).priority,
|
||||
).toBe("info");
|
||||
});
|
||||
|
||||
it("allows explicit priority override", () => {
|
||||
const event = createEvent("session.spawned", {
|
||||
sessionId: "a",
|
||||
projectId: "p",
|
||||
message: "",
|
||||
priority: "urgent",
|
||||
});
|
||||
expect(event.priority).toBe("urgent");
|
||||
});
|
||||
|
||||
it("generates unique IDs", () => {
|
||||
const e1 = createEvent("session.spawned", { sessionId: "a", projectId: "p", message: "" });
|
||||
const e2 = createEvent("session.spawned", { sessionId: "a", projectId: "p", message: "" });
|
||||
expect(e1.id).not.toBe(e2.id);
|
||||
});
|
||||
});
|
||||
|
||||
describe("createEventBus (no persistence)", () => {
|
||||
it("emits events to typed handlers", () => {
|
||||
const bus = createEventBus(null);
|
||||
const received: OrchestratorEvent[] = [];
|
||||
|
||||
bus.on("session.spawned", (e) => received.push(e));
|
||||
|
||||
const event = createEvent("session.spawned", {
|
||||
sessionId: "app-1",
|
||||
projectId: "p",
|
||||
message: "spawned",
|
||||
});
|
||||
bus.emit(event);
|
||||
|
||||
expect(received).toHaveLength(1);
|
||||
expect(received[0].sessionId).toBe("app-1");
|
||||
});
|
||||
|
||||
it("emits events to wildcard handlers", () => {
|
||||
const bus = createEventBus(null);
|
||||
const received: OrchestratorEvent[] = [];
|
||||
|
||||
bus.on("*", (e) => received.push(e));
|
||||
|
||||
bus.emit(createEvent("session.spawned", { sessionId: "a", projectId: "p", message: "" }));
|
||||
bus.emit(createEvent("ci.failing", { sessionId: "b", projectId: "p", message: "" }));
|
||||
|
||||
expect(received).toHaveLength(2);
|
||||
});
|
||||
|
||||
it("does not deliver events to wrong type", () => {
|
||||
const bus = createEventBus(null);
|
||||
const received: OrchestratorEvent[] = [];
|
||||
|
||||
bus.on("ci.failing", (e) => received.push(e));
|
||||
|
||||
bus.emit(createEvent("session.spawned", { sessionId: "a", projectId: "p", message: "" }));
|
||||
|
||||
expect(received).toHaveLength(0);
|
||||
});
|
||||
|
||||
it("unsubscribes handlers with off()", () => {
|
||||
const bus = createEventBus(null);
|
||||
const received: OrchestratorEvent[] = [];
|
||||
const handler = (e: OrchestratorEvent) => received.push(e);
|
||||
|
||||
bus.on("session.spawned", handler);
|
||||
bus.emit(createEvent("session.spawned", { sessionId: "a", projectId: "p", message: "" }));
|
||||
expect(received).toHaveLength(1);
|
||||
|
||||
bus.off("session.spawned", handler);
|
||||
bus.emit(createEvent("session.spawned", { sessionId: "b", projectId: "p", message: "" }));
|
||||
expect(received).toHaveLength(1); // still 1, not 2
|
||||
});
|
||||
|
||||
it("survives handler errors without breaking", () => {
|
||||
const bus = createEventBus(null);
|
||||
const received: OrchestratorEvent[] = [];
|
||||
|
||||
bus.on("session.spawned", () => {
|
||||
throw new Error("boom");
|
||||
});
|
||||
bus.on("session.spawned", (e) => received.push(e));
|
||||
|
||||
bus.emit(createEvent("session.spawned", { sessionId: "a", projectId: "p", message: "" }));
|
||||
|
||||
expect(received).toHaveLength(1);
|
||||
});
|
||||
|
||||
it("supports multiple handlers on same event", () => {
|
||||
const bus = createEventBus(null);
|
||||
let count = 0;
|
||||
|
||||
bus.on("session.spawned", () => count++);
|
||||
bus.on("session.spawned", () => count++);
|
||||
|
||||
bus.emit(createEvent("session.spawned", { sessionId: "a", projectId: "p", message: "" }));
|
||||
expect(count).toBe(2);
|
||||
});
|
||||
});
|
||||
|
||||
describe("getHistory", () => {
|
||||
it("returns all events when no filter", () => {
|
||||
const bus = createEventBus(null);
|
||||
|
||||
bus.emit(createEvent("session.spawned", { sessionId: "a", projectId: "p1", message: "" }));
|
||||
bus.emit(createEvent("ci.failing", { sessionId: "b", projectId: "p2", message: "" }));
|
||||
|
||||
const history = bus.getHistory();
|
||||
expect(history).toHaveLength(2);
|
||||
});
|
||||
|
||||
it("filters by sessionId", () => {
|
||||
const bus = createEventBus(null);
|
||||
|
||||
bus.emit(createEvent("session.spawned", { sessionId: "a", projectId: "p", message: "" }));
|
||||
bus.emit(createEvent("session.spawned", { sessionId: "b", projectId: "p", message: "" }));
|
||||
|
||||
expect(bus.getHistory({ sessionId: "a" })).toHaveLength(1);
|
||||
});
|
||||
|
||||
it("filters by projectId", () => {
|
||||
const bus = createEventBus(null);
|
||||
|
||||
bus.emit(createEvent("session.spawned", { sessionId: "a", projectId: "p1", message: "" }));
|
||||
bus.emit(createEvent("session.spawned", { sessionId: "b", projectId: "p2", message: "" }));
|
||||
|
||||
expect(bus.getHistory({ projectId: "p1" })).toHaveLength(1);
|
||||
});
|
||||
|
||||
it("filters by type", () => {
|
||||
const bus = createEventBus(null);
|
||||
|
||||
bus.emit(createEvent("session.spawned", { sessionId: "a", projectId: "p", message: "" }));
|
||||
bus.emit(createEvent("ci.failing", { sessionId: "b", projectId: "p", message: "" }));
|
||||
|
||||
expect(bus.getHistory({ type: "ci.failing" })).toHaveLength(1);
|
||||
});
|
||||
|
||||
it("filters by priority", () => {
|
||||
const bus = createEventBus(null);
|
||||
|
||||
bus.emit(createEvent("session.spawned", { sessionId: "a", projectId: "p", message: "" }));
|
||||
bus.emit(createEvent("session.stuck", { sessionId: "b", projectId: "p", message: "" }));
|
||||
|
||||
expect(bus.getHistory({ priority: "urgent" })).toHaveLength(1);
|
||||
});
|
||||
|
||||
it("filters by since", () => {
|
||||
const bus = createEventBus(null);
|
||||
const beforeTime = new Date(Date.now() - 1000);
|
||||
|
||||
bus.emit(createEvent("session.spawned", { sessionId: "a", projectId: "p", message: "" }));
|
||||
|
||||
const afterTime = new Date(Date.now() + 1000);
|
||||
expect(bus.getHistory({ since: beforeTime })).toHaveLength(1);
|
||||
expect(bus.getHistory({ since: afterTime })).toHaveLength(0);
|
||||
});
|
||||
|
||||
it("applies limit (takes last N)", () => {
|
||||
const bus = createEventBus(null);
|
||||
|
||||
for (let i = 0; i < 10; i++) {
|
||||
bus.emit(
|
||||
createEvent("session.spawned", {
|
||||
sessionId: `s-${i}`,
|
||||
projectId: "p",
|
||||
message: `msg-${i}`,
|
||||
}),
|
||||
);
|
||||
}
|
||||
|
||||
const result = bus.getHistory({ limit: 3 });
|
||||
expect(result).toHaveLength(3);
|
||||
expect(result[0].sessionId).toBe("s-7");
|
||||
expect(result[2].sessionId).toBe("s-9");
|
||||
});
|
||||
|
||||
it("combines multiple filters", () => {
|
||||
const bus = createEventBus(null);
|
||||
|
||||
bus.emit(createEvent("session.spawned", { sessionId: "a", projectId: "p1", message: "" }));
|
||||
bus.emit(createEvent("session.spawned", { sessionId: "b", projectId: "p1", message: "" }));
|
||||
bus.emit(createEvent("ci.failing", { sessionId: "a", projectId: "p1", message: "" }));
|
||||
bus.emit(createEvent("session.spawned", { sessionId: "a", projectId: "p2", message: "" }));
|
||||
|
||||
expect(bus.getHistory({ sessionId: "a", projectId: "p1" })).toHaveLength(2);
|
||||
expect(bus.getHistory({ sessionId: "a", type: "session.spawned" })).toHaveLength(2);
|
||||
expect(
|
||||
bus.getHistory({ sessionId: "a", projectId: "p1", type: "session.spawned" }),
|
||||
).toHaveLength(1);
|
||||
});
|
||||
});
|
||||
|
||||
describe("JSONL persistence", () => {
|
||||
it("persists events to JSONL file", () => {
|
||||
const logPath = join(tmpDir, "events.jsonl");
|
||||
const bus = createEventBus(logPath);
|
||||
|
||||
bus.emit(createEvent("session.spawned", { sessionId: "a", projectId: "p", message: "hello" }));
|
||||
bus.emit(createEvent("ci.failing", { sessionId: "b", projectId: "p", message: "oops" }));
|
||||
|
||||
const content = readFileSync(logPath, "utf-8");
|
||||
const lines = content.trim().split("\n");
|
||||
expect(lines).toHaveLength(2);
|
||||
|
||||
const parsed = JSON.parse(lines[0]);
|
||||
expect(parsed.type).toBe("session.spawned");
|
||||
expect(parsed.sessionId).toBe("a");
|
||||
expect(parsed.timestamp).toBeTruthy();
|
||||
});
|
||||
|
||||
it("loads history from existing JSONL file on creation", async () => {
|
||||
const logPath = join(tmpDir, "existing.jsonl");
|
||||
const event = createEvent("session.spawned", {
|
||||
sessionId: "old",
|
||||
projectId: "p",
|
||||
message: "old event",
|
||||
});
|
||||
const serialized = JSON.stringify({ ...event, timestamp: event.timestamp.toISOString() });
|
||||
const { writeFileSync } = await import("node:fs");
|
||||
writeFileSync(logPath, serialized + "\n", "utf-8");
|
||||
|
||||
const bus = createEventBus(logPath);
|
||||
const history = bus.getHistory();
|
||||
expect(history).toHaveLength(1);
|
||||
expect(history[0].sessionId).toBe("old");
|
||||
expect(history[0].timestamp).toBeInstanceOf(Date);
|
||||
});
|
||||
|
||||
it("creates log directory if it does not exist", () => {
|
||||
const logPath = join(tmpDir, "subdir", "nested", "events.jsonl");
|
||||
const bus = createEventBus(logPath);
|
||||
|
||||
bus.emit(createEvent("session.spawned", { sessionId: "a", projectId: "p", message: "" }));
|
||||
|
||||
expect(existsSync(logPath)).toBe(true);
|
||||
});
|
||||
|
||||
it("appends to existing log file", () => {
|
||||
const logPath = join(tmpDir, "append.jsonl");
|
||||
|
||||
const bus1 = createEventBus(logPath);
|
||||
bus1.emit(createEvent("session.spawned", { sessionId: "a", projectId: "p", message: "" }));
|
||||
|
||||
const bus2 = createEventBus(logPath);
|
||||
bus2.emit(createEvent("ci.failing", { sessionId: "b", projectId: "p", message: "" }));
|
||||
|
||||
const content = readFileSync(logPath, "utf-8");
|
||||
const lines = content.trim().split("\n");
|
||||
expect(lines).toHaveLength(2);
|
||||
|
||||
// bus2 should have loaded the first event + emitted the second
|
||||
expect(bus2.getHistory()).toHaveLength(2);
|
||||
});
|
||||
});
|
||||
|
|
@ -4,7 +4,6 @@ import { join } from "node:path";
|
|||
import { tmpdir } from "node:os";
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { createLifecycleManager } from "../lifecycle-manager.js";
|
||||
import { createEventBus } from "../event-bus.js";
|
||||
import { writeMetadata, readMetadataRaw } from "../metadata.js";
|
||||
import type {
|
||||
OrchestratorConfig,
|
||||
|
|
@ -14,14 +13,11 @@ import type {
|
|||
Runtime,
|
||||
Agent,
|
||||
SCM,
|
||||
EventBus,
|
||||
OrchestratorEvent,
|
||||
ActivityState,
|
||||
PRInfo,
|
||||
} from "../types.js";
|
||||
|
||||
let dataDir: string;
|
||||
let eventBus: EventBus;
|
||||
let mockSessionManager: SessionManager;
|
||||
let mockRuntime: Runtime;
|
||||
let mockAgent: Agent;
|
||||
|
|
@ -65,8 +61,6 @@ beforeEach(() => {
|
|||
dataDir = join(tmpdir(), `ao-test-lifecycle-${randomUUID()}`);
|
||||
mkdirSync(join(dataDir, "sessions"), { recursive: true });
|
||||
|
||||
eventBus = createEventBus(null);
|
||||
|
||||
mockRuntime = {
|
||||
name: "mock",
|
||||
create: vi.fn(),
|
||||
|
|
@ -148,7 +142,6 @@ describe("start / stop", () => {
|
|||
config,
|
||||
registry: mockRegistry,
|
||||
sessionManager: mockSessionManager,
|
||||
eventBus,
|
||||
});
|
||||
|
||||
lm.start(60_000);
|
||||
|
|
@ -173,21 +166,15 @@ describe("check (single session)", () => {
|
|||
project: "my-app",
|
||||
});
|
||||
|
||||
const received: OrchestratorEvent[] = [];
|
||||
eventBus.on("session.working", (e) => received.push(e));
|
||||
|
||||
const lm = createLifecycleManager({
|
||||
config,
|
||||
registry: mockRegistry,
|
||||
sessionManager: mockSessionManager,
|
||||
eventBus,
|
||||
});
|
||||
|
||||
await lm.check("app-1");
|
||||
|
||||
expect(lm.getStates().get("app-1")).toBe("working");
|
||||
expect(received).toHaveLength(1);
|
||||
expect(received[0].type).toBe("session.working");
|
||||
|
||||
// Metadata should be updated
|
||||
const meta = readMetadataRaw(dataDir, "app-1");
|
||||
|
|
@ -211,7 +198,6 @@ describe("check (single session)", () => {
|
|||
config,
|
||||
registry: mockRegistry,
|
||||
sessionManager: mockSessionManager,
|
||||
eventBus,
|
||||
});
|
||||
|
||||
await lm.check("app-1");
|
||||
|
|
@ -236,7 +222,6 @@ describe("check (single session)", () => {
|
|||
config,
|
||||
registry: mockRegistry,
|
||||
sessionManager: mockSessionManager,
|
||||
eventBus,
|
||||
});
|
||||
|
||||
await lm.check("app-1");
|
||||
|
|
@ -261,7 +246,6 @@ describe("check (single session)", () => {
|
|||
config,
|
||||
registry: mockRegistry,
|
||||
sessionManager: mockSessionManager,
|
||||
eventBus,
|
||||
});
|
||||
|
||||
await lm.check("app-1");
|
||||
|
|
@ -309,7 +293,6 @@ describe("check (single session)", () => {
|
|||
config,
|
||||
registry: registryWithSCM,
|
||||
sessionManager: mockSessionManager,
|
||||
eventBus,
|
||||
});
|
||||
|
||||
await lm.check("app-1");
|
||||
|
|
@ -357,7 +340,6 @@ describe("check (single session)", () => {
|
|||
config,
|
||||
registry: registryWithSCM,
|
||||
sessionManager: mockSessionManager,
|
||||
eventBus,
|
||||
});
|
||||
|
||||
await lm.check("app-1");
|
||||
|
|
@ -411,7 +393,6 @@ describe("check (single session)", () => {
|
|||
config,
|
||||
registry: registryWithSCM,
|
||||
sessionManager: mockSessionManager,
|
||||
eventBus,
|
||||
});
|
||||
|
||||
await lm.check("app-1");
|
||||
|
|
@ -426,13 +407,12 @@ describe("check (single session)", () => {
|
|||
config,
|
||||
registry: mockRegistry,
|
||||
sessionManager: mockSessionManager,
|
||||
eventBus,
|
||||
});
|
||||
|
||||
await expect(lm.check("nonexistent")).rejects.toThrow("not found");
|
||||
});
|
||||
|
||||
it("does not emit event when status unchanged", async () => {
|
||||
it("does not change state when status is unchanged", async () => {
|
||||
const session = makeSession({ status: "working" });
|
||||
vi.mocked(mockSessionManager.get).mockResolvedValue(session);
|
||||
|
||||
|
|
@ -443,28 +423,18 @@ describe("check (single session)", () => {
|
|||
project: "my-app",
|
||||
});
|
||||
|
||||
const received: OrchestratorEvent[] = [];
|
||||
eventBus.on("*", (e) => received.push(e));
|
||||
|
||||
const lm = createLifecycleManager({
|
||||
config,
|
||||
registry: mockRegistry,
|
||||
sessionManager: mockSessionManager,
|
||||
eventBus,
|
||||
});
|
||||
|
||||
// First check: spawning (from states map, empty) → working = transition
|
||||
// But the session's status is "working" and determineStatus returns "working"
|
||||
// Since states map is empty, oldStatus = session.status = "working", newStatus = "working"
|
||||
// So no transition
|
||||
// We need to seed the states map first
|
||||
// Actually, let's check twice: first time sets it, second time same = no event
|
||||
await lm.check("app-1");
|
||||
const eventsAfterFirst = [...received];
|
||||
expect(lm.getStates().get("app-1")).toBe("working");
|
||||
|
||||
// Second check — status remains working, no transition
|
||||
await lm.check("app-1");
|
||||
// No new events emitted since status didn't change
|
||||
expect(received.length).toBe(eventsAfterFirst.length);
|
||||
expect(lm.getStates().get("app-1")).toBe("working");
|
||||
});
|
||||
});
|
||||
|
||||
|
|
@ -519,7 +489,6 @@ describe("reactions", () => {
|
|||
config,
|
||||
registry: registryWithSCM,
|
||||
sessionManager: mockSessionManager,
|
||||
eventBus,
|
||||
});
|
||||
|
||||
await lm.check("app-1");
|
||||
|
|
@ -575,7 +544,6 @@ describe("reactions", () => {
|
|||
config,
|
||||
registry: registryWithSCM,
|
||||
sessionManager: mockSessionManager,
|
||||
eventBus,
|
||||
});
|
||||
|
||||
await lm.check("app-1");
|
||||
|
|
@ -584,92 +552,6 @@ describe("reactions", () => {
|
|||
});
|
||||
});
|
||||
|
||||
describe("on / off event handlers", () => {
|
||||
it("subscribes and receives lifecycle events", async () => {
|
||||
const session = makeSession({ status: "spawning" });
|
||||
vi.mocked(mockSessionManager.get).mockResolvedValue(session);
|
||||
|
||||
writeMetadata(dataDir, "app-1", {
|
||||
worktree: "/tmp",
|
||||
branch: "main",
|
||||
status: "spawning",
|
||||
project: "my-app",
|
||||
});
|
||||
|
||||
const received: OrchestratorEvent[] = [];
|
||||
|
||||
const lm = createLifecycleManager({
|
||||
config,
|
||||
registry: mockRegistry,
|
||||
sessionManager: mockSessionManager,
|
||||
eventBus,
|
||||
});
|
||||
|
||||
lm.on("session.working", (e) => received.push(e));
|
||||
|
||||
await lm.check("app-1");
|
||||
|
||||
expect(received).toHaveLength(1);
|
||||
expect(received[0].type).toBe("session.working");
|
||||
});
|
||||
|
||||
it("unsubscribes handlers", async () => {
|
||||
const session = makeSession({ status: "spawning" });
|
||||
vi.mocked(mockSessionManager.get).mockResolvedValue(session);
|
||||
|
||||
writeMetadata(dataDir, "app-1", {
|
||||
worktree: "/tmp",
|
||||
branch: "main",
|
||||
status: "spawning",
|
||||
project: "my-app",
|
||||
});
|
||||
|
||||
const received: OrchestratorEvent[] = [];
|
||||
const handler = (e: OrchestratorEvent) => received.push(e);
|
||||
|
||||
const lm = createLifecycleManager({
|
||||
config,
|
||||
registry: mockRegistry,
|
||||
sessionManager: mockSessionManager,
|
||||
eventBus,
|
||||
});
|
||||
|
||||
lm.on("session.working", handler);
|
||||
lm.off("session.working", handler);
|
||||
|
||||
await lm.check("app-1");
|
||||
|
||||
expect(received).toHaveLength(0);
|
||||
});
|
||||
|
||||
it("wildcard handler receives all events", async () => {
|
||||
const session = makeSession({ status: "spawning" });
|
||||
vi.mocked(mockSessionManager.get).mockResolvedValue(session);
|
||||
|
||||
writeMetadata(dataDir, "app-1", {
|
||||
worktree: "/tmp",
|
||||
branch: "main",
|
||||
status: "spawning",
|
||||
project: "my-app",
|
||||
});
|
||||
|
||||
const received: OrchestratorEvent[] = [];
|
||||
|
||||
const lm = createLifecycleManager({
|
||||
config,
|
||||
registry: mockRegistry,
|
||||
sessionManager: mockSessionManager,
|
||||
eventBus,
|
||||
});
|
||||
|
||||
lm.on("*", (e) => received.push(e));
|
||||
|
||||
await lm.check("app-1");
|
||||
|
||||
expect(received.length).toBeGreaterThan(0);
|
||||
});
|
||||
});
|
||||
|
||||
describe("getStates", () => {
|
||||
it("returns copy of states map", async () => {
|
||||
const session = makeSession({ status: "spawning" });
|
||||
|
|
@ -686,7 +568,6 @@ describe("getStates", () => {
|
|||
config,
|
||||
registry: mockRegistry,
|
||||
sessionManager: mockSessionManager,
|
||||
eventBus,
|
||||
});
|
||||
|
||||
await lm.check("app-1");
|
||||
|
|
|
|||
|
|
@ -4,7 +4,6 @@ import { join } from "node:path";
|
|||
import { tmpdir } from "node:os";
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { createSessionManager } from "../session-manager.js";
|
||||
import { createEventBus } from "../event-bus.js";
|
||||
import { writeMetadata, readMetadata } from "../metadata.js";
|
||||
import type {
|
||||
OrchestratorConfig,
|
||||
|
|
@ -15,12 +14,9 @@ import type {
|
|||
Tracker,
|
||||
SCM,
|
||||
RuntimeHandle,
|
||||
EventBus,
|
||||
OrchestratorEvent,
|
||||
} from "../types.js";
|
||||
|
||||
let dataDir: string;
|
||||
let eventBus: EventBus;
|
||||
let mockRuntime: Runtime;
|
||||
let mockAgent: Agent;
|
||||
let mockWorkspace: Workspace;
|
||||
|
|
@ -35,8 +31,6 @@ beforeEach(() => {
|
|||
dataDir = join(tmpdir(), `ao-test-session-mgr-${randomUUID()}`);
|
||||
mkdirSync(join(dataDir, "sessions"), { recursive: true });
|
||||
|
||||
eventBus = createEventBus(null);
|
||||
|
||||
mockRuntime = {
|
||||
name: "mock",
|
||||
create: vi.fn().mockResolvedValue(makeHandle("rt-1")),
|
||||
|
|
@ -119,7 +113,7 @@ afterEach(() => {
|
|||
|
||||
describe("spawn", () => {
|
||||
it("creates a session with workspace, runtime, and agent", async () => {
|
||||
const sm = createSessionManager({ config, registry: mockRegistry, eventBus });
|
||||
const sm = createSessionManager({ config, registry: mockRegistry });
|
||||
|
||||
const session = await sm.spawn({ projectId: "my-app" });
|
||||
|
||||
|
|
@ -137,7 +131,7 @@ describe("spawn", () => {
|
|||
});
|
||||
|
||||
it("uses issue ID to derive branch name", async () => {
|
||||
const sm = createSessionManager({ config, registry: mockRegistry, eventBus });
|
||||
const sm = createSessionManager({ config, registry: mockRegistry });
|
||||
|
||||
const session = await sm.spawn({ projectId: "my-app", issueId: "INT-100" });
|
||||
|
||||
|
|
@ -169,7 +163,6 @@ describe("spawn", () => {
|
|||
const sm = createSessionManager({
|
||||
config,
|
||||
registry: registryWithTracker,
|
||||
eventBus,
|
||||
});
|
||||
|
||||
const session = await sm.spawn({ projectId: "my-app", issueId: "INT-100" });
|
||||
|
|
@ -177,7 +170,7 @@ describe("spawn", () => {
|
|||
});
|
||||
|
||||
it("increments session numbers correctly", async () => {
|
||||
const sm = createSessionManager({ config, registry: mockRegistry, eventBus });
|
||||
const sm = createSessionManager({ config, registry: mockRegistry });
|
||||
|
||||
// Pre-create some metadata to simulate existing sessions
|
||||
writeMetadata(dataDir, "app-3", { worktree: "/tmp", branch: "b", status: "working" });
|
||||
|
|
@ -187,20 +180,8 @@ describe("spawn", () => {
|
|||
expect(session.id).toBe("app-8");
|
||||
});
|
||||
|
||||
it("emits session.spawned event", async () => {
|
||||
const received: OrchestratorEvent[] = [];
|
||||
eventBus.on("session.spawned", (e) => received.push(e));
|
||||
|
||||
const sm = createSessionManager({ config, registry: mockRegistry, eventBus });
|
||||
await sm.spawn({ projectId: "my-app" });
|
||||
|
||||
expect(received).toHaveLength(1);
|
||||
expect(received[0].type).toBe("session.spawned");
|
||||
expect(received[0].sessionId).toBe("app-1");
|
||||
});
|
||||
|
||||
it("writes metadata file", async () => {
|
||||
const sm = createSessionManager({ config, registry: mockRegistry, eventBus });
|
||||
const sm = createSessionManager({ config, registry: mockRegistry });
|
||||
await sm.spawn({ projectId: "my-app", issueId: "INT-42" });
|
||||
|
||||
const meta = readMetadata(dataDir, "app-1");
|
||||
|
|
@ -211,7 +192,7 @@ describe("spawn", () => {
|
|||
});
|
||||
|
||||
it("throws for unknown project", async () => {
|
||||
const sm = createSessionManager({ config, registry: mockRegistry, eventBus });
|
||||
const sm = createSessionManager({ config, registry: mockRegistry });
|
||||
await expect(sm.spawn({ projectId: "nonexistent" })).rejects.toThrow("Unknown project");
|
||||
});
|
||||
|
||||
|
|
@ -221,7 +202,7 @@ describe("spawn", () => {
|
|||
get: vi.fn().mockReturnValue(null),
|
||||
};
|
||||
|
||||
const sm = createSessionManager({ config, registry: emptyRegistry, eventBus });
|
||||
const sm = createSessionManager({ config, registry: emptyRegistry });
|
||||
await expect(sm.spawn({ projectId: "my-app" })).rejects.toThrow("not found");
|
||||
});
|
||||
});
|
||||
|
|
@ -241,7 +222,7 @@ describe("list", () => {
|
|||
project: "my-app",
|
||||
});
|
||||
|
||||
const sm = createSessionManager({ config, registry: mockRegistry, eventBus });
|
||||
const sm = createSessionManager({ config, registry: mockRegistry });
|
||||
const sessions = await sm.list();
|
||||
|
||||
expect(sessions).toHaveLength(2);
|
||||
|
|
@ -262,7 +243,7 @@ describe("list", () => {
|
|||
project: "other",
|
||||
});
|
||||
|
||||
const sm = createSessionManager({ config, registry: mockRegistry, eventBus });
|
||||
const sm = createSessionManager({ config, registry: mockRegistry });
|
||||
const sessions = await sm.list("my-app");
|
||||
|
||||
expect(sessions).toHaveLength(1);
|
||||
|
|
@ -291,7 +272,7 @@ describe("list", () => {
|
|||
runtimeHandle: JSON.stringify(makeHandle("rt-1")),
|
||||
});
|
||||
|
||||
const sm = createSessionManager({ config, registry: registryWithDead, eventBus });
|
||||
const sm = createSessionManager({ config, registry: registryWithDead });
|
||||
const sessions = await sm.list();
|
||||
|
||||
expect(sessions[0].status).toBe("killed");
|
||||
|
|
@ -309,7 +290,7 @@ describe("get", () => {
|
|||
pr: "https://github.com/org/repo/pull/42",
|
||||
});
|
||||
|
||||
const sm = createSessionManager({ config, registry: mockRegistry, eventBus });
|
||||
const sm = createSessionManager({ config, registry: mockRegistry });
|
||||
const session = await sm.get("app-1");
|
||||
|
||||
expect(session).not.toBeNull();
|
||||
|
|
@ -320,7 +301,7 @@ describe("get", () => {
|
|||
});
|
||||
|
||||
it("returns null for nonexistent session", async () => {
|
||||
const sm = createSessionManager({ config, registry: mockRegistry, eventBus });
|
||||
const sm = createSessionManager({ config, registry: mockRegistry });
|
||||
expect(await sm.get("nonexistent")).toBeNull();
|
||||
});
|
||||
});
|
||||
|
|
@ -335,7 +316,7 @@ describe("kill", () => {
|
|||
runtimeHandle: JSON.stringify(makeHandle("rt-1")),
|
||||
});
|
||||
|
||||
const sm = createSessionManager({ config, registry: mockRegistry, eventBus });
|
||||
const sm = createSessionManager({ config, registry: mockRegistry });
|
||||
await sm.kill("app-1");
|
||||
|
||||
expect(mockRuntime.destroy).toHaveBeenCalledWith(makeHandle("rt-1"));
|
||||
|
|
@ -343,26 +324,8 @@ describe("kill", () => {
|
|||
expect(readMetadata(dataDir, "app-1")).toBeNull(); // archived + deleted
|
||||
});
|
||||
|
||||
it("emits session.killed event", async () => {
|
||||
writeMetadata(dataDir, "app-1", {
|
||||
worktree: "/tmp",
|
||||
branch: "main",
|
||||
status: "working",
|
||||
project: "my-app",
|
||||
});
|
||||
|
||||
const received: OrchestratorEvent[] = [];
|
||||
eventBus.on("session.killed", (e) => received.push(e));
|
||||
|
||||
const sm = createSessionManager({ config, registry: mockRegistry, eventBus });
|
||||
await sm.kill("app-1");
|
||||
|
||||
expect(received).toHaveLength(1);
|
||||
expect(received[0].sessionId).toBe("app-1");
|
||||
});
|
||||
|
||||
it("throws for nonexistent session", async () => {
|
||||
const sm = createSessionManager({ config, registry: mockRegistry, eventBus });
|
||||
const sm = createSessionManager({ config, registry: mockRegistry });
|
||||
await expect(sm.kill("nonexistent")).rejects.toThrow("not found");
|
||||
});
|
||||
|
||||
|
|
@ -388,7 +351,7 @@ describe("kill", () => {
|
|||
runtimeHandle: JSON.stringify(makeHandle("rt-1")),
|
||||
});
|
||||
|
||||
const sm = createSessionManager({ config, registry: registryWithFail, eventBus });
|
||||
const sm = createSessionManager({ config, registry: registryWithFail });
|
||||
// Should not throw even though runtime.destroy fails
|
||||
await expect(sm.kill("app-1")).resolves.toBeUndefined();
|
||||
});
|
||||
|
|
@ -431,7 +394,7 @@ describe("cleanup", () => {
|
|||
runtimeHandle: JSON.stringify(makeHandle("rt-1")),
|
||||
});
|
||||
|
||||
const sm = createSessionManager({ config, registry: registryWithSCM, eventBus });
|
||||
const sm = createSessionManager({ config, registry: registryWithSCM });
|
||||
const result = await sm.cleanup();
|
||||
|
||||
expect(result.killed).toContain("app-1");
|
||||
|
|
@ -446,7 +409,7 @@ describe("cleanup", () => {
|
|||
project: "my-app",
|
||||
});
|
||||
|
||||
const sm = createSessionManager({ config, registry: mockRegistry, eventBus });
|
||||
const sm = createSessionManager({ config, registry: mockRegistry });
|
||||
const result = await sm.cleanup();
|
||||
|
||||
expect(result.killed).toHaveLength(0);
|
||||
|
|
@ -476,7 +439,7 @@ describe("cleanup", () => {
|
|||
runtimeHandle: JSON.stringify(makeHandle("rt-1")),
|
||||
});
|
||||
|
||||
const sm = createSessionManager({ config, registry: registryWithDead, eventBus });
|
||||
const sm = createSessionManager({ config, registry: registryWithDead });
|
||||
const result = await sm.cleanup();
|
||||
|
||||
expect(result.killed).toContain("app-1");
|
||||
|
|
@ -493,14 +456,14 @@ describe("send", () => {
|
|||
runtimeHandle: JSON.stringify(makeHandle("rt-1")),
|
||||
});
|
||||
|
||||
const sm = createSessionManager({ config, registry: mockRegistry, eventBus });
|
||||
const sm = createSessionManager({ config, registry: mockRegistry });
|
||||
await sm.send("app-1", "Fix the CI failures");
|
||||
|
||||
expect(mockRuntime.sendMessage).toHaveBeenCalledWith(makeHandle("rt-1"), "Fix the CI failures");
|
||||
});
|
||||
|
||||
it("throws for nonexistent session", async () => {
|
||||
const sm = createSessionManager({ config, registry: mockRegistry, eventBus });
|
||||
const sm = createSessionManager({ config, registry: mockRegistry });
|
||||
await expect(sm.send("nope", "hello")).rejects.toThrow("not found");
|
||||
});
|
||||
|
||||
|
|
@ -512,7 +475,7 @@ describe("send", () => {
|
|||
project: "my-app",
|
||||
});
|
||||
|
||||
const sm = createSessionManager({ config, registry: mockRegistry, eventBus });
|
||||
const sm = createSessionManager({ config, registry: mockRegistry });
|
||||
await expect(sm.send("app-1", "hello")).rejects.toThrow("No runtime handle");
|
||||
});
|
||||
});
|
||||
|
|
|
|||
|
|
@ -1,206 +0,0 @@
|
|||
/**
|
||||
* Event Bus — in-process pub/sub with JSONL persistence.
|
||||
*
|
||||
* Events are OrchestratorEvent objects. Supports:
|
||||
* - Typed event listeners (by EventType)
|
||||
* - Wildcard listeners ("*")
|
||||
* - JSONL file persistence (append-only)
|
||||
* - Filtered history queries
|
||||
*/
|
||||
|
||||
import { appendFileSync, readFileSync, existsSync, mkdirSync } from "node:fs";
|
||||
import { dirname } from "node:path";
|
||||
import { randomUUID } from "node:crypto";
|
||||
import type {
|
||||
EventBus,
|
||||
EventFilter,
|
||||
EventType,
|
||||
EventPriority,
|
||||
OrchestratorEvent,
|
||||
SessionId,
|
||||
} from "./types.js";
|
||||
|
||||
type EventHandler = (event: OrchestratorEvent) => void;
|
||||
|
||||
/**
|
||||
* Create an OrchestratorEvent with defaults filled in.
|
||||
*/
|
||||
export function createEvent(
|
||||
type: EventType,
|
||||
opts: {
|
||||
sessionId: SessionId;
|
||||
projectId: string;
|
||||
message: string;
|
||||
priority?: EventPriority;
|
||||
data?: Record<string, unknown>;
|
||||
},
|
||||
): OrchestratorEvent {
|
||||
return {
|
||||
id: randomUUID(),
|
||||
type,
|
||||
priority: opts.priority ?? inferPriority(type),
|
||||
sessionId: opts.sessionId,
|
||||
projectId: opts.projectId,
|
||||
timestamp: new Date(),
|
||||
message: opts.message,
|
||||
data: opts.data ?? {},
|
||||
};
|
||||
}
|
||||
|
||||
/** Infer a reasonable priority from event type. */
|
||||
function inferPriority(type: EventType): EventPriority {
|
||||
if (type.includes("stuck") || type.includes("needs_input") || type.includes("errored")) {
|
||||
return "urgent";
|
||||
}
|
||||
// summary.all_complete is informational, not actionable
|
||||
if (type.startsWith("summary.")) {
|
||||
return "info";
|
||||
}
|
||||
if (
|
||||
type.includes("approved") ||
|
||||
type.includes("ready") ||
|
||||
type.includes("merged") ||
|
||||
type.includes("completed")
|
||||
) {
|
||||
return "action";
|
||||
}
|
||||
if (type.includes("fail") || type.includes("changes_requested") || type.includes("conflicts")) {
|
||||
return "warning";
|
||||
}
|
||||
return "info";
|
||||
}
|
||||
|
||||
/** Serialize an event for JSONL storage. */
|
||||
function serializeEvent(event: OrchestratorEvent): string {
|
||||
return JSON.stringify({
|
||||
...event,
|
||||
timestamp: event.timestamp.toISOString(),
|
||||
});
|
||||
}
|
||||
|
||||
/** Deserialize an event from JSONL. */
|
||||
function deserializeEvent(line: string): OrchestratorEvent | null {
|
||||
try {
|
||||
const raw = JSON.parse(line) as Record<string, unknown>;
|
||||
return {
|
||||
...raw,
|
||||
timestamp: new Date(raw["timestamp"] as string),
|
||||
} as OrchestratorEvent;
|
||||
} catch {
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Create an EventBus implementation.
|
||||
*
|
||||
* @param logPath - Path to the JSONL event log file. If null, persistence is disabled.
|
||||
*/
|
||||
export function createEventBus(logPath: string | null): EventBus {
|
||||
const handlers = new Map<string, Set<EventHandler>>();
|
||||
const history: OrchestratorEvent[] = [];
|
||||
|
||||
// Load existing history from JSONL file
|
||||
if (logPath && existsSync(logPath)) {
|
||||
const content = readFileSync(logPath, "utf-8");
|
||||
for (const line of content.split("\n")) {
|
||||
if (!line.trim()) continue;
|
||||
const event = deserializeEvent(line);
|
||||
if (event) history.push(event);
|
||||
}
|
||||
}
|
||||
|
||||
// Ensure log directory exists
|
||||
if (logPath) {
|
||||
mkdirSync(dirname(logPath), { recursive: true });
|
||||
}
|
||||
|
||||
function getOrCreateSet(key: string): Set<EventHandler> {
|
||||
let set = handlers.get(key);
|
||||
if (!set) {
|
||||
set = new Set();
|
||||
handlers.set(key, set);
|
||||
}
|
||||
return set;
|
||||
}
|
||||
|
||||
return {
|
||||
emit(event: OrchestratorEvent): void {
|
||||
// Persist to JSONL (non-fatal on disk errors)
|
||||
if (logPath) {
|
||||
try {
|
||||
appendFileSync(logPath, serializeEvent(event) + "\n", "utf-8");
|
||||
} catch {
|
||||
// Disk error — event still propagated in-memory
|
||||
}
|
||||
}
|
||||
|
||||
// Store in memory
|
||||
history.push(event);
|
||||
|
||||
// Notify type-specific handlers
|
||||
const typeHandlers = handlers.get(event.type);
|
||||
if (typeHandlers) {
|
||||
for (const handler of typeHandlers) {
|
||||
try {
|
||||
handler(event);
|
||||
} catch {
|
||||
// Don't let handler errors break the bus
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Notify wildcard handlers
|
||||
const wildcardHandlers = handlers.get("*");
|
||||
if (wildcardHandlers) {
|
||||
for (const handler of wildcardHandlers) {
|
||||
try {
|
||||
handler(event);
|
||||
} catch {
|
||||
// Don't let handler errors break the bus
|
||||
}
|
||||
}
|
||||
}
|
||||
},
|
||||
|
||||
on(event: EventType | "*", handler: EventHandler): void {
|
||||
getOrCreateSet(event).add(handler);
|
||||
},
|
||||
|
||||
off(event: EventType | "*", handler: EventHandler): void {
|
||||
const set = handlers.get(event);
|
||||
if (set) {
|
||||
set.delete(handler);
|
||||
if (set.size === 0) handlers.delete(event);
|
||||
}
|
||||
},
|
||||
|
||||
getHistory(filter?: EventFilter): OrchestratorEvent[] {
|
||||
let result = history;
|
||||
|
||||
if (filter) {
|
||||
if (filter.sessionId) {
|
||||
result = result.filter((e) => e.sessionId === filter.sessionId);
|
||||
}
|
||||
if (filter.projectId) {
|
||||
result = result.filter((e) => e.projectId === filter.projectId);
|
||||
}
|
||||
if (filter.type) {
|
||||
result = result.filter((e) => e.type === filter.type);
|
||||
}
|
||||
if (filter.priority) {
|
||||
result = result.filter((e) => e.priority === filter.priority);
|
||||
}
|
||||
if (filter.since) {
|
||||
const since = filter.since;
|
||||
result = result.filter((e) => e.timestamp >= since);
|
||||
}
|
||||
if (filter.limit) {
|
||||
result = result.slice(-filter.limit);
|
||||
}
|
||||
}
|
||||
|
||||
return result;
|
||||
},
|
||||
};
|
||||
}
|
||||
|
|
@ -24,9 +24,6 @@ export {
|
|||
listMetadata,
|
||||
} from "./metadata.js";
|
||||
|
||||
// Event bus — pub/sub with JSONL persistence
|
||||
export { createEventBus, createEvent } from "./event-bus.js";
|
||||
|
||||
// tmux — command wrappers
|
||||
export {
|
||||
isTmuxAvailable,
|
||||
|
|
|
|||
|
|
@ -10,13 +10,13 @@
|
|||
* Reference: scripts/claude-session-status, scripts/claude-review-check
|
||||
*/
|
||||
|
||||
import { randomUUID } from "node:crypto";
|
||||
import type {
|
||||
LifecycleManager,
|
||||
SessionManager,
|
||||
SessionId,
|
||||
SessionStatus,
|
||||
EventType,
|
||||
EventBus,
|
||||
OrchestratorEvent,
|
||||
OrchestratorConfig,
|
||||
ReactionConfig,
|
||||
|
|
@ -29,11 +29,8 @@ import type {
|
|||
Session,
|
||||
EventPriority,
|
||||
} from "./types.js";
|
||||
import { createEvent } from "./event-bus.js";
|
||||
import { updateMetadata } from "./metadata.js";
|
||||
|
||||
type EventHandler = (event: OrchestratorEvent) => void;
|
||||
|
||||
/** Parse a duration string like "10m", "30s", "1h" to milliseconds. */
|
||||
function parseDuration(str: string): number {
|
||||
const match = str.match(/^(\d+)(s|m|h)$/);
|
||||
|
|
@ -51,6 +48,51 @@ function parseDuration(str: string): number {
|
|||
}
|
||||
}
|
||||
|
||||
/** Infer a reasonable priority from event type. */
|
||||
function inferPriority(type: EventType): EventPriority {
|
||||
if (type.includes("stuck") || type.includes("needs_input") || type.includes("errored")) {
|
||||
return "urgent";
|
||||
}
|
||||
if (type.startsWith("summary.")) {
|
||||
return "info";
|
||||
}
|
||||
if (
|
||||
type.includes("approved") ||
|
||||
type.includes("ready") ||
|
||||
type.includes("merged") ||
|
||||
type.includes("completed")
|
||||
) {
|
||||
return "action";
|
||||
}
|
||||
if (type.includes("fail") || type.includes("changes_requested") || type.includes("conflicts")) {
|
||||
return "warning";
|
||||
}
|
||||
return "info";
|
||||
}
|
||||
|
||||
/** Create an OrchestratorEvent with defaults filled in. */
|
||||
function createEvent(
|
||||
type: EventType,
|
||||
opts: {
|
||||
sessionId: SessionId;
|
||||
projectId: string;
|
||||
message: string;
|
||||
priority?: EventPriority;
|
||||
data?: Record<string, unknown>;
|
||||
},
|
||||
): OrchestratorEvent {
|
||||
return {
|
||||
id: randomUUID(),
|
||||
type,
|
||||
priority: opts.priority ?? inferPriority(type),
|
||||
sessionId: opts.sessionId,
|
||||
projectId: opts.projectId,
|
||||
timestamp: new Date(),
|
||||
message: opts.message,
|
||||
data: opts.data ?? {},
|
||||
};
|
||||
}
|
||||
|
||||
/** Determine which event type corresponds to a status transition. */
|
||||
function statusToEventType(_from: SessionStatus | undefined, to: SessionStatus): EventType | null {
|
||||
switch (to) {
|
||||
|
|
@ -113,7 +155,6 @@ export interface LifecycleManagerDeps {
|
|||
config: OrchestratorConfig;
|
||||
registry: PluginRegistry;
|
||||
sessionManager: SessionManager;
|
||||
eventBus: EventBus;
|
||||
}
|
||||
|
||||
/** Track attempt counts for reactions per session. */
|
||||
|
|
@ -124,47 +165,14 @@ interface ReactionTracker {
|
|||
|
||||
/** Create a LifecycleManager instance. */
|
||||
export function createLifecycleManager(deps: LifecycleManagerDeps): LifecycleManager {
|
||||
const { config, registry, sessionManager, eventBus } = deps;
|
||||
const { config, registry, sessionManager } = deps;
|
||||
|
||||
const states = new Map<SessionId, SessionStatus>();
|
||||
const handlers = new Map<string, Set<EventHandler>>();
|
||||
const reactionTrackers = new Map<string, ReactionTracker>(); // "sessionId:reactionKey"
|
||||
let pollTimer: ReturnType<typeof setInterval> | null = null;
|
||||
let polling = false; // re-entrancy guard
|
||||
let allCompleteEmitted = false; // guard against repeated all_complete
|
||||
|
||||
function getOrCreateSet(key: string): Set<EventHandler> {
|
||||
let set = handlers.get(key);
|
||||
if (!set) {
|
||||
set = new Set();
|
||||
handlers.set(key, set);
|
||||
}
|
||||
return set;
|
||||
}
|
||||
|
||||
function emitToHandlers(event: OrchestratorEvent): void {
|
||||
const typeHandlers = handlers.get(event.type);
|
||||
if (typeHandlers) {
|
||||
for (const handler of typeHandlers) {
|
||||
try {
|
||||
handler(event);
|
||||
} catch {
|
||||
/* ignore */
|
||||
}
|
||||
}
|
||||
}
|
||||
const wildcardHandlers = handlers.get("*");
|
||||
if (wildcardHandlers) {
|
||||
for (const handler of wildcardHandlers) {
|
||||
try {
|
||||
handler(event);
|
||||
} catch {
|
||||
/* ignore */
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/** Determine current status for a session by polling plugins. */
|
||||
async function determineStatus(session: Session): Promise<SessionStatus> {
|
||||
const project = config.projects[session.projectId];
|
||||
|
|
@ -229,13 +237,14 @@ export function createLifecycleManager(deps: LifecycleManagerDeps): LifecycleMan
|
|||
return session.status;
|
||||
}
|
||||
|
||||
/** Execute a reaction for an event. */
|
||||
/** Execute a reaction for a session. */
|
||||
async function executeReaction(
|
||||
event: OrchestratorEvent,
|
||||
sessionId: SessionId,
|
||||
projectId: string,
|
||||
reactionKey: string,
|
||||
reactionConfig: ReactionConfig,
|
||||
): Promise<ReactionResult> {
|
||||
const trackerKey = `${event.sessionId}:${reactionKey}`;
|
||||
const trackerKey = `${sessionId}:${reactionKey}`;
|
||||
let tracker = reactionTrackers.get(trackerKey);
|
||||
|
||||
if (!tracker) {
|
||||
|
|
@ -268,6 +277,12 @@ export function createLifecycleManager(deps: LifecycleManagerDeps): LifecycleMan
|
|||
|
||||
if (shouldEscalate) {
|
||||
// Escalate to human
|
||||
const event = createEvent("reaction.escalated", {
|
||||
sessionId,
|
||||
projectId,
|
||||
message: `Reaction '${reactionKey}' escalated after ${tracker.attempts} attempts`,
|
||||
data: { reactionKey, attempts: tracker.attempts },
|
||||
});
|
||||
await notifyHuman(event, reactionConfig.priority ?? "urgent");
|
||||
return {
|
||||
reactionType: reactionKey,
|
||||
|
|
@ -284,16 +299,7 @@ export function createLifecycleManager(deps: LifecycleManagerDeps): LifecycleMan
|
|||
case "send-to-agent": {
|
||||
if (reactionConfig.message) {
|
||||
try {
|
||||
await sessionManager.send(event.sessionId, reactionConfig.message);
|
||||
|
||||
eventBus.emit(
|
||||
createEvent("reaction.triggered", {
|
||||
sessionId: event.sessionId,
|
||||
projectId: event.projectId,
|
||||
message: `Reaction '${reactionKey}' sent message to agent`,
|
||||
data: { reactionKey, attempts: tracker.attempts },
|
||||
}),
|
||||
);
|
||||
await sessionManager.send(sessionId, reactionConfig.message);
|
||||
|
||||
return {
|
||||
reactionType: reactionKey,
|
||||
|
|
@ -316,6 +322,12 @@ export function createLifecycleManager(deps: LifecycleManagerDeps): LifecycleMan
|
|||
}
|
||||
|
||||
case "notify": {
|
||||
const event = createEvent("reaction.triggered", {
|
||||
sessionId,
|
||||
projectId,
|
||||
message: `Reaction '${reactionKey}' triggered notification`,
|
||||
data: { reactionKey },
|
||||
});
|
||||
await notifyHuman(event, reactionConfig.priority ?? "info");
|
||||
return {
|
||||
reactionType: reactionKey,
|
||||
|
|
@ -328,6 +340,12 @@ export function createLifecycleManager(deps: LifecycleManagerDeps): LifecycleMan
|
|||
case "auto-merge": {
|
||||
// Auto-merge is handled by the SCM plugin
|
||||
// For now, just notify
|
||||
const event = createEvent("reaction.triggered", {
|
||||
sessionId,
|
||||
projectId,
|
||||
message: `Reaction '${reactionKey}' triggered auto-merge`,
|
||||
data: { reactionKey },
|
||||
});
|
||||
await notifyHuman(event, "action");
|
||||
return {
|
||||
reactionType: reactionKey,
|
||||
|
|
@ -394,20 +412,9 @@ export function createLifecycleManager(deps: LifecycleManagerDeps): LifecycleMan
|
|||
}
|
||||
}
|
||||
|
||||
// Emit transition event
|
||||
// Check if there's a reaction for this transition
|
||||
const eventType = statusToEventType(oldStatus, newStatus);
|
||||
if (eventType) {
|
||||
const event = createEvent(eventType, {
|
||||
sessionId: session.id,
|
||||
projectId: session.projectId,
|
||||
message: `${session.id}: ${oldStatus} → ${newStatus}`,
|
||||
data: { oldStatus, newStatus },
|
||||
});
|
||||
|
||||
eventBus.emit(event);
|
||||
emitToHandlers(event);
|
||||
|
||||
// Check if there's a reaction for this event
|
||||
const reactionKey = eventToReactionKey(eventType);
|
||||
if (reactionKey) {
|
||||
// Merge project-specific overrides with global defaults
|
||||
|
|
@ -421,7 +428,12 @@ export function createLifecycleManager(deps: LifecycleManagerDeps): LifecycleMan
|
|||
if (reactionConfig && reactionConfig.action) {
|
||||
// auto: false skips automated agent actions but still allows notifications
|
||||
if (reactionConfig.auto !== false || reactionConfig.action === "notify") {
|
||||
await executeReaction(event, reactionKey, reactionConfig as ReactionConfig);
|
||||
await executeReaction(
|
||||
session.id,
|
||||
session.projectId,
|
||||
reactionKey,
|
||||
reactionConfig as ReactionConfig,
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -468,18 +480,10 @@ export function createLifecycleManager(deps: LifecycleManagerDeps): LifecycleMan
|
|||
}
|
||||
}
|
||||
|
||||
// Check if all sessions are complete (emit only once)
|
||||
// Check if all sessions are complete (trigger reaction only once)
|
||||
const activeSessions = sessions.filter((s) => s.status !== "merged" && s.status !== "killed");
|
||||
if (sessions.length > 0 && activeSessions.length === 0 && !allCompleteEmitted) {
|
||||
allCompleteEmitted = true;
|
||||
const event = createEvent("summary.all_complete", {
|
||||
sessionId: "system",
|
||||
projectId: "all",
|
||||
message: `All ${sessions.length} sessions complete`,
|
||||
data: { totalSessions: sessions.length },
|
||||
});
|
||||
eventBus.emit(event);
|
||||
emitToHandlers(event);
|
||||
|
||||
// Execute all-complete reaction if configured
|
||||
const reactionKey = eventToReactionKey("summary.all_complete");
|
||||
|
|
@ -487,7 +491,7 @@ export function createLifecycleManager(deps: LifecycleManagerDeps): LifecycleMan
|
|||
const reactionConfig = config.reactions[reactionKey];
|
||||
if (reactionConfig && reactionConfig.action) {
|
||||
if (reactionConfig.auto !== false || reactionConfig.action === "notify") {
|
||||
await executeReaction(event, reactionKey, reactionConfig as ReactionConfig);
|
||||
await executeReaction("system", "all", reactionKey, reactionConfig as ReactionConfig);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -523,17 +527,5 @@ export function createLifecycleManager(deps: LifecycleManagerDeps): LifecycleMan
|
|||
if (!session) throw new Error(`Session ${sessionId} not found`);
|
||||
await checkSession(session);
|
||||
},
|
||||
|
||||
on(event: EventType | "*", handler: EventHandler): void {
|
||||
getOrCreateSet(event).add(handler);
|
||||
},
|
||||
|
||||
off(event: EventType | "*", handler: EventHandler): void {
|
||||
const set = handlers.get(event);
|
||||
if (set) {
|
||||
set.delete(handler);
|
||||
if (set.size === 0) handlers.delete(event);
|
||||
}
|
||||
},
|
||||
};
|
||||
}
|
||||
|
|
|
|||
|
|
@ -19,6 +19,7 @@ import {
|
|||
mkdirSync,
|
||||
unlinkSync,
|
||||
readdirSync,
|
||||
statSync,
|
||||
openSync,
|
||||
closeSync,
|
||||
constants,
|
||||
|
|
@ -189,7 +190,14 @@ export function listMetadata(dataDir: string): SessionId[] {
|
|||
const dir = join(dataDir, "sessions");
|
||||
if (!existsSync(dir)) return [];
|
||||
|
||||
return readdirSync(dir).filter((name) => name !== "archive" && !name.startsWith("."));
|
||||
return readdirSync(dir).filter((name) => {
|
||||
if (name === "archive" || name.startsWith(".")) return false;
|
||||
try {
|
||||
return statSync(join(dir, name)).isFile();
|
||||
} catch {
|
||||
return false;
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
|
|||
|
|
@ -25,7 +25,6 @@ import type {
|
|||
Workspace,
|
||||
Tracker,
|
||||
SCM,
|
||||
EventBus,
|
||||
PluginRegistry,
|
||||
RuntimeHandle,
|
||||
} from "./types.js";
|
||||
|
|
@ -36,7 +35,6 @@ import {
|
|||
listMetadata,
|
||||
reserveSessionId,
|
||||
} from "./metadata.js";
|
||||
import { createEvent } from "./event-bus.js";
|
||||
|
||||
/** Escape regex metacharacters in a string. */
|
||||
function escapeRegex(str: string): string {
|
||||
|
|
@ -125,12 +123,11 @@ function metadataToSession(sessionId: SessionId, meta: Record<string, string>):
|
|||
export interface SessionManagerDeps {
|
||||
config: OrchestratorConfig;
|
||||
registry: PluginRegistry;
|
||||
eventBus: EventBus;
|
||||
}
|
||||
|
||||
/** Create a SessionManager instance. */
|
||||
export function createSessionManager(deps: SessionManagerDeps): SessionManager {
|
||||
const { config, registry, eventBus } = deps;
|
||||
const { config, registry } = deps;
|
||||
|
||||
/** Resolve which plugins to use for a project. */
|
||||
function resolvePlugins(project: ProjectConfig) {
|
||||
|
|
@ -321,16 +318,6 @@ export function createSessionManager(deps: SessionManagerDeps): SessionManager {
|
|||
throw err;
|
||||
}
|
||||
|
||||
// Emit event
|
||||
eventBus.emit(
|
||||
createEvent("session.spawned", {
|
||||
sessionId,
|
||||
projectId: spawnConfig.projectId,
|
||||
message: `Session ${sessionId} spawned${spawnConfig.issueId ? ` for ${spawnConfig.issueId}` : ""}`,
|
||||
data: { branch, workspacePath, issueId: spawnConfig.issueId },
|
||||
}),
|
||||
);
|
||||
|
||||
return session;
|
||||
}
|
||||
|
||||
|
|
@ -421,15 +408,6 @@ export function createSessionManager(deps: SessionManagerDeps): SessionManager {
|
|||
|
||||
// Archive metadata
|
||||
deleteMetadata(config.dataDir, sessionId, true);
|
||||
|
||||
// Emit event
|
||||
eventBus.emit(
|
||||
createEvent("session.killed", {
|
||||
sessionId,
|
||||
projectId,
|
||||
message: `Session ${sessionId} killed`,
|
||||
}),
|
||||
);
|
||||
}
|
||||
|
||||
async function cleanup(projectId?: string): Promise<CleanupResult> {
|
||||
|
|
|
|||
|
|
@ -800,27 +800,6 @@ export interface LifecycleManager {
|
|||
|
||||
/** Force-check a specific session now */
|
||||
check(sessionId: SessionId): Promise<void>;
|
||||
|
||||
/** Subscribe to lifecycle events */
|
||||
on(event: EventType | "*", handler: (event: OrchestratorEvent) => void): void;
|
||||
off(event: EventType | "*", handler: (event: OrchestratorEvent) => void): void;
|
||||
}
|
||||
|
||||
/** Event bus — pub/sub + persistence */
|
||||
export interface EventBus {
|
||||
emit(event: OrchestratorEvent): void;
|
||||
on(event: EventType | "*", handler: (event: OrchestratorEvent) => void): void;
|
||||
off(event: EventType | "*", handler: (event: OrchestratorEvent) => void): void;
|
||||
getHistory(filter?: EventFilter): OrchestratorEvent[];
|
||||
}
|
||||
|
||||
export interface EventFilter {
|
||||
sessionId?: SessionId;
|
||||
projectId?: string;
|
||||
type?: EventType;
|
||||
priority?: EventPriority;
|
||||
since?: Date;
|
||||
limit?: number;
|
||||
}
|
||||
|
||||
/** Plugin registry — discovery + loading */
|
||||
|
|
|
|||
Loading…
Reference in New Issue