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

feat: event-driven reactions architecture — event log, bugbot detection, persistent retrigger 

Core infrastructure for the autonomous reactions system:

**Event Log (`packages/core/src/event-log.ts`)**
- New JSONL append-only event log for all orchestrator events
- `createEventLog(logPath)` — writes to `~/.agent-orchestrator/{hash}-events.jsonl`
- `createNullEventLog()` — no-op for tests/opt-out
- `getEventLogPath(configPath)` added to paths.ts
- All state transitions and reaction executions are now logged
- Exported from `@composio/ao-core`

**Automated Comment Detection (bugbot)**
- Polls `scm.getAutomatedComments()` every cycle when a PR exists
- Emits `automated_review.found` event with comment details
- Triggers `bugbot-comments` reaction (send-to-agent by default)
- Deduplication via comment-ID fingerprinting — only retriggers when
  new/changed comments appear, not every poll cycle

**Persistent State Retrigger (`retriggerAfter`)**
- New `retriggerAfter?: string` field on `ReactionConfig` (e.g. "10m")
- When a session stays in a bad state (ci_failed, changes_requested, etc.)
  without transitioning, reactions re-fire after the configured interval
- Handles the case where an agent received CI failure message but didn't fix it
- Combined with `retries`/`escalateAfter` to eventually escalate to human
- Opt-in per reaction — preserves existing behaviour when not configured

**Default reactions updated:**
- `ci-failed`: retriggerAfter 15m
- `changes-requested`: retriggerAfter 20m
- `bugbot-comments`: retriggerAfter 20m (new default)
- `merge-conflicts`: retriggerAfter 10m

**Architecture coordination with ao-48:**
- Review comment polling (`getPendingComments`) is explicitly NOT implemented here
- ao-48 can add it using the same checkAutomatedComments pattern
- `eventToReactionKey("review.changes_requested")` → "changes-requested" left intact

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Prateek 2026-02-18 12:03:47 +05:30
parent 40c1906d41
commit 461be2b47e
8 changed files with 775 additions and 25 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

31
agent-orchestrator.yaml.example
View File

@ -80,17 +80,35 @@ projects:
# info: [slack] # summary, all done
# Reactions — auto-responses to events (these are the defaults)
# Each event type maps to a reaction key. All defaults can be overridden here
# or per-project under projects.<id>.reactions.<key>.
#
# reactions:
# ci-failed:
# auto: true
# action: send-to-agent
# retries: 2
# escalateAfter: 2
# message: "CI is failing on your PR. Run `gh pr checks` to see the failures, fix them, and push."
# retries: 2 # try sending to agent 2 times before escalating
# escalateAfter: 2 # after 2 failures, escalate to human
# retriggerAfter: 15m # re-send if CI is still failing after 15 minutes
#
# changes-requested:
# auto: true
# action: send-to-agent
# escalateAfter: 30m
# retriggerAfter: 20m # re-send if review comments still unaddressed after 20 minutes
#
# bugbot-comments:
# auto: true # react to automated review bot comments
# action: send-to-agent
# escalateAfter: 30m
# retriggerAfter: 20m # re-send when new bot comments appear
#
# merge-conflicts:
# auto: true
# action: send-to-agent
# escalateAfter: 15m
# retriggerAfter: 10m
#
# approved-and-green:
# auto: false # set to true for auto-merge
@ -101,3 +119,12 @@ projects:
# threshold: 10m
# action: notify
# priority: urgent
#
# agent-needs-input:
# action: notify
# priority: urgent
#
# all-complete:
# action: notify
# priority: info
# includeSummary: true # include session summary in notification

412
packages/core/src/__tests__/lifecycle-manager.test.ts
View File

@ -4,6 +4,7 @@ import { join } from "node:path";
import { tmpdir } from "node:os";
import { randomUUID } from "node:crypto";
import { createLifecycleManager } from "../lifecycle-manager.js";
import { createNullEventLog } from "../event-log.js";
import { writeMetadata, readMetadataRaw } from "../metadata.js";
import { getSessionsDir, getProjectBaseDir } from "../paths.js";
import type {
@ -17,6 +18,8 @@ import type {
Notifier,
ActivityState,
PRInfo,
EventLog,
OrchestratorEvent,
} from "../types.js";
let tmpDir: string;
@ -865,3 +868,412 @@ describe("getStates", () => {
expect(lm.getStates().get("app-1")).toBe("working");
});
});
describe("event log integration", () => {
it("logs events to the provided event log on state transition", async () => {
const loggedEvents: OrchestratorEvent[] = [];
const eventLog: EventLog = {
log: (event) => loggedEvents.push(event),
readRecent: () => [],
};
const session = makeSession({ status: "spawning" });
vi.mocked(mockSessionManager.get).mockResolvedValue(session);
writeMetadata(sessionsDir, "app-1", {
worktree: "/tmp",
branch: "main",
status: "spawning",
project: "my-app",
});
const lm = createLifecycleManager({
config,
registry: mockRegistry,
sessionManager: mockSessionManager,
eventLog,
});
await lm.check("app-1");
expect(lm.getStates().get("app-1")).toBe("working");
expect(loggedEvents.length).toBeGreaterThan(0);
expect(loggedEvents[0]).toMatchObject({
type: "session.working",
sessionId: "app-1",
projectId: "my-app",
});
});
it("accepts createNullEventLog as a no-op default", async () => {
const nullLog = createNullEventLog();
const session = makeSession({ status: "spawning" });
vi.mocked(mockSessionManager.get).mockResolvedValue(session);
writeMetadata(sessionsDir, "app-1", {
worktree: "/tmp",
branch: "main",
status: "spawning",
project: "my-app",
});
const lm = createLifecycleManager({
config,
registry: mockRegistry,
sessionManager: mockSessionManager,
eventLog: nullLog,
});
// Should not throw
await lm.check("app-1");
expect(lm.getStates().get("app-1")).toBe("working");
expect(nullLog.readRecent()).toEqual([]);
});
});
describe("automated comment detection (bugbot)", () => {
function makeSCMWithAutomatedComments(
automatedComments: ReturnType<SCM["getAutomatedComments"]> extends Promise<infer T>
? T
: never,
): SCM {
return {
name: "mock-scm",
detectPR: vi.fn(),
getPRState: vi.fn().mockResolvedValue("open"),
mergePR: vi.fn(),
closePR: vi.fn(),
getCIChecks: vi.fn(),
getCISummary: vi.fn().mockResolvedValue("passing"),
getReviews: vi.fn(),
getReviewDecision: vi.fn().mockResolvedValue("none"),
getPendingComments: vi.fn().mockResolvedValue([]),
getAutomatedComments: vi.fn().mockResolvedValue(automatedComments),
getMergeability: vi.fn(),
};
}
it("triggers bugbot-comments reaction when automated comments appear", async () => {
const botComment = {
id: "comment-1",
botName: "reviewdog",
body: "Error: unused import",
path: "src/index.ts",
line: 5,
severity: "error" as const,
createdAt: new Date(),
url: "https://github.com/org/repo/pull/42#comment-1",
};
config.reactions = {
"bugbot-comments": {
auto: true,
action: "send-to-agent",
message: "Fix bot comments",
},
};
const mockSCM = makeSCMWithAutomatedComments([botComment]);
const registryWithSCM: PluginRegistry = {
...mockRegistry,
get: vi.fn().mockImplementation((slot: string) => {
if (slot === "runtime") return mockRuntime;
if (slot === "agent") return mockAgent;
if (slot === "scm") return mockSCM;
return null;
}),
};
// Session in pr_open state — status doesn't change, but automated comments appear
const session = makeSession({ status: "pr_open", pr: makePR() });
vi.mocked(mockSessionManager.get).mockResolvedValue(session);
writeMetadata(sessionsDir, "app-1", {
worktree: "/tmp",
branch: "main",
status: "pr_open",
project: "my-app",
});
const lm = createLifecycleManager({
config,
registry: registryWithSCM,
sessionManager: mockSessionManager,
});
// First check: sets pr_open state
await lm.check("app-1");
// Second check: status unchanged (pr_open → pr_open) → automated comments check fires
await lm.check("app-1");
expect(mockSessionManager.send).toHaveBeenCalledWith("app-1", "Fix bot comments");
});
it("deduplicates automated comments — does not retrigger for same comment set", async () => {
const botComment = {
id: "comment-1",
botName: "reviewdog",
body: "Error: unused import",
severity: "error" as const,
createdAt: new Date(),
url: "https://github.com/org/repo/pull/42#comment-1",
};
config.reactions = {
"bugbot-comments": {
auto: true,
action: "send-to-agent",
message: "Fix bot comments",
},
};
const mockSCM = makeSCMWithAutomatedComments([botComment]);
const registryWithSCM: PluginRegistry = {
...mockRegistry,
get: vi.fn().mockImplementation((slot: string) => {
if (slot === "runtime") return mockRuntime;
if (slot === "agent") return mockAgent;
if (slot === "scm") return mockSCM;
return null;
}),
};
const session = makeSession({ status: "pr_open", pr: makePR() });
vi.mocked(mockSessionManager.get).mockResolvedValue(session);
writeMetadata(sessionsDir, "app-1", {
worktree: "/tmp",
branch: "main",
status: "pr_open",
project: "my-app",
});
const lm = createLifecycleManager({
config,
registry: registryWithSCM,
sessionManager: mockSessionManager,
});
// First check: transitions pr_open → pr_open (status unchanged)
// Automated comments check NOT called on first encounter (no prior state to compare)
await lm.check("app-1");
// Second check: same status → automated check fires, sees comment-1, sends message
await lm.check("app-1");
expect(mockSessionManager.send).toHaveBeenCalledTimes(1);
// Third check: same comment set → fingerprint matches, no resend
await lm.check("app-1");
expect(mockSessionManager.send).toHaveBeenCalledTimes(1);
});
it("retriggers when new automated comments appear (fingerprint changes)", async () => {
const firstComment = {
id: "comment-1",
botName: "reviewdog",
body: "Error 1",
severity: "error" as const,
createdAt: new Date(),
url: "https://github.com/org/repo/pull/42#comment-1",
};
const secondComment = {
id: "comment-2",
botName: "reviewdog",
body: "Error 2",
severity: "error" as const,
createdAt: new Date(),
url: "https://github.com/org/repo/pull/42#comment-2",
};
config.reactions = {
"bugbot-comments": {
auto: true,
action: "send-to-agent",
message: "Fix bot comments",
},
};
const getAutomatedComments = vi.fn().mockResolvedValue([firstComment]);
const mockSCM: SCM = {
...makeSCMWithAutomatedComments([firstComment]),
getAutomatedComments,
};
const registryWithSCM: PluginRegistry = {
...mockRegistry,
get: vi.fn().mockImplementation((slot: string) => {
if (slot === "runtime") return mockRuntime;
if (slot === "agent") return mockAgent;
if (slot === "scm") return mockSCM;
return null;
}),
};
const session = makeSession({ status: "pr_open", pr: makePR() });
vi.mocked(mockSessionManager.get).mockResolvedValue(session);
writeMetadata(sessionsDir, "app-1", {
worktree: "/tmp",
branch: "main",
status: "pr_open",
project: "my-app",
});
const lm = createLifecycleManager({
config,
registry: registryWithSCM,
sessionManager: mockSessionManager,
});
// Check 1: transitions to pr_open (initial)
await lm.check("app-1");
// Check 2: automated comment check fires with comment-1 → sends message once
await lm.check("app-1");
expect(mockSessionManager.send).toHaveBeenCalledTimes(1);
// Now a new comment appears
getAutomatedComments.mockResolvedValue([firstComment, secondComment]);
// Check 3: fingerprint changes → retrigger
await lm.check("app-1");
expect(mockSessionManager.send).toHaveBeenCalledTimes(2);
});
});
describe("persistent retrigger (retriggerAfter)", () => {
it("retriggers send-to-agent when session stays in same bad state past retriggerAfter", async () => {
config.reactions = {
"ci-failed": {
auto: true,
action: "send-to-agent",
message: "CI still failing. Try again.",
retries: 3,
// 30 seconds — will be fast-forwarded via fake timers
retriggerAfter: "30s",
},
};
const mockSCM: SCM = {
name: "mock-scm",
detectPR: vi.fn(),
getPRState: vi.fn().mockResolvedValue("open"),
mergePR: vi.fn(),
closePR: vi.fn(),
getCIChecks: vi.fn(),
getCISummary: vi.fn().mockResolvedValue("failing"),
getReviews: vi.fn(),
getReviewDecision: vi.fn().mockResolvedValue("none"),
getPendingComments: vi.fn().mockResolvedValue([]),
getAutomatedComments: vi.fn().mockResolvedValue([]),
getMergeability: vi.fn(),
};
const registryWithSCM: PluginRegistry = {
...mockRegistry,
get: vi.fn().mockImplementation((slot: string) => {
if (slot === "runtime") return mockRuntime;
if (slot === "agent") return mockAgent;
if (slot === "scm") return mockSCM;
return null;
}),
};
// Start in pr_open so first check transitions to ci_failed
const session = makeSession({ status: "pr_open", pr: makePR() });
vi.mocked(mockSessionManager.get).mockResolvedValue(session);
writeMetadata(sessionsDir, "app-1", {
worktree: "/tmp",
branch: "main",
status: "pr_open",
project: "my-app",
});
vi.useFakeTimers();
try {
const lm = createLifecycleManager({
config,
registry: registryWithSCM,
sessionManager: mockSessionManager,
});
// Check 1: pr_open → ci_failed (state transition) → initial reaction fires
await lm.check("app-1");
expect(mockSessionManager.send).toHaveBeenCalledTimes(1);
// Advance time past retriggerAfter (30s)
vi.advanceTimersByTime(31_000);
// Check 2: ci_failed → ci_failed (no transition) → retrigger fires
await lm.check("app-1");
expect(mockSessionManager.send).toHaveBeenCalledTimes(2);
} finally {
vi.useRealTimers();
}
});
it("does not retrigger when retriggerAfter is not configured", async () => {
config.reactions = {
"ci-failed": {
auto: true,
action: "send-to-agent",
message: "CI failing.",
retries: 3,
// No retriggerAfter
},
};
const mockSCM: SCM = {
name: "mock-scm",
detectPR: vi.fn(),
getPRState: vi.fn().mockResolvedValue("open"),
mergePR: vi.fn(),
closePR: vi.fn(),
getCIChecks: vi.fn(),
getCISummary: vi.fn().mockResolvedValue("failing"),
getReviews: vi.fn(),
getReviewDecision: vi.fn().mockResolvedValue("none"),
getPendingComments: vi.fn().mockResolvedValue([]),
getAutomatedComments: vi.fn().mockResolvedValue([]),
getMergeability: vi.fn(),
};
const registryWithSCM: PluginRegistry = {
...mockRegistry,
get: vi.fn().mockImplementation((slot: string) => {
if (slot === "runtime") return mockRuntime;
if (slot === "agent") return mockAgent;
if (slot === "scm") return mockSCM;
return null;
}),
};
const session = makeSession({ status: "pr_open", pr: makePR() });
vi.mocked(mockSessionManager.get).mockResolvedValue(session);
writeMetadata(sessionsDir, "app-1", {
worktree: "/tmp",
branch: "main",
status: "pr_open",
project: "my-app",
});
const lm = createLifecycleManager({
config,
registry: registryWithSCM,
sessionManager: mockSessionManager,
});
// Check 1: transition → initial reaction fires
await lm.check("app-1");
expect(mockSessionManager.send).toHaveBeenCalledTimes(1);
// Check 2: same state, no retriggerAfter → no extra send
await lm.check("app-1");
expect(mockSessionManager.send).toHaveBeenCalledTimes(1);
// Check 3: still no retrigger
await lm.check("app-1");
expect(mockSessionManager.send).toHaveBeenCalledTimes(1);
});
});

13
packages/core/src/config.ts
View File

@ -29,6 +29,8 @@ const ReactionConfigSchema = z.object({
priority: z.enum(["urgent", "action", "warning", "info"]).optional(),
retries: z.number().optional(),
escalateAfter: z.union([z.number(), z.string()]).optional(),
/** How long to wait between re-sends when session stays in a problematic state. */
retriggerAfter: z.string().optional(),
threshold: z.string().optional(),
includeSummary: z.boolean().optional(),
});
@ -221,6 +223,8 @@ function applyDefaultReactions(config: OrchestratorConfig): OrchestratorConfig {
"CI is failing on your PR. Run `gh pr checks` to see the failures, fix them, and push.",
retries: 2,
escalateAfter: 2,
// Re-send if CI is still failing after 15 minutes (e.g., agent pushed but CI is still red)
retriggerAfter: "15m",
},
"changes-requested": {
auto: true,
@ -228,18 +232,25 @@ function applyDefaultReactions(config: OrchestratorConfig): OrchestratorConfig {
message:
"There are review comments on your PR. Check with `gh pr view --comments` and `gh api` for inline comments. Address each one, push fixes, and reply.",
escalateAfter: "30m",
// Re-send if review comments are still unaddressed after 20 minutes
retriggerAfter: "20m",
},
"bugbot-comments": {
auto: true,
action: "send-to-agent",
message: "Automated review comments found on your PR. Fix the issues flagged by the bot.",
message:
"Automated review tool found issues on your PR. Run `gh pr view --comments` to see the bot feedback and fix the flagged issues.",
escalateAfter: "30m",
// Re-send if bot comments are still present (new comments appeared) — deduplication
// ensures this only fires when comment fingerprint changes, not every poll cycle.
retriggerAfter: "20m",
},
"merge-conflicts": {
auto: true,
action: "send-to-agent",
message: "Your branch has merge conflicts. Rebase on the default branch and resolve them.",
escalateAfter: "15m",
retriggerAfter: "10m",
},
"approved-and-green": {
auto: false,

63
packages/core/src/event-log.ts Normal file
View File

@ -0,0 +1,63 @@
/**
* Event log — append-only JSONL file recording all orchestrator events.
*
* Format: one JSON object per line (JSONL / newline-delimited JSON).
* Location: ~/.agent-orchestrator/{hash}-events.jsonl
*
* Designed to be:
* - Best-effort: logging errors never crash the orchestrator
* - Append-only: safe for concurrent writers (append is atomic on POSIX)
* - Human-readable: plain JSONL, greppable with jq
*/
import { appendFileSync, existsSync, readFileSync, mkdirSync } from "node:fs";
import { dirname } from "node:path";
import type { EventLog, OrchestratorEvent } from "./types.js";
/** Create an event log that appends events to a JSONL file at `logPath`. */
export function createEventLog(logPath: string): EventLog {
return {
log(event: OrchestratorEvent): void {
try {
mkdirSync(dirname(logPath), { recursive: true });
const entry = JSON.stringify({
...event,
timestamp: event.timestamp.toISOString(),
});
appendFileSync(logPath, entry + "\n", "utf-8");
} catch {
// Event logging is best-effort — never crash the orchestrator
}
},
readRecent(limit = 100): OrchestratorEvent[] {
if (!existsSync(logPath)) return [];
try {
const content = readFileSync(logPath, "utf-8");
const lines = content.trim().split("\n").filter(Boolean);
const slice = limit > 0 ? lines.slice(-limit) : lines;
return slice.flatMap((line) => {
try {
const parsed = JSON.parse(line) as unknown;
const entry = parsed as OrchestratorEvent & { timestamp: string };
return [{ ...entry, timestamp: new Date(entry.timestamp) }];
} catch {
return [];
}
});
} catch {
return [];
}
},
};
}
/** Create a no-op event log (when logging is disabled or in tests). */
export function createNullEventLog(): EventLog {
return {
log(): void {},
readRecent(): OrchestratorEvent[] {
return [];
},
};
}

4
packages/core/src/index.ts
View File

@ -51,6 +51,9 @@ export type { SessionManagerDeps } from "./session-manager.js";
export { createLifecycleManager } from "./lifecycle-manager.js";
export type { LifecycleManagerDeps } from "./lifecycle-manager.js";
// Event log — append-only JSONL audit trail
export { createEventLog, createNullEventLog } from "./event-log.js";
// Prompt builder — layered prompt composition
export { buildPrompt, BASE_AGENT_PROMPT } from "./prompt-builder.js";
export type { PromptBuildConfig } from "./prompt-builder.js";
@ -73,6 +76,7 @@ export {
getWorktreesDir,
getArchiveDir,
getOriginFilePath,
getEventLogPath,
generateSessionName,
generateTmuxName,
parseTmuxName,

242
packages/core/src/lifecycle-manager.ts
View File

@ -3,9 +3,21 @@
*
* Periodically polls all sessions and:
* 1. Detects state transitions (spawning → working → pr_open → etc.)
* 2. Emits events on transitions
* 3. Triggers reactions (auto-handle CI failures, review comments, etc.)
* 4. Escalates to human notification when auto-handling fails
* 2. Emits events on transitions (logged to JSONL event log)
* 3. Triggers reactions (auto-handle CI failures, review comments, bugbot, etc.)
* 4. Re-triggers reactions for persistent conditions (CI still failing, etc.)
* 5. Escalates to human notification when auto-handling fails
*
* Event sources polled each cycle:
* - Runtime liveness (is the agent process still running?)
* - Agent activity state (active / ready / idle / stuck / waiting_input)
* - PR state (open / merged / closed)
* - CI checks summary (passing / failing / pending)
* - Review decision (approved / changes_requested / pending)
* - Automated review comments / bugbot comments (getAutomatedComments)
*
* Note: Human review comment polling (getPendingComments) is handled by the
* review-comments reaction (ao-48). See: eventToReactionKey for "changes-requested".
*
* Reference: scripts/claude-session-status, scripts/claude-review-check
*/
@ -31,10 +43,12 @@ import {
type Notifier,
type Session,
type EventPriority,
type EventLog,
type ProjectConfig as _ProjectConfig,
} from "./types.js";
import { updateMetadata } from "./metadata.js";
import { getSessionsDir } from "./paths.js";
import { createNullEventLog } from "./event-log.js";
/** Parse a duration string like "10m", "30s", "1h" to milliseconds. */
function parseDuration(str: string): number {
@ -156,21 +170,37 @@ function eventToReactionKey(eventType: EventType): string | null {
}
}
/**
* States where we should re-trigger reactions if the session stays stuck.
* These are "actionable" states — the agent should fix them.
*/
const RETRIGGER_STATES: ReadonlySet<SessionStatus> = new Set([
"ci_failed",
"changes_requested",
"stuck",
"needs_input",
]);
export interface LifecycleManagerDeps {
config: OrchestratorConfig;
registry: PluginRegistry;
sessionManager: SessionManager;
/** Optional event log for auditing all emitted events. Defaults to no-op. */
eventLog?: EventLog;
}
/** Track attempt counts for reactions per session. */
/** Track attempt counts and timing for reactions per session. */
interface ReactionTracker {
attempts: number;
firstTriggered: Date;
/** When the reaction was last attempted (for retrigger cooldown). */
lastAttemptAt: Date;
}
/** Create a LifecycleManager instance. */
export function createLifecycleManager(deps: LifecycleManagerDeps): LifecycleManager {
const { config, registry, sessionManager } = deps;
const eventLog = deps.eventLog ?? createNullEventLog();
const states = new Map<SessionId, SessionStatus>();
const reactionTrackers = new Map<string, ReactionTracker>(); // "sessionId:reactionKey"
@ -178,6 +208,13 @@ export function createLifecycleManager(deps: LifecycleManagerDeps): LifecycleMan
let polling = false; // re-entrancy guard
let allCompleteEmitted = false; // guard against repeated all_complete
/**
* Fingerprints of automated (bot) comments seen per session.
* Key: sessionId, Value: sorted comma-joined comment IDs.
* Prevents re-triggering the bugbot-comments reaction for the same set of comments.
*/
const automatedCommentFingerprints = new Map<SessionId, string>();
/** Determine current status for a session by polling plugins. */
async function determineStatus(session: Session): Promise<SessionStatus> {
const project = config.projects[session.projectId];
@ -268,6 +305,22 @@ export function createLifecycleManager(deps: LifecycleManagerDeps): LifecycleMan
return session.status;
}
/**
* Get the effective reaction config for a session, merging project overrides
* with global defaults (project wins on any key that is set).
*/
function getEffectiveReactionConfig(
session: Session,
reactionKey: string,
): ReactionConfig | null {
const project = config.projects[session.projectId];
const globalReaction = config.reactions[reactionKey];
const projectReaction = project?.reactions?.[reactionKey];
if (!globalReaction && !projectReaction) return null;
if (projectReaction) return { ...globalReaction, ...projectReaction } as ReactionConfig;
return globalReaction ?? null;
}
/** Execute a reaction for a session. */
async function executeReaction(
sessionId: SessionId,
@ -279,12 +332,14 @@ export function createLifecycleManager(deps: LifecycleManagerDeps): LifecycleMan
let tracker = reactionTrackers.get(trackerKey);
if (!tracker) {
tracker = { attempts: 0, firstTriggered: new Date() };
const now = new Date();
tracker = { attempts: 0, firstTriggered: now, lastAttemptAt: now };
reactionTrackers.set(trackerKey, tracker);
}
// Increment attempts before checking escalation
tracker.attempts++;
tracker.lastAttemptAt = new Date();
// Check if we should escalate
const maxRetries = reactionConfig.retries ?? Infinity;
@ -314,6 +369,7 @@ export function createLifecycleManager(deps: LifecycleManagerDeps): LifecycleMan
message: `Reaction '${reactionKey}' escalated after ${tracker.attempts} attempts`,
data: { reactionKey, attempts: tracker.attempts },
});
eventLog.log(event);
await notifyHuman(event, reactionConfig.priority ?? "urgent");
return {
reactionType: reactionKey,
@ -331,7 +387,13 @@ export function createLifecycleManager(deps: LifecycleManagerDeps): LifecycleMan
if (reactionConfig.message) {
try {
await sessionManager.send(sessionId, reactionConfig.message);
const event = createEvent("reaction.triggered", {
sessionId,
projectId,
message: `Reaction '${reactionKey}' sent message to agent (attempt ${tracker.attempts})`,
data: { reactionKey, attempts: tracker.attempts },
});
eventLog.log(event);
return {
reactionType: reactionKey,
success: true,
@ -359,6 +421,7 @@ export function createLifecycleManager(deps: LifecycleManagerDeps): LifecycleMan
message: `Reaction '${reactionKey}' triggered notification`,
data: { reactionKey },
});
eventLog.log(event);
await notifyHuman(event, reactionConfig.priority ?? "info");
return {
reactionType: reactionKey,
@ -377,6 +440,7 @@ export function createLifecycleManager(deps: LifecycleManagerDeps): LifecycleMan
message: `Reaction '${reactionKey}' triggered auto-merge`,
data: { reactionKey },
});
eventLog.log(event);
await notifyHuman(event, "action");
return {
reactionType: reactionKey,
@ -412,6 +476,121 @@ export function createLifecycleManager(deps: LifecycleManagerDeps): LifecycleMan
}
}
/**
* Check for automated (bot) review comments on a PR and trigger the
* bugbot-comments reaction when new comments appear.
*
* Deduplication: tracks a fingerprint (sorted comment IDs) per session.
* The reaction only fires when new comments appear that we haven't seen before.
* When the comment set changes (e.g., after agent fixes some), we retrigger
* with the new set.
*/
async function checkAutomatedComments(session: Session): Promise<void> {
if (!session.pr) return;
const project = config.projects[session.projectId];
if (!project?.scm) return;
const scm = registry.get<SCM>("scm", project.scm.plugin);
if (!scm) return;
let comments;
try {
comments = await scm.getAutomatedComments(session.pr);
} catch {
return; // SCM error — skip this cycle
}
if (comments.length === 0) {
// No automated comments — clear any stored fingerprint so future comments retrigger
automatedCommentFingerprints.delete(session.id);
return;
}
// Compute fingerprint from sorted comment IDs
const fingerprint = comments
.map((c) => c.id)
.sort()
.join(",");
const lastFingerprint = automatedCommentFingerprints.get(session.id);
if (fingerprint === lastFingerprint) {
return; // Same comments as before — already reacted, skip
}
// New or changed automated comments — update fingerprint and trigger reaction
automatedCommentFingerprints.set(session.id, fingerprint);
const event = createEvent("automated_review.found", {
sessionId: session.id,
projectId: session.projectId,
message: `${comments.length} automated review comment(s) on PR #${session.pr.number}`,
priority: "warning",
data: {
prNumber: session.pr.number,
count: comments.length,
comments: comments.map((c) => ({
id: c.id,
botName: c.botName,
severity: c.severity,
path: c.path,
url: c.url,
})),
},
});
eventLog.log(event);
const reactionKey = "bugbot-comments";
const reactionConfig = getEffectiveReactionConfig(session, reactionKey);
if (reactionConfig && reactionConfig.action && reactionConfig.auto !== false) {
await executeReaction(session.id, session.projectId, reactionKey, reactionConfig);
} else if (!reactionConfig || reactionConfig.action === "notify") {
// No configured reaction or notify-only: surface to human
await notifyHuman(event, event.priority);
}
}
/**
* Re-trigger reactions for sessions that stay in a problematic state without
* transitioning. This handles the case where an agent received the initial
* "CI is failing" message but didn't fix it — we want to retry after a delay
* rather than silently giving up.
*
* Only fires when:
* 1. The session is in a RETRIGGER_STATE (ci_failed, changes_requested, etc.)
* 2. A reaction was previously triggered for this state (tracker exists)
* 3. Enough time has passed since the last attempt (retriggerAfter cooldown)
* 4. The reaction is configured with retriggerAfter (opt-in)
*/
async function checkPersistentConditions(
session: Session,
status: SessionStatus,
): Promise<void> {
if (!RETRIGGER_STATES.has(status)) return;
const eventType = statusToEventType(undefined, status);
if (!eventType) return;
const reactionKey = eventToReactionKey(eventType);
if (!reactionKey) return;
const reactionConfig = getEffectiveReactionConfig(session, reactionKey);
if (!reactionConfig?.retriggerAfter) return; // Opt-in only
if (reactionConfig.auto === false) return;
if (reactionConfig.action !== "send-to-agent") return; // Only retrigger agent sends
const trackerKey = `${session.id}:${reactionKey}`;
const tracker = reactionTrackers.get(trackerKey);
if (!tracker) return; // Never triggered before — initial trigger happens on state transition
const retriggerMs = parseDuration(reactionConfig.retriggerAfter);
if (retriggerMs <= 0) return;
const timeSinceLastAttempt = Date.now() - tracker.lastAttemptAt.getTime();
if (timeSinceLastAttempt < retriggerMs) return; // Too soon
// Re-trigger: agent is still stuck in the same bad state
await executeReaction(session.id, session.projectId, reactionKey, reactionConfig);
}
/** Poll a single session and handle state transitions. */
async function checkSession(session: Session): Promise<void> {
// Use tracked state if available; otherwise use the persisted metadata status
@ -438,7 +617,8 @@ export function createLifecycleManager(deps: LifecycleManagerDeps): LifecycleMan
allCompleteEmitted = false;
}
// Clear reaction trackers for the old status so retries reset on state changes
// Clear reaction trackers for the old status so retries reset on state changes.
// Also clear automated comment fingerprints so new comments retrigger after a fix.
const oldEventType = statusToEventType(undefined, oldStatus);
if (oldEventType) {
const oldReactionKey = eventToReactionKey(oldEventType);
@ -454,13 +634,7 @@ export function createLifecycleManager(deps: LifecycleManagerDeps): LifecycleMan
const reactionKey = eventToReactionKey(eventType);
if (reactionKey) {
// Merge project-specific overrides with global defaults
const project = config.projects[session.projectId];
const globalReaction = config.reactions[reactionKey];
const projectReaction = project?.reactions?.[reactionKey];
const reactionConfig = projectReaction
? { ...globalReaction, ...projectReaction }
: globalReaction;
const reactionConfig = getEffectiveReactionConfig(session, reactionKey);
if (reactionConfig && reactionConfig.action) {
// auto: false skips automated agent actions but still allows notifications
@ -469,7 +643,7 @@ export function createLifecycleManager(deps: LifecycleManagerDeps): LifecycleMan
session.id,
session.projectId,
reactionKey,
reactionConfig as ReactionConfig,
reactionConfig,
);
// Reaction is handling this event — suppress immediate human notification.
// "send-to-agent" retries + escalates on its own; "notify"/"auto-merge"
@ -480,23 +654,34 @@ export function createLifecycleManager(deps: LifecycleManagerDeps): LifecycleMan
}
}
// Emit event to log
const event = createEvent(eventType, {
sessionId: session.id,
projectId: session.projectId,
message: `${session.id}: ${oldStatus} → ${newStatus}`,
data: { oldStatus, newStatus },
});
eventLog.log(event);
// For significant transitions not already notified by a reaction, notify humans
if (!reactionHandledNotify) {
const priority = inferPriority(eventType);
if (priority !== "info") {
const event = createEvent(eventType, {
sessionId: session.id,
projectId: session.projectId,
message: `${session.id}: ${oldStatus} → ${newStatus}`,
data: { oldStatus, newStatus },
});
await notifyHuman(event, priority);
}
}
}
} else {
// No transition but track current state
// No status transition — track current state
states.set(session.id, newStatus);
// Check for automated review comments (bot/linter feedback) each poll cycle.
// Deduplication is handled inside checkAutomatedComments via fingerprinting.
await checkAutomatedComments(session);
// Check if we should re-trigger a reaction for persistent problematic conditions
// (e.g., CI has been failing for 10 minutes and agent hasn't fixed it yet).
await checkPersistentConditions(session, newStatus);
}
}
@ -535,6 +720,11 @@ export function createLifecycleManager(deps: LifecycleManagerDeps): LifecycleMan
reactionTrackers.delete(trackerKey);
}
}
for (const sessionId of automatedCommentFingerprints.keys()) {
if (!currentSessionIds.has(sessionId)) {
automatedCommentFingerprints.delete(sessionId);
}
}
// Check if all sessions are complete (trigger reaction only once)
const activeSessions = sessions.filter((s) => s.status !== "merged" && s.status !== "killed");
@ -546,6 +736,14 @@ export function createLifecycleManager(deps: LifecycleManagerDeps): LifecycleMan
if (reactionKey) {
const reactionConfig = config.reactions[reactionKey];
if (reactionConfig && reactionConfig.action) {
const event = createEvent("summary.all_complete", {
sessionId: "system",
projectId: "all",
message: `All ${sessions.length} session(s) complete`,
priority: "info",
data: { totalSessions: sessions.length },
});
eventLog.log(event);
if (reactionConfig.auto !== false || reactionConfig.action === "notify") {
await executeReaction("system", "all", reactionKey, reactionConfig as ReactionConfig);
}

12
packages/core/src/paths.ts
View File

@ -156,6 +156,18 @@ export function parseTmuxName(tmuxName: string): {
};
}
/**
* Get the path to the JSONL event log for a given config.
* Format: ~/.agent-orchestrator/{hash}-events.jsonl
*
* All events from all projects under this config are written to one log,
* making it easy to audit the full orchestrator history.
*/
export function getEventLogPath(configPath: string): string {
const hash = generateConfigHash(configPath);
return join(expandHome("~/.agent-orchestrator"), `${hash}-events.jsonl`);
}
/**
* Expand ~ to home directory.
*/

23
packages/core/src/types.ts
View File

@ -745,6 +745,19 @@ export interface OrchestratorEvent {
data: Record<string, unknown>;
}
// =============================================================================
// EVENT LOG
// =============================================================================
/** Append-only JSONL event log that records all orchestrator events to disk. */
export interface EventLog {
/** Append an event to the log. Best-effort — errors are swallowed. */
log(event: OrchestratorEvent): void;
/** Read the last N events from the log. Returns [] if the log doesn't exist. */
readRecent(limit?: number): OrchestratorEvent[];
}
// =============================================================================
// REACTIONS
// =============================================================================
@ -769,6 +782,16 @@ export interface ReactionConfig {
/** Escalate to human notification after this many failures or this duration */
escalateAfter?: number | string;
/**
* Re-trigger the reaction when session stays in the same problematic state.
* Duration string ("10m", "30s", "1h") — how long to wait between re-sends.
* Only applies to "send-to-agent" actions. Without this, reactions only fire
* on state transitions; with this, they also fire periodically while stuck.
*
* Example: retriggerAfter: "10m" — re-send every 10 minutes if CI still failing.
*/
retriggerAfter?: string;
/** Threshold duration for time-based triggers (e.g. "10m" for stuck detection) */
threshold?: string;