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

fix: address PR review — unsafe casts, tmux race, concurrent polling 

- session-manager: replace readMetadata with readMetadataRaw (no unsafe casts),
  add safeJsonParse for JSON.parse guards, escapeRegex for session prefix,
  explicit branch priority, convert from this-based to closure-based methods
- tmux: named buffers (load-buffer -b / paste-buffer -b -d) to prevent
  global paste buffer race on concurrent sendKeys; send Escape before text
  to clear partial input
- lifecycle-manager: concurrent polling with Promise.allSettled instead of
  sequential for-of await, allCompleteEmitted guard to prevent repeated
  summary.all_complete events, polling re-entrancy guard, safe reaction
  config handling with action/auto checks
- tests: update tmux tests for Escape-first and named buffer changes

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Prateek 2026-02-13 17:55:29 +05:30
parent 2b9a28a031
commit d6ec95402d
4 changed files with 378 additions and 321 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

79
packages/core/src/__tests__/tmux.test.ts
View File

@ -159,48 +159,63 @@ describe("newSession", () => {
});
it("sends initial command after creation", async () => {
// First call: new-session, Second call: send-keys, Third call: send-keys Enter
// Calls: new-session, send-keys Escape, send-keys text, send-keys Enter
mockTmuxSequence([
{ stdout: "" },
{ stdout: "" },
{ stdout: "" },
{ stdout: "" },
]);
await newSession({ name: "test-4", cwd: "/tmp", command: "echo hello" });
// Should have called new-session then send-keys twice (text + Enter)
expect(mockExecFile).toHaveBeenCalledTimes(3);
const secondCallArgs = mockExecFile.mock.calls[1][1] as string[];
expect(secondCallArgs).toContain("send-keys");
expect(secondCallArgs).toContain("echo hello");
expect(mockExecFile).toHaveBeenCalledTimes(4);
// Call 0: new-session
// Call 1: send-keys Escape (clear partial input)
const escapeArgs = mockExecFile.mock.calls[1][1] as string[];
expect(escapeArgs).toEqual(["send-keys", "-t", "test-4", "Escape"]);
// Call 2: send-keys text
const textArgs = mockExecFile.mock.calls[2][1] as string[];
expect(textArgs).toContain("send-keys");
expect(textArgs).toContain("echo hello");
});
});
describe("sendKeys", () => {
it("sends short text with send-keys", async () => {
mockTmuxSequence([{ stdout: "" }, { stdout: "" }]);
// Calls: send-keys Escape, send-keys text, send-keys Enter
mockTmuxSequence([{ stdout: "" }, { stdout: "" }, { stdout: "" }]);
await sendKeys("app-1", "hello world");
expect(mockExecFile).toHaveBeenCalledTimes(2);
const firstArgs = mockExecFile.mock.calls[0][1] as string[];
expect(firstArgs).toEqual(["send-keys", "-t", "app-1", "hello world"]);
// Second call is Enter
const secondArgs = mockExecFile.mock.calls[1][1] as string[];
expect(secondArgs).toEqual(["send-keys", "-t", "app-1", "Enter"]);
expect(mockExecFile).toHaveBeenCalledTimes(3);
// Call 0: Escape to clear partial input
const escapeArgs = mockExecFile.mock.calls[0][1] as string[];
expect(escapeArgs).toEqual(["send-keys", "-t", "app-1", "Escape"]);
// Call 1: text
const textArgs = mockExecFile.mock.calls[1][1] as string[];
expect(textArgs).toEqual(["send-keys", "-t", "app-1", "hello world"]);
// Call 2: Enter
const enterArgs = mockExecFile.mock.calls[2][1] as string[];
expect(enterArgs).toEqual(["send-keys", "-t", "app-1", "Enter"]);
});
it("skips Enter when pressEnter=false", async () => {
mockTmuxSuccess("");
// Calls: send-keys Escape, send-keys text (no Enter)
mockTmuxSequence([{ stdout: "" }, { stdout: "" }]);
await sendKeys("app-1", "hello", false);
expect(mockExecFile).toHaveBeenCalledTimes(1);
expect(mockExecFile).toHaveBeenCalledTimes(2);
const escapeArgs = mockExecFile.mock.calls[0][1] as string[];
expect(escapeArgs).toEqual(["send-keys", "-t", "app-1", "Escape"]);
});
it("uses load-buffer for long text", async () => {
it("uses load-buffer with named buffer for long text", async () => {
const longText = "a".repeat(250);
// Calls: send-keys Escape, load-buffer -b name, paste-buffer -b name -d, send-keys Enter
mockTmuxSequence([
{ stdout: "" }, // send-keys Escape
{ stdout: "" }, // load-buffer
{ stdout: "" }, // paste-buffer
{ stdout: "" }, // send-keys Enter
@ -208,15 +223,32 @@ describe("sendKeys", () => {
await sendKeys("app-1", longText);
const firstArgs = mockExecFile.mock.calls[0][1] as string[];
expect(firstArgs[0]).toBe("load-buffer");
expect(mockExecFile).toHaveBeenCalledTimes(4);
const secondArgs = mockExecFile.mock.calls[1][1] as string[];
expect(secondArgs).toEqual(["paste-buffer", "-t", "app-1"]);
// Call 0: Escape
const escapeArgs = mockExecFile.mock.calls[0][1] as string[];
expect(escapeArgs).toEqual(["send-keys", "-t", "app-1", "Escape"]);
// Call 1: load-buffer with named buffer
const loadArgs = mockExecFile.mock.calls[1][1] as string[];
expect(loadArgs[0]).toBe("load-buffer");
expect(loadArgs[1]).toBe("-b");
expect(loadArgs[2]).toMatch(/^ao-/); // named buffer
// Call 2: paste-buffer with named buffer and -d (delete after paste)
const pasteArgs = mockExecFile.mock.calls[2][1] as string[];
expect(pasteArgs[0]).toBe("paste-buffer");
expect(pasteArgs[1]).toBe("-b");
expect(pasteArgs[2]).toMatch(/^ao-/);
expect(pasteArgs).toContain("-d");
expect(pasteArgs).toContain("-t");
expect(pasteArgs).toContain("app-1");
});
it("uses load-buffer for multiline text", async () => {
// Calls: send-keys Escape, load-buffer, paste-buffer, send-keys Enter
mockTmuxSequence([
{ stdout: "" }, // send-keys Escape
{ stdout: "" }, // load-buffer
{ stdout: "" }, // paste-buffer
{ stdout: "" }, // send-keys Enter
@ -224,8 +256,11 @@ describe("sendKeys", () => {
await sendKeys("app-1", "line1\nline2");
const firstArgs = mockExecFile.mock.calls[0][1] as string[];
expect(firstArgs[0]).toBe("load-buffer");
expect(mockExecFile).toHaveBeenCalledTimes(4);
// Call 1 (after Escape) should be load-buffer
const loadArgs = mockExecFile.mock.calls[1][1] as string[];
expect(loadArgs[0]).toBe("load-buffer");
expect(loadArgs[1]).toBe("-b"); // named buffer
});
});

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

@ -19,10 +19,10 @@ import type {
EventBus,
OrchestratorEvent,
OrchestratorConfig,
ProjectConfig,
ReactionConfig,
ReactionResult,
PluginRegistry,
Runtime,
Agent,
SCM,
Notifier,
@ -47,25 +47,9 @@ function parseDuration(str: string): number {
}
}
/** Map from event type prefix to session status. */
const STATUS_FROM_EVENT: Partial<Record<EventType, SessionStatus>> = {
"session.spawned": "spawning",
"session.working": "working",
"session.stuck": "stuck",
"session.needs_input": "needs_input",
"session.errored": "errored",
"session.killed": "killed",
"pr.created": "pr_open",
"ci.failing": "ci_failed",
"review.changes_requested": "changes_requested",
"review.approved": "approved",
"merge.ready": "mergeable",
"merge.completed": "merged",
};
/** Determine which event type corresponds to a status transition. */
function statusToEventType(
from: SessionStatus | undefined,
_from: SessionStatus | undefined,
to: SessionStatus
): EventType | null {
switch (to) {
@ -124,6 +108,8 @@ export function createLifecycleManager(
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);
@ -164,7 +150,7 @@ export function createLifecycleManager(
// 1. Check if runtime is alive
if (session.runtimeHandle) {
const runtime = registry.get<import("./types.js").Runtime>(
const runtime = registry.get<Runtime>(
"runtime",
project.runtime ?? config.defaults.runtime
);
@ -260,7 +246,9 @@ export function createLifecycleManager(
}
// Execute the reaction action
switch (reactionConfig.action) {
const action = reactionConfig.action ?? "notify";
switch (action) {
case "send-to-agent": {
if (reactionConfig.message) {
try {
@ -323,7 +311,7 @@ export function createLifecycleManager(
return {
reactionType: reactionKey,
success: false,
action: reactionConfig.action,
action,
escalated: false,
};
}
@ -360,6 +348,11 @@ export function createLifecycleManager(
// Update metadata
updateMetadata(config.dataDir, session.id, { status: newStatus });
// Reset allCompleteEmitted when any session becomes active again
if (newStatus !== "merged" && newStatus !== "killed") {
allCompleteEmitted = false;
}
// Emit transition event
const eventType = statusToEventType(oldStatus, newStatus);
if (eventType) {
@ -382,7 +375,7 @@ export function createLifecycleManager(
project?.reactions?.[reactionKey] ??
config.reactions[reactionKey];
if (reactionConfig && (reactionConfig as ReactionConfig).auto !== false) {
if (reactionConfig && reactionConfig.auto !== false && reactionConfig.action) {
await executeReaction(
event,
reactionKey,
@ -399,19 +392,25 @@ export function createLifecycleManager(
/** Run one polling cycle across all sessions. */
async function pollAll(): Promise<void> {
// Re-entrancy guard: skip if previous poll is still running
if (polling) return;
polling = true;
try {
const sessions = await sessionManager.list();
for (const session of sessions) {
if (session.status === "merged" || session.status === "killed") continue;
await checkSession(session);
}
// Check if all sessions are complete
const activeSessions = sessions.filter(
(s) => s.status !== "merged" && s.status !== "killed"
);
if (sessions.length > 0 && activeSessions.length === 0) {
// Poll all active sessions concurrently
await Promise.allSettled(
activeSessions.map((s) => checkSession(s))
);
// Check if all sessions are complete (emit only once)
if (sessions.length > 0 && activeSessions.length === 0 && !allCompleteEmitted) {
allCompleteEmitted = true;
const event = createEvent("summary.all_complete", {
sessionId: "system",
projectId: "all",
@ -423,6 +422,8 @@ export function createLifecycleManager(
}
} catch {
// Poll cycle failed — will retry next interval
} finally {
polling = false;
}
}

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

@ -11,7 +11,6 @@
* Reference: scripts/claude-ao-session, scripts/send-to-session
*/
import { execFile } from "node:child_process";
import type {
SessionManager,
Session,
@ -28,23 +27,19 @@ import type {
SCM,
EventBus,
PluginRegistry,
RuntimeHandle,
} from "./types.js";
import {
readMetadata,
readMetadataRaw,
writeMetadata,
updateMetadata,
deleteMetadata,
listMetadata,
} from "./metadata.js";
import { createEvent } from "./event-bus.js";
/** Get the current git branch in a directory. */
function getGitBranch(dir: string): Promise<string> {
return new Promise((resolve) => {
execFile("git", ["-C", dir, "branch", "--show-current"], (error, stdout) => {
resolve(error ? "unknown" : stdout.trim());
});
});
/** Escape regex metacharacters in a string. */
function escapeRegex(str: string): string {
return str.replace(/[.*+?^${}()|[\]\\]/g, "\\$&");
}
/** Get the next session number for a project. */
@ -53,7 +48,7 @@ function getNextSessionNumber(
prefix: string
): number {
let max = 0;
const pattern = new RegExp(`^${prefix}-(\\d+)$`);
const pattern = new RegExp(`^${escapeRegex(prefix)}-(\\d+)$`);
for (const name of existingSessions) {
const match = name.match(pattern);
if (match) {
@ -64,7 +59,16 @@ function getNextSessionNumber(
return max + 1;
}
/** Reconstruct a Session object from metadata + live data. */
/** Safely parse JSON, returning null on failure. */
function safeJsonParse<T>(str: string): T | null {
try {
return JSON.parse(str) as T;
} catch {
return null;
}
}
/** Reconstruct a Session object from raw metadata key=value pairs. */
function metadataToSession(
sessionId: SessionId,
meta: Record<string, string>
@ -90,7 +94,7 @@ function metadataToSession(
: null,
workspacePath: meta["worktree"] || null,
runtimeHandle: meta["runtimeHandle"]
? JSON.parse(meta["runtimeHandle"])
? safeJsonParse<RuntimeHandle>(meta["runtimeHandle"])
: null,
agentInfo: meta["summary"]
? { summary: meta["summary"], agentSessionId: null }
@ -135,8 +139,8 @@ export function createSessionManager(deps: SessionManagerDeps): SessionManager {
return { runtime, agent, workspace, tracker, scm };
}
return {
async spawn(spawnConfig: SessionSpawnConfig): Promise<Session> {
// Define methods as local functions so `this` is not needed
async function spawn(spawnConfig: SessionSpawnConfig): Promise<Session> {
const project = config.projects[spawnConfig.projectId];
if (!project) {
throw new Error(`Unknown project: ${spawnConfig.projectId}`);
@ -155,12 +159,16 @@ export function createSessionManager(deps: SessionManagerDeps): SessionManager {
const num = getNextSessionNumber(existingSessions, project.sessionPrefix);
const sessionId = `${project.sessionPrefix}-${num}`;
// Determine branch name
let branch = spawnConfig.branch ?? project.defaultBranch;
if (spawnConfig.issueId && plugins.tracker) {
// Determine branch name — explicit branch always takes priority
let branch: string;
if (spawnConfig.branch) {
branch = spawnConfig.branch;
} else if (spawnConfig.issueId && plugins.tracker) {
branch = plugins.tracker.branchName(spawnConfig.issueId, project);
} else if (spawnConfig.issueId) {
branch = `feat/${spawnConfig.issueId}`;
} else {
branch = project.defaultBranch;
}
// Create workspace (if workspace plugin is available)
@ -247,20 +255,20 @@ export function createSessionManager(deps: SessionManagerDeps): SessionManager {
);
return session;
},
}
async list(projectId?: string): Promise<Session[]> {
async function list(projectId?: string): Promise<Session[]> {
const sessionIds = listMetadata(config.dataDir);
const sessions: Session[] = [];
for (const sid of sessionIds) {
const raw = readMetadata(config.dataDir, sid);
const raw = readMetadataRaw(config.dataDir, sid);
if (!raw) continue;
// Filter by project if specified
if (projectId && raw.project !== projectId) continue;
if (projectId && raw["project"] !== projectId) continue;
const session = metadataToSession(sid, raw as unknown as Record<string, string>);
const session = metadataToSession(sid, raw);
// Check if runtime is still alive
if (session.runtimeHandle) {
@ -285,40 +293,42 @@ export function createSessionManager(deps: SessionManagerDeps): SessionManager {
}
return sessions;
},
}
async get(sessionId: SessionId): Promise<Session | null> {
const raw = readMetadata(config.dataDir, sessionId);
async function get(sessionId: SessionId): Promise<Session | null> {
const raw = readMetadataRaw(config.dataDir, sessionId);
if (!raw) return null;
return metadataToSession(sessionId, raw as unknown as Record<string, string>);
},
return metadataToSession(sessionId, raw);
}
async kill(sessionId: SessionId): Promise<void> {
const raw = readMetadata(config.dataDir, sessionId);
async function kill(sessionId: SessionId): Promise<void> {
const raw = readMetadataRaw(config.dataDir, sessionId);
if (!raw) throw new Error(`Session ${sessionId} not found`);
const projectId = raw.project ?? "";
const projectId = raw["project"] ?? "";
const project = config.projects[projectId];
// Destroy runtime
if (raw.runtimeHandle && project) {
if (raw["runtimeHandle"] && project) {
const plugins = resolvePlugins(project);
if (plugins.runtime) {
const handle = safeJsonParse<RuntimeHandle>(raw["runtimeHandle"]);
if (handle) {
try {
const handle = JSON.parse(raw.runtimeHandle);
await plugins.runtime.destroy(handle);
} catch {
// Runtime might already be gone
}
}
}
}
// Destroy workspace
if (raw.worktree && project) {
if (raw["worktree"] && project) {
const plugins = resolvePlugins(project);
if (plugins.workspace) {
try {
await plugins.workspace.destroy(raw.worktree);
await plugins.workspace.destroy(raw["worktree"]);
} catch {
// Workspace might already be gone
}
@ -336,11 +346,11 @@ export function createSessionManager(deps: SessionManagerDeps): SessionManager {
message: `Session ${sessionId} killed`,
})
);
},
}
async cleanup(projectId?: string): Promise<CleanupResult> {
async function cleanup(projectId?: string): Promise<CleanupResult> {
const result: CleanupResult = { killed: [], skipped: [], errors: [] };
const sessions = await this.list(projectId);
const sessions = await list(projectId);
for (const session of sessions) {
try {
@ -389,7 +399,7 @@ export function createSessionManager(deps: SessionManagerDeps): SessionManager {
}
if (shouldKill) {
await this.kill(session.id);
await kill(session.id);
result.killed.push(session.id);
} else {
result.skipped.push(session.id);
@ -403,13 +413,13 @@ export function createSessionManager(deps: SessionManagerDeps): SessionManager {
}
return result;
},
}
async send(sessionId: SessionId, message: string): Promise<void> {
const raw = readMetadata(config.dataDir, sessionId);
async function send(sessionId: SessionId, message: string): Promise<void> {
const raw = readMetadataRaw(config.dataDir, sessionId);
if (!raw) throw new Error(`Session ${sessionId} not found`);
const project = config.projects[raw.project ?? ""];
const project = config.projects[raw["project"] ?? ""];
if (!project) throw new Error(`Project not found for session ${sessionId}`);
const plugins = resolvePlugins(project);
@ -417,12 +427,17 @@ export function createSessionManager(deps: SessionManagerDeps): SessionManager {
throw new Error(`No runtime plugin for session ${sessionId}`);
}
if (!raw.runtimeHandle) {
if (!raw["runtimeHandle"]) {
throw new Error(`No runtime handle for session ${sessionId}`);
}
const handle = JSON.parse(raw.runtimeHandle);
const handle = safeJsonParse<RuntimeHandle>(raw["runtimeHandle"]);
if (!handle) {
throw new Error(`Corrupted runtime handle for session ${sessionId}`);
}
await plugins.runtime.sendMessage(handle, message);
},
};
}
return { spawn, list, get, kill, cleanup, send };
}

16
packages/core/src/tmux.ts
View File

@ -118,26 +118,32 @@ export async function newSession(opts: NewSessionOptions): Promise<void> {
/**
* Send keys (text + Enter) to a tmux session.
* For long/multiline messages, uses load-buffer + paste-buffer.
* For long/multiline messages, uses load-buffer + paste-buffer with
* a named buffer to avoid racing on the global paste buffer.
* Sends Escape first to clear any partial input in the agent.
*/
export async function sendKeys(
sessionName: string,
text: string,
pressEnter = true
): Promise<void> {
// Clear any partial input first (matches bash reference scripts)
await tmux("send-keys", "-t", sessionName, "Escape");
if (text.includes("\n") || text.length > 200) {
// Use load-buffer for long/multiline text (avoids tmux escaping issues)
// Use a named buffer to avoid global paste buffer race conditions
const { writeFileSync, unlinkSync } = await import("node:fs");
const { tmpdir } = await import("node:os");
const { join } = await import("node:path");
const { randomUUID } = await import("node:crypto");
const tmpFile = join(tmpdir(), `ao-tmux-${randomUUID()}.txt`);
const bufferName = `ao-${randomUUID().slice(0, 8)}`;
const tmpFile = join(tmpdir(), `ao-tmux-${bufferName}.txt`);
writeFileSync(tmpFile, text, "utf-8");
try {
await tmux("load-buffer", tmpFile);
await tmux("paste-buffer", "-t", sessionName);
await tmux("load-buffer", "-b", bufferName, tmpFile);
await tmux("paste-buffer", "-b", bufferName, "-d", "-t", sessionName);
} finally {
try { unlinkSync(tmpFile); } catch { /* ignore cleanup errors */ }
}