From 34bc5bb2d562ac4e973a5545c885704185a29084 Mon Sep 17 00:00:00 2001 From: Harsh Batheja <40922251+harsh-batheja@users.noreply.github.com> Date: Sat, 4 Apr 2026 12:30:51 +0530 Subject: [PATCH] feat: support multiple concurrent orchestrators with isolated worktrees (#870) * feat: support multiple concurrent orchestrators with isolated worktrees - Each orchestrator session gets a numbered ID ({prefix}-orchestrator-N) and an isolated git worktree, replacing the single shared orchestrator - reserveNextOrchestratorIdentity atomically reserves the next available number and detects conflicting project prefix configurations early - isOrchestratorSession / isOrchestratorSessionName accept optional allSessionPrefixes for cross-project false-positive prevention - getProjectPause and resolveGlobalPause track the longest active pause across all concurrent orchestrators instead of returning the first found - cleanupWorktreeAndMetadata helper used consistently on all failure paths including post-launch; system prompt file included in cleanup - pollBacklog, status.ts, ProjectSidebar, sessions page, global-pause, project-utils, serialize, and sessions API route all pass allSessionPrefixes to isOrchestratorSession - Web sessions/[id]/page.tsx fetches project prefix map and passes it through prefixByProjectRef to avoid stale closure in fetchProjectSessions * fix: seed sseAttentionLevels from fresh sessions on full refresh When scheduleRefresh dispatches a reset action, derive sseAttentionLevels from the freshly fetched sessions using getAttentionLevel. This clears phantom entries for removed sessions and seeds correct levels for new sessions, preventing stale favicon color and attention counts until the next SSE snapshot. * fix: merge duplicate @/lib/types imports into single import statement Combining the separate import type and value import from @/lib/types into one import to satisfy the no-duplicate-imports lint rule. --- .../cli/__tests__/lib/session-utils.test.ts | 45 ++ packages/cli/src/commands/status.ts | 8 +- packages/cli/src/lib/session-utils.ts | 26 +- .../__tests__/session-manager/spawn.test.ts | 719 +++++++----------- packages/core/src/__tests__/types.test.ts | 30 +- packages/core/src/agent-selection.ts | 5 +- packages/core/src/lifecycle-manager.ts | 2 +- packages/core/src/recovery/validator.ts | 2 +- packages/core/src/session-manager.ts | 407 ++++++---- packages/core/src/types.ts | 36 +- packages/web/src/app/api/sessions/route.ts | 14 +- packages/web/src/app/sessions/[id]/page.tsx | 38 +- .../web/src/components/ProjectSidebar.tsx | 28 +- packages/web/src/hooks/useSessionEvents.ts | 15 +- packages/web/src/lib/dashboard-page-data.ts | 2 +- packages/web/src/lib/global-pause.ts | 24 +- packages/web/src/lib/project-name.ts | 2 + packages/web/src/lib/project-utils.ts | 12 +- packages/web/src/lib/serialize.ts | 11 +- packages/web/src/lib/services.ts | 10 +- 20 files changed, 794 insertions(+), 642 deletions(-) diff --git a/packages/cli/__tests__/lib/session-utils.test.ts b/packages/cli/__tests__/lib/session-utils.test.ts index 5d42dadab..a2d072736 100644 --- a/packages/cli/__tests__/lib/session-utils.test.ts +++ b/packages/cli/__tests__/lib/session-utils.test.ts @@ -146,4 +146,49 @@ describe("isOrchestratorSessionName", () => { const config = makeConfig({ "my-app": { sessionPrefix: "app" } }); expect(isOrchestratorSessionName(config, "app-12", "my-app")).toBe(false); }); + + it("matches worktree orchestrator IDs (orchestrator-N) for a known project", () => { + const config = makeConfig({ "my-app": { sessionPrefix: "app" } }); + expect(isOrchestratorSessionName(config, "app-orchestrator-1", "my-app")).toBe(true); + expect(isOrchestratorSessionName(config, "app-orchestrator-42", "my-app")).toBe(true); + }); + + it("matches worktree orchestrator IDs without an explicit project", () => { + const config = makeConfig({ "my-app": { sessionPrefix: "app" } }); + expect(isOrchestratorSessionName(config, "app-orchestrator-1")).toBe(true); + }); + + it("does not false-positive on a worker when prefix ends with -orchestrator", () => { + const config = makeConfig({ "my-app": { sessionPrefix: "my-orchestrator" } }); + // my-orchestrator-1 is a worker session, not a worktree orchestrator + expect(isOrchestratorSessionName(config, "my-orchestrator-1", "my-app")).toBe(false); + // my-orchestrator-orchestrator is the canonical orchestrator + expect(isOrchestratorSessionName(config, "my-orchestrator-orchestrator", "my-app")).toBe(true); + }); + + it("does not cross-project false-positive when one prefix is another's {prefix}-orchestrator", () => { + // project A prefix "app", project B prefix "app-orchestrator" + // "app-orchestrator-1" is a worker of B, NOT an orchestrator of A + const config = makeConfig({ + "project-a": { sessionPrefix: "app" }, + "project-b": { sessionPrefix: "app-orchestrator" }, + }); + expect(isOrchestratorSessionName(config, "app-orchestrator-1")).toBe(false); + // "app-orchestrator-orchestrator-1" IS an orchestrator of B + expect(isOrchestratorSessionName(config, "app-orchestrator-orchestrator-1")).toBe(true); + }); + + it("does not cross-project false-positive when projectId is provided", () => { + // project A prefix "app", project B prefix "app-orchestrator" + // "app-orchestrator-1" is a worker of B — must not be classified as orchestrator of A + // even when called with projectId="project-a" + const config = makeConfig({ + "project-a": { sessionPrefix: "app" }, + "project-b": { sessionPrefix: "app-orchestrator" }, + }); + expect(isOrchestratorSessionName(config, "app-orchestrator-1", "project-a")).toBe(false); + // The canonical orchestrator of A is still recognized + expect(isOrchestratorSessionName(config, "app-orchestrator", "project-a")).toBe(true); + expect(isOrchestratorSessionName(config, "app-orchestrator-2", "project-a")).toBe(false); + }); }); diff --git a/packages/cli/src/commands/status.ts b/packages/cli/src/commands/status.ts index 0e17d10a1..a65499cf3 100644 --- a/packages/cli/src/commands/status.ts +++ b/packages/cli/src/commands/status.ts @@ -74,7 +74,11 @@ async function gatherSessionInfo( scm: SCM, projectConfig: ReturnType, ): Promise { - const suppressPROwnership = isOrchestratorSession(session); + const sessionPrefix = projectConfig.projects[session.projectId]?.sessionPrefix ?? session.projectId; + const allSessionPrefixes = Object.entries(projectConfig.projects).map( + ([id, p]) => p.sessionPrefix ?? id, + ); + const suppressPROwnership = isOrchestratorSession(session, sessionPrefix, allSessionPrefixes); let branch = session.branch; const status = session.status; const summary = session.metadata["summary"] ?? null; @@ -144,7 +148,7 @@ async function gatherSessionInfo( return { name: session.id, - role: isOrchestratorSession(session) ? "orchestrator" : "worker", + role: isOrchestratorSession(session, sessionPrefix, allSessionPrefixes) ? "orchestrator" : "worker", branch, status, summary, diff --git a/packages/cli/src/lib/session-utils.ts b/packages/cli/src/lib/session-utils.ts index 2837afdb2..7f8e0c4b8 100644 --- a/packages/cli/src/lib/session-utils.ts +++ b/packages/cli/src/lib/session-utils.ts @@ -30,10 +30,27 @@ export function isOrchestratorSessionName( sessionName: string, projectId?: string, ): boolean { + // If sessionName is a numbered worker for any configured project, it is not an orchestrator. + // This guard runs first to prevent cross-project false positives: e.g. prefix "app" would + // match "app-orchestrator-1" as an orchestrator pattern, but if another project has prefix + // "app-orchestrator" then "app-orchestrator-1" is a worker, not an orchestrator. + for (const [id, project] of Object.entries(config.projects) as Array< + [string, OrchestratorConfig["projects"][string]] + >) { + const prefix = project.sessionPrefix || id; + if (matchesPrefix(sessionName, prefix)) return false; + } + if (projectId) { const project = config.projects[projectId]; - if (project && sessionName === `${project.sessionPrefix || projectId}-orchestrator`) { - return true; + if (project) { + const prefix = project.sessionPrefix || projectId; + if ( + sessionName === `${prefix}-orchestrator` || + new RegExp(`^${escapeRegex(prefix)}-orchestrator-\\d+$`).test(sessionName) + ) { + return true; + } } } @@ -41,7 +58,10 @@ export function isOrchestratorSessionName( [string, OrchestratorConfig["projects"][string]] >) { const prefix = project.sessionPrefix || id; - if (sessionName === `${prefix}-orchestrator`) { + if ( + sessionName === `${prefix}-orchestrator` || + new RegExp(`^${escapeRegex(prefix)}-orchestrator-\\d+$`).test(sessionName) + ) { return true; } } diff --git a/packages/core/src/__tests__/session-manager/spawn.test.ts b/packages/core/src/__tests__/session-manager/spawn.test.ts index 07aa3b741..b3c959998 100644 --- a/packages/core/src/__tests__/session-manager/spawn.test.ts +++ b/packages/core/src/__tests__/session-manager/spawn.test.ts @@ -1,5 +1,5 @@ import { describe, it, expect, beforeEach, afterEach, vi } from "vitest"; -import { chmodSync, mkdirSync, readFileSync, writeFileSync, existsSync } from "node:fs"; +import { mkdirSync, readFileSync, existsSync } from "node:fs"; import { join } from "node:path"; import { createSessionManager } from "../../session-manager.js"; import { validateConfig } from "../../config.js"; @@ -7,8 +7,6 @@ import { writeMetadata, readMetadata, readMetadataRaw, - deleteMetadata, - reserveSessionId, updateMetadata, } from "../../metadata.js"; import type { @@ -18,7 +16,6 @@ import type { Agent, Workspace, Tracker, - RuntimeHandle, } from "../../types.js"; import { setupTestContext, @@ -996,15 +993,15 @@ describe("spawn", () => { describe("spawnOrchestrator", () => { it("blocks orchestrator spawn while the project is globally paused", async () => { - writeMetadata(sessionsDir, "app-orchestrator", { + writeMetadata(sessionsDir, "app-orchestrator-1", { worktree: join(tmpDir, "my-app"), - branch: "main", + branch: "orchestrator/app-orchestrator-1", status: "working", role: "orchestrator", project: "my-app", runtimeHandle: JSON.stringify(makeHandle("rt-orchestrator")), }); - updateMetadata(sessionsDir, "app-orchestrator", { + updateMetadata(sessionsDir, "app-orchestrator-1", { globalPauseUntil: new Date(Date.now() + 60_000).toISOString(), globalPauseReason: "Rate limit reached", globalPauseSource: "app-9", @@ -1018,17 +1015,67 @@ describe("spawn", () => { expect(mockRuntime.create).not.toHaveBeenCalled(); }); + it("throws when no workspace plugin is configured", async () => { + const registryNoWorkspace: PluginRegistry = { + ...mockRegistry, + get: vi.fn().mockImplementation((slot: string) => { + if (slot === "runtime") return mockRuntime; + if (slot === "agent") return mockAgent; + return null; // no workspace plugin + }), + }; + const sm = createSessionManager({ config, registry: registryNoWorkspace }); + + await expect(sm.spawnOrchestrator({ projectId: "my-app" })).rejects.toThrow( + "spawnOrchestrator requires a workspace plugin", + ); + + // Reserved session metadata should be cleaned up + expect(readMetadataRaw(sessionsDir, "app-orchestrator-1")).toBeNull(); + expect(mockRuntime.create).not.toHaveBeenCalled(); + }); + it("creates orchestrator session with correct ID", async () => { const sm = createSessionManager({ config, registry: mockRegistry }); const session = await sm.spawnOrchestrator({ projectId: "my-app" }); - expect(session.id).toBe("app-orchestrator"); + expect(session.id).toBe("app-orchestrator-1"); expect(session.status).toBe("working"); expect(session.projectId).toBe("my-app"); - expect(session.branch).toBe("main"); + expect(session.branch).toBe("orchestrator/app-orchestrator-1"); expect(session.issueId).toBeNull(); - expect(session.workspacePath).toBe(join(tmpDir, "my-app")); + expect(session.workspacePath).toBe("/tmp/ws"); + }); + + it("creates a worktree with an orchestrator branch", async () => { + const sm = createSessionManager({ config, registry: mockRegistry }); + + await sm.spawnOrchestrator({ projectId: "my-app" }); + + expect(mockWorkspace.create).toHaveBeenCalledWith( + expect.objectContaining({ + sessionId: "app-orchestrator-1", + branch: "orchestrator/app-orchestrator-1", + projectId: "my-app", + }), + ); + }); + + it("uses the worktree path returned by the workspace plugin", async () => { + const worktreePath = join(tmpDir, "orchestrator-ws"); + (mockWorkspace.create as ReturnType).mockResolvedValueOnce({ + path: worktreePath, + branch: "orchestrator/app-orchestrator-1", + sessionId: "app-orchestrator-1", + projectId: "my-app", + }); + const sm = createSessionManager({ config, registry: mockRegistry }); + + const session = await sm.spawnOrchestrator({ projectId: "my-app" }); + + expect(session.workspacePath).toBe(worktreePath); + expect(session.branch).toBe("orchestrator/app-orchestrator-1"); }); it("writes metadata with proper fields", async () => { @@ -1036,23 +1083,113 @@ describe("spawn", () => { await sm.spawnOrchestrator({ projectId: "my-app" }); - const meta = readMetadata(sessionsDir, "app-orchestrator"); + const meta = readMetadata(sessionsDir, "app-orchestrator-1"); expect(meta).not.toBeNull(); expect(meta!.status).toBe("working"); expect(meta!.project).toBe("my-app"); - expect(meta!.worktree).toBe(join(tmpDir, "my-app")); - expect(meta!.branch).toBe("main"); + expect(meta!.branch).toBe("orchestrator/app-orchestrator-1"); expect(meta!.tmuxName).toBeDefined(); expect(meta!.runtimeHandle).toBeDefined(); }); + it("writes metadata with worktree path and orchestrator role", async () => { + const sm = createSessionManager({ config, registry: mockRegistry }); + + await sm.spawnOrchestrator({ projectId: "my-app" }); + + const meta = readMetadataRaw(sessionsDir, "app-orchestrator-1"); + expect(meta?.["role"]).toBe("orchestrator"); + expect(meta?.["branch"]).toBe("orchestrator/app-orchestrator-1"); + expect(meta?.["status"]).toBe("working"); + expect(meta?.["project"]).toBe("my-app"); + }); + + it("increments the orchestrator counter for each new session", async () => { + const sm = createSessionManager({ config, registry: mockRegistry }); + + const s1 = await sm.spawnOrchestrator({ projectId: "my-app" }); + const s2 = await sm.spawnOrchestrator({ projectId: "my-app" }); + + expect(s1.id).toBe("app-orchestrator-1"); + expect(s2.id).toBe("app-orchestrator-2"); + expect(mockWorkspace.create).toHaveBeenCalledTimes(2); + }); + + it("cleans up reserved metadata on workspace creation failure", async () => { + (mockWorkspace.create as ReturnType).mockRejectedValueOnce( + new Error("workspace creation failed"), + ); + const sm = createSessionManager({ config, registry: mockRegistry }); + + await expect(sm.spawnOrchestrator({ projectId: "my-app" })).rejects.toThrow( + "workspace creation failed", + ); + + // Reserved session file should be cleaned up + expect(readMetadataRaw(sessionsDir, "app-orchestrator-1")).toBeNull(); + }); + + it("destroys the worktree and metadata when runtime creation fails", async () => { + const worktreePath = join(tmpDir, "orchestrator-ws-rt-fail"); + (mockWorkspace.create as ReturnType).mockResolvedValueOnce({ + path: worktreePath, + branch: "orchestrator/app-orchestrator-1", + sessionId: "app-orchestrator-1", + projectId: "my-app", + }); + (mockRuntime.create as ReturnType).mockRejectedValueOnce( + new Error("runtime creation failed"), + ); + const sm = createSessionManager({ config, registry: mockRegistry }); + + await expect(sm.spawnOrchestrator({ projectId: "my-app" })).rejects.toThrow( + "runtime creation failed", + ); + + expect(mockWorkspace.destroy).toHaveBeenCalledWith(worktreePath); + expect(readMetadataRaw(sessionsDir, "app-orchestrator-1")).toBeNull(); + }); + + it("destroys the worktree when post-launch setup fails", async () => { + const worktreePath = join(tmpDir, "orchestrator-ws-postlaunch-fail"); + (mockWorkspace.create as ReturnType).mockResolvedValueOnce({ + path: worktreePath, + branch: "orchestrator/app-orchestrator-1", + sessionId: "app-orchestrator-1", + projectId: "my-app", + }); + const postLaunchError = new Error("post-launch setup failed"); + const agentWithPostLaunch: typeof mockAgent = { + ...mockAgent, + postLaunchSetup: vi.fn().mockRejectedValueOnce(postLaunchError), + }; + const registryWithPostLaunch: PluginRegistry = { + ...mockRegistry, + get: vi.fn().mockImplementation((slot: string) => { + if (slot === "runtime") return mockRuntime; + if (slot === "agent") return agentWithPostLaunch; + if (slot === "workspace") return mockWorkspace; + return null; + }), + }; + const sm = createSessionManager({ config, registry: registryWithPostLaunch }); + + await expect(sm.spawnOrchestrator({ projectId: "my-app" })).rejects.toThrow( + "post-launch setup failed", + ); + + expect(mockRuntime.destroy).toHaveBeenCalled(); + expect(mockWorkspace.destroy).toHaveBeenCalledWith(worktreePath); + expect(readMetadataRaw(sessionsDir, "app-orchestrator-1")).toBeNull(); + }); + it("deletes previous OpenCode orchestrator sessions before starting", async () => { const deleteLogPath = join(tmpDir, "opencode-delete-orchestrator.log"); const mockBin = installMockOpencode( tmpDir, JSON.stringify([ - { id: "ses_old", title: "AO:app-orchestrator", updated: "2025-01-01T00:00:00.000Z" }, - { id: "ses_new", title: "AO:app-orchestrator", updated: "2025-01-02T00:00:00.000Z" }, + { id: "ses_old", title: "AO:app-orchestrator-1", updated: "2025-01-01T00:00:00.000Z" }, + { id: "ses_new", title: "AO:app-orchestrator-1", updated: "2025-01-02T00:00:00.000Z" }, ]), deleteLogPath, ); @@ -1094,14 +1231,14 @@ describe("spawn", () => { expect(opencodeAgent.getLaunchCommand).toHaveBeenCalledWith( expect.objectContaining({ - sessionId: "app-orchestrator", + sessionId: "app-orchestrator-1", projectConfig: expect.objectContaining({ agentConfig: expect.not.objectContaining({ opencodeSessionId: expect.any(String) }), }), }), ); - const meta = readMetadataRaw(sessionsDir, "app-orchestrator"); + const meta = readMetadataRaw(sessionsDir, "app-orchestrator-1"); expect(meta?.["agent"]).toBe("opencode"); expect(meta?.["opencodeSessionId"]).toBeUndefined(); }); @@ -1113,7 +1250,7 @@ describe("spawn", () => { JSON.stringify([ { id: "ses_discovered_orchestrator", - title: "AO:app-orchestrator", + title: "AO:app-orchestrator-1", updated: 1_772_777_000_000, }, ]), @@ -1151,205 +1288,20 @@ describe("spawn", () => { const sm = createSessionManager({ config: configWithReuse, registry: registryWithOpenCode }); await sm.spawnOrchestrator({ projectId: "my-app" }); - const meta = readMetadataRaw(sessionsDir, "app-orchestrator"); + const meta = readMetadataRaw(sessionsDir, "app-orchestrator-1"); expect(meta?.["opencodeSessionId"]).toBe("ses_discovered_orchestrator"); }); - it("reuses an existing orchestrator session when strategy is reuse", async () => { - const listLogPath = join(tmpDir, "opencode-list-orchestrator-reuse.log"); - const mockBin = join(tmpDir, "mock-bin-reuse-no-list"); - mkdirSync(mockBin, { recursive: true }); - const scriptPath = join(mockBin, "opencode"); - writeFileSync( - scriptPath, - [ - "#!/usr/bin/env bash", - "set -euo pipefail", - 'if [[ "$1" == "session" && "$2" == "list" ]]; then', - ` printf '%s\\n' "$*" >> '${listLogPath.replace(/'/g, "'\\''")}'`, - " printf '[]\\n'", - " exit 0", - "fi", - "exit 0", - "", - ].join("\n"), - "utf-8", - ); - chmodSync(scriptPath, 0o755); - process.env.PATH = `${mockBin}:${originalPath ?? ""}`; - - const opencodeAgent: Agent = { - ...mockAgent, - name: "opencode", - }; - const registryWithOpenCode: PluginRegistry = { - ...mockRegistry, - get: vi.fn().mockImplementation((slot: string) => { - if (slot === "runtime") return mockRuntime; - if (slot === "agent") return opencodeAgent; - if (slot === "workspace") return mockWorkspace; - return null; - }), - }; - - const configWithReuse: OrchestratorConfig = { - ...config, - defaults: { ...config.defaults, agent: "opencode" }, - projects: { - ...config.projects, - "my-app": { - ...config.projects["my-app"], - agent: "opencode", - orchestratorSessionStrategy: "reuse", - }, - }, - }; - - writeMetadata(sessionsDir, "app-orchestrator", { - worktree: join(tmpDir, "my-app"), - branch: "main", - status: "working", - role: "orchestrator", - project: "my-app", - agent: "opencode", - runtimeHandle: JSON.stringify(makeHandle("rt-existing")), - opencodeSessionId: "ses_existing", - createdAt: new Date().toISOString(), - }); - - const sm = createSessionManager({ config: configWithReuse, registry: registryWithOpenCode }); - const session = await sm.spawnOrchestrator({ projectId: "my-app" }); - - expect(session.id).toBe("app-orchestrator"); - expect(session.metadata["orchestratorSessionReused"]).toBe("true"); - expect(mockRuntime.create).not.toHaveBeenCalled(); - expect(mockRuntime.destroy).not.toHaveBeenCalled(); - expect(existsSync(listLogPath)).toBe(false); - }); - - it("destroys orphaned runtime when reuse strategy finds alive runtime but get returns null", async () => { - const deleteLogPath = join(tmpDir, "opencode-delete-orphaned-runtime.log"); - const mockBin = installMockOpencode(tmpDir, "[]", deleteLogPath); - process.env.PATH = `${mockBin}:${originalPath ?? ""}`; - - const opencodeAgent: Agent = { - ...mockAgent, - name: "opencode", - }; - const registryWithOpenCode: PluginRegistry = { - ...mockRegistry, - get: vi.fn().mockImplementation((slot: string) => { - if (slot === "runtime") return mockRuntime; - if (slot === "agent") return opencodeAgent; - if (slot === "workspace") return mockWorkspace; - return null; - }), - }; - - const configWithReuse: OrchestratorConfig = { - ...config, - defaults: { ...config.defaults, agent: "opencode" }, - projects: { - ...config.projects, - "my-app": { - ...config.projects["my-app"], - agent: "opencode", - orchestratorSessionStrategy: "reuse", - }, - }, - }; - - const orphanedHandle = makeHandle("rt-orphaned"); - writeMetadata(sessionsDir, "app-orchestrator", { - worktree: join(tmpDir, "my-app"), - branch: "main", - status: "working", - role: "orchestrator", - project: "my-app", - agent: "opencode", - runtimeHandle: JSON.stringify(orphanedHandle), - createdAt: new Date().toISOString(), - }); - - vi.mocked(mockRuntime.isAlive).mockImplementation(async (handle: RuntimeHandle) => { - if (handle?.id === "rt-orphaned") { - deleteMetadata(sessionsDir, "app-orchestrator"); - return true; - } - return false; - }); - - const sm = createSessionManager({ config: configWithReuse, registry: registryWithOpenCode }); - const session = await sm.spawnOrchestrator({ projectId: "my-app" }); - - expect(session.id).toBe("app-orchestrator"); - expect(mockRuntime.destroy).toHaveBeenCalledWith(orphanedHandle); - expect(mockRuntime.create).toHaveBeenCalled(); - }); - - it("reuses mapped OpenCode session id when strategy is reuse and runtime is restarted", async () => { - const opencodeAgent: Agent = { - ...mockAgent, - name: "opencode", - }; - const registryWithOpenCode: PluginRegistry = { - ...mockRegistry, - get: vi.fn().mockImplementation((slot: string) => { - if (slot === "runtime") return mockRuntime; - if (slot === "agent") return opencodeAgent; - if (slot === "workspace") return mockWorkspace; - return null; - }), - }; - - const configWithReuse: OrchestratorConfig = { - ...config, - defaults: { ...config.defaults, agent: "opencode" }, - projects: { - ...config.projects, - "my-app": { - ...config.projects["my-app"], - agent: "opencode", - orchestratorSessionStrategy: "reuse", - }, - }, - }; - - writeMetadata(sessionsDir, "app-orchestrator", { - worktree: join(tmpDir, "my-app"), - branch: "main", - status: "working", - role: "orchestrator", - project: "my-app", - agent: "opencode", - runtimeHandle: JSON.stringify(makeHandle("rt-existing")), - opencodeSessionId: "ses_existing", - createdAt: new Date().toISOString(), - }); - - vi.mocked(mockRuntime.isAlive).mockResolvedValue(false); - - const sm = createSessionManager({ config: configWithReuse, registry: registryWithOpenCode }); - await sm.spawnOrchestrator({ projectId: "my-app" }); - - expect(opencodeAgent.getLaunchCommand).toHaveBeenCalledWith( - expect.objectContaining({ - projectConfig: expect.objectContaining({ - agentConfig: expect.objectContaining({ opencodeSessionId: "ses_existing" }), - }), - }), - ); - const meta = readMetadataRaw(sessionsDir, "app-orchestrator"); - expect(meta?.["opencodeSessionId"]).toBe("ses_existing"); - }); - - it("reuses archived OpenCode mapping for orchestrator when active metadata has no mapping", async () => { - const deleteLogPath = join(tmpDir, "opencode-delete-orchestrator-reuse-archived.log"); + it("reuses mapped OpenCode session id when strategy is reuse and opencode lists it by title", async () => { + const deleteLogPath = join(tmpDir, "opencode-delete-orchestrator-reuse-restart.log"); const mockBin = installMockOpencode( tmpDir, JSON.stringify([ - null, - { id: "ses_existing", title: "AO:app-orchestrator", updated: 1_772_777_000_000 }, + { + id: "ses_existing", + title: "AO:app-orchestrator-1", + updated: 1_772_777_000_000, + }, ]), deleteLogPath, ); @@ -1382,30 +1334,57 @@ describe("spawn", () => { }, }; - writeMetadata(sessionsDir, "app-orchestrator", { - worktree: join(tmpDir, "my-app"), - branch: "main", - status: "working", - role: "orchestrator", - project: "my-app", - agent: "opencode", - runtimeHandle: JSON.stringify(makeHandle("rt-existing")), - opencodeSessionId: "ses_existing", - createdAt: new Date().toISOString(), - }); - deleteMetadata(sessionsDir, "app-orchestrator", true); - writeMetadata(sessionsDir, "app-orchestrator", { - worktree: join(tmpDir, "my-app"), - branch: "main", - status: "working", - role: "orchestrator", - project: "my-app", - agent: "opencode", - runtimeHandle: JSON.stringify(makeHandle("rt-existing")), - createdAt: new Date().toISOString(), - }); + const sm = createSessionManager({ config: configWithReuse, registry: registryWithOpenCode }); + await sm.spawnOrchestrator({ projectId: "my-app" }); - vi.mocked(mockRuntime.isAlive).mockResolvedValue(false); + expect(opencodeAgent.getLaunchCommand).toHaveBeenCalledWith( + expect.objectContaining({ + projectConfig: expect.objectContaining({ + agentConfig: expect.objectContaining({ opencodeSessionId: "ses_existing" }), + }), + }), + ); + const meta = readMetadataRaw(sessionsDir, "app-orchestrator-1"); + expect(meta?.["opencodeSessionId"]).toBe("ses_existing"); + }); + + it("discovers OpenCode mapping by title when no archived mapping exists for new session id", async () => { + const deleteLogPath = join(tmpDir, "opencode-delete-orchestrator-reuse-title-fallback.log"); + const mockBin = installMockOpencode( + tmpDir, + JSON.stringify([ + { id: "ses_existing", title: "AO:app-orchestrator-1", updated: 1_772_777_000_000 }, + ]), + deleteLogPath, + ); + process.env.PATH = `${mockBin}:${originalPath ?? ""}`; + + const opencodeAgent: Agent = { + ...mockAgent, + name: "opencode", + }; + const registryWithOpenCode: PluginRegistry = { + ...mockRegistry, + get: vi.fn().mockImplementation((slot: string) => { + if (slot === "runtime") return mockRuntime; + if (slot === "agent") return opencodeAgent; + if (slot === "workspace") return mockWorkspace; + return null; + }), + }; + + const configWithReuse: OrchestratorConfig = { + ...config, + defaults: { ...config.defaults, agent: "opencode" }, + projects: { + ...config.projects, + "my-app": { + ...config.projects["my-app"], + agent: "opencode", + orchestratorSessionStrategy: "reuse", + }, + }, + }; const sm = createSessionManager({ config: configWithReuse, registry: registryWithOpenCode }); await sm.spawnOrchestrator({ projectId: "my-app" }); @@ -1424,8 +1403,7 @@ describe("spawn", () => { const mockBin = installMockOpencode( tmpDir, JSON.stringify([ - null, - { id: "ses_title_match", title: "AO:app-orchestrator", updated: 1_772_777_000_000 }, + { id: "ses_title_match", title: "AO:app-orchestrator-1", updated: 1_772_777_000_000 }, ]), deleteLogPath, ); @@ -1458,19 +1436,6 @@ describe("spawn", () => { }, }; - writeMetadata(sessionsDir, "app-orchestrator", { - worktree: join(tmpDir, "my-app"), - branch: "main", - status: "working", - role: "orchestrator", - project: "my-app", - agent: "opencode", - runtimeHandle: JSON.stringify(makeHandle("rt-existing")), - createdAt: new Date().toISOString(), - }); - - vi.mocked(mockRuntime.isAlive).mockResolvedValue(false); - const sm = createSessionManager({ config: configWithReuse, registry: registryWithOpenCode }); await sm.spawnOrchestrator({ projectId: "my-app" }); @@ -1481,81 +1446,11 @@ describe("spawn", () => { }), }), ); - const meta = readMetadataRaw(sessionsDir, "app-orchestrator"); + const meta = readMetadataRaw(sessionsDir, "app-orchestrator-1"); expect(meta?.["opencodeSessionId"]).toBe("ses_title_match"); }); - it("starts fresh without deleting prior OpenCode sessions when strategy is ignore", async () => { - const deleteLogPath = join(tmpDir, "opencode-delete-ignore.log"); - const mockBin = installMockOpencode( - tmpDir, - JSON.stringify([ - { id: "ses_old", title: "AO:app-orchestrator", updated: "2025-01-01T00:00:00.000Z" }, - ]), - deleteLogPath, - ); - process.env.PATH = `${mockBin}:${originalPath ?? ""}`; - - const opencodeAgent: Agent = { - ...mockAgent, - name: "opencode", - }; - const registryWithOpenCode: PluginRegistry = { - ...mockRegistry, - get: vi.fn().mockImplementation((slot: string) => { - if (slot === "runtime") return mockRuntime; - if (slot === "agent") return opencodeAgent; - if (slot === "workspace") return mockWorkspace; - return null; - }), - }; - - const configWithIgnoreNew: OrchestratorConfig = { - ...config, - defaults: { ...config.defaults, agent: "opencode" }, - projects: { - ...config.projects, - "my-app": { - ...config.projects["my-app"], - agent: "opencode", - orchestratorSessionStrategy: "ignore", - }, - }, - }; - - writeMetadata(sessionsDir, "app-orchestrator", { - worktree: join(tmpDir, "my-app"), - branch: "main", - status: "working", - role: "orchestrator", - project: "my-app", - agent: "opencode", - runtimeHandle: JSON.stringify(makeHandle("rt-existing")), - createdAt: new Date().toISOString(), - }); - - vi.mocked(mockRuntime.isAlive).mockResolvedValueOnce(true); - - const sm = createSessionManager({ - config: configWithIgnoreNew, - registry: registryWithOpenCode, - }); - await sm.spawnOrchestrator({ projectId: "my-app" }); - - expect(mockRuntime.destroy).toHaveBeenCalledWith(makeHandle("rt-existing")); - expect(mockRuntime.create).toHaveBeenCalled(); - expect(existsSync(deleteLogPath)).toBe(false); - }); - - it("skips workspace creation", async () => { - const sm = createSessionManager({ config, registry: mockRegistry }); - - await sm.spawnOrchestrator({ projectId: "my-app" }); - - expect(mockWorkspace.create).not.toHaveBeenCalled(); - }); - - it("calls agent.setupWorkspaceHooks on project path", async () => { + it("calls agent.setupWorkspaceHooks on worktree path", async () => { const agentWithHooks: Agent = { ...mockAgent, setupWorkspaceHooks: vi.fn().mockResolvedValue(undefined), @@ -1574,7 +1469,7 @@ describe("spawn", () => { await sm.spawnOrchestrator({ projectId: "my-app" }); expect(agentWithHooks.setupWorkspaceHooks).toHaveBeenCalledWith( - join(tmpDir, "my-app"), + "/tmp/ws", expect.objectContaining({ dataDir: sessionsDir }), ); }); @@ -1586,7 +1481,7 @@ describe("spawn", () => { expect(mockRuntime.create).toHaveBeenCalledWith( expect.objectContaining({ - workspacePath: join(tmpDir, "my-app"), + workspacePath: "/tmp/ws", launchCommand: "mock-agent --start", }), ); @@ -1597,31 +1492,10 @@ describe("spawn", () => { await sm.spawnOrchestrator({ projectId: "my-app" }); - const meta = readMetadataRaw(sessionsDir, "app-orchestrator"); + const meta = readMetadataRaw(sessionsDir, "app-orchestrator-1"); expect(meta?.["orchestratorSessionReused"]).toBeUndefined(); }); - it("respawns the orchestrator when stale metadata exists but the runtime is dead", async () => { - writeMetadata(sessionsDir, "app-orchestrator", { - worktree: join(tmpDir, "my-app"), - branch: "main", - status: "working", - project: "my-app", - role: "orchestrator", - runtimeHandle: JSON.stringify(makeHandle("rt-stale")), - createdAt: new Date().toISOString(), - }); - - vi.mocked(mockRuntime.isAlive).mockResolvedValue(false); - - const sm = createSessionManager({ config, registry: mockRegistry }); - await sm.spawnOrchestrator({ projectId: "my-app" }); - - expect(mockRuntime.create).toHaveBeenCalledTimes(1); - const meta = readMetadataRaw(sessionsDir, "app-orchestrator"); - expect(meta?.["runtimeHandle"]).toBe(JSON.stringify(makeHandle("rt-1"))); - }); - it("uses orchestratorModel when configured", async () => { const configWithOrchestratorModel: OrchestratorConfig = { ...config, @@ -1720,7 +1594,7 @@ describe("spawn", () => { expect(mockCodexAgent.getLaunchCommand).toHaveBeenCalled(); expect(mockAgent.getLaunchCommand).not.toHaveBeenCalled(); - expect(readMetadataRaw(sessionsDir, "app-orchestrator")?.["agent"]).toBe("codex"); + expect(readMetadataRaw(sessionsDir, "app-orchestrator-1")?.["agent"]).toBe("codex"); }); it("uses defaults orchestrator agent when project agent is not set", async () => { @@ -1767,7 +1641,7 @@ describe("spawn", () => { await sm.spawnOrchestrator({ projectId: "my-app" }); expect(mockCodexAgent.getLaunchCommand).toHaveBeenCalled(); - expect(readMetadataRaw(sessionsDir, "app-orchestrator")?.["agent"]).toBe("codex"); + expect(readMetadataRaw(sessionsDir, "app-orchestrator-1")?.["agent"]).toBe("codex"); }); it("keeps shared worker permissions when role-specific config only overrides model", async () => { @@ -1869,8 +1743,8 @@ describe("spawn", () => { // Should pass systemPromptFile (not inline systemPrompt) to avoid tmux truncation expect(mockAgent.getLaunchCommand).toHaveBeenCalledWith( expect.objectContaining({ - sessionId: "app-orchestrator", - systemPromptFile: expect.stringContaining("orchestrator-prompt.md"), + sessionId: "app-orchestrator-1", + systemPromptFile: expect.stringContaining("orchestrator-prompt-app-orchestrator-1.md"), }), ); @@ -1909,101 +1783,80 @@ describe("spawn", () => { expect(session.runtimeHandle).toEqual(makeHandle("rt-1")); }); - it("reuses existing orchestrator on reservation conflict when strategy is reuse", async () => { - const opencodeAgent: Agent = { - ...mockAgent, - name: "opencode", - }; - const registryWithOpenCode: PluginRegistry = { - ...mockRegistry, - get: vi.fn().mockImplementation((slot: string) => { - if (slot === "runtime") return mockRuntime; - if (slot === "agent") return opencodeAgent; - if (slot === "workspace") return mockWorkspace; - return null; - }), - }; - - const configWithReuse: OrchestratorConfig = { - ...config, - defaults: { ...config.defaults, agent: "opencode" }, - projects: { - ...config.projects, - "my-app": { - ...config.projects["my-app"], - agent: "opencode", - orchestratorSessionStrategy: "reuse", - }, - }, - }; - - writeMetadata(sessionsDir, "app-orchestrator", { + it("blocks spawn while the project is globally paused (orchestrator-N orchestrator)", async () => { + // Pause set by a worktree-based orchestrator + writeMetadata(sessionsDir, "app-orchestrator-1", { worktree: join(tmpDir, "my-app"), - branch: "main", + branch: "orchestrator/app-orchestrator-1", status: "working", role: "orchestrator", project: "my-app", - agent: "opencode", - runtimeHandle: JSON.stringify(makeHandle("rt-concurrent")), - opencodeSessionId: "ses_concurrent", - createdAt: new Date().toISOString(), + runtimeHandle: JSON.stringify(makeHandle("rt-orchestrator-1")), }); - - vi.mocked(mockRuntime.isAlive).mockResolvedValue(true); - - const sm = createSessionManager({ config: configWithReuse, registry: registryWithOpenCode }); - const session = await sm.spawnOrchestrator({ projectId: "my-app" }); - - expect(session.metadata["orchestratorSessionReused"]).toBe("true"); - expect(mockRuntime.create).not.toHaveBeenCalled(); - }); - - it("recovers reservation conflict when existing session is not usable", async () => { - writeMetadata(sessionsDir, "app-orchestrator", { - worktree: join(tmpDir, "my-app"), - branch: "main", - status: "killed", - role: "orchestrator", - project: "my-app", - runtimeHandle: JSON.stringify(makeHandle("rt-dead")), - createdAt: new Date().toISOString(), + updateMetadata(sessionsDir, "app-orchestrator-1", { + globalPauseUntil: new Date(Date.now() + 60_000).toISOString(), + globalPauseReason: "Rate limit reached", + globalPauseSource: "app-9", }); - vi.mocked(mockRuntime.isAlive).mockResolvedValue(false); - - const sm = createSessionManager({ config, registry: mockRegistry }); - await expect(sm.spawnOrchestrator({ projectId: "my-app" })).resolves.toBeDefined(); - expect(mockRuntime.create).toHaveBeenCalledTimes(1); - }); - - it("creates only one runtime on reservation conflict", async () => { - writeMetadata(sessionsDir, "app-orchestrator", { - worktree: join(tmpDir, "my-app"), - branch: "main", - status: "working", - role: "orchestrator", - project: "my-app", - runtimeHandle: JSON.stringify(makeHandle("rt-existing")), - createdAt: new Date().toISOString(), - }); - - vi.mocked(mockRuntime.isAlive).mockResolvedValue(false); - - const sm = createSessionManager({ config, registry: mockRegistry }); - await expect(sm.spawnOrchestrator({ projectId: "my-app" })).resolves.toBeDefined(); - expect(mockRuntime.create).toHaveBeenCalledTimes(1); - }); - - it("does not delete an in-progress reservation file without runtime metadata", async () => { - expect(reserveSessionId(sessionsDir, "app-orchestrator")).toBe(true); - const sm = createSessionManager({ config, registry: mockRegistry }); await expect(sm.spawnOrchestrator({ projectId: "my-app" })).rejects.toThrow( - "already exists but is not in a reusable state", + "Project is paused due to model rate limit until", ); - expect(mockRuntime.create).not.toHaveBeenCalled(); - expect(readMetadataRaw(sessionsDir, "app-orchestrator")).toEqual({}); + }); + + it("does not skip pause check for orchestrator-N orchestrator when sending to workers", async () => { + // Worker session + writeMetadata(sessionsDir, "app-1", { + worktree: join(tmpDir, "ws-1"), + branch: "session/app-1", + status: "working", + project: "my-app", + runtimeHandle: JSON.stringify(makeHandle("rt-1")), + }); + // Pause set by orchestrator-1 + writeMetadata(sessionsDir, "app-orchestrator-1", { + worktree: join(tmpDir, "my-app"), + branch: "orchestrator/app-orchestrator-1", + status: "working", + role: "orchestrator", + project: "my-app", + runtimeHandle: JSON.stringify(makeHandle("rt-orchestrator-1")), + }); + updateMetadata(sessionsDir, "app-orchestrator-1", { + globalPauseUntil: new Date(Date.now() + 60_000).toISOString(), + globalPauseReason: "Rate limit", + globalPauseSource: "app-orchestrator-1", + }); + + const sm = createSessionManager({ config, registry: mockRegistry }); + + await expect(sm.send("app-1", "hello")).rejects.toThrow( + "Project is paused due to model rate limit until", + ); + }); + + it("allows orchestrator-N orchestrator session to send despite project pause", async () => { + // Orch-1 session itself is alive + writeMetadata(sessionsDir, "app-orchestrator-1", { + worktree: join(tmpDir, "my-app"), + branch: "orchestrator/app-orchestrator-1", + status: "working", + role: "orchestrator", + project: "my-app", + runtimeHandle: JSON.stringify(makeHandle("rt-orchestrator-1")), + }); + updateMetadata(sessionsDir, "app-orchestrator-1", { + globalPauseUntil: new Date(Date.now() + 60_000).toISOString(), + globalPauseReason: "Rate limit", + globalPauseSource: "app-orchestrator-1", + }); + + const sm = createSessionManager({ config, registry: mockRegistry }); + + // The orchestrator itself should be exempt from the pause + await expect(sm.send("app-orchestrator-1", "continue")).resolves.not.toThrow(); }); }); }); diff --git a/packages/core/src/__tests__/types.test.ts b/packages/core/src/__tests__/types.test.ts index bf54e4b0c..911d36be0 100644 --- a/packages/core/src/__tests__/types.test.ts +++ b/packages/core/src/__tests__/types.test.ts @@ -4,19 +4,35 @@ import { isOrchestratorSession, isIssueNotFoundError } from "../types.js"; describe("isOrchestratorSession", () => { it("detects orchestrators by explicit role metadata", () => { expect( - isOrchestratorSession({ - id: "app-control", - metadata: { role: "orchestrator" }, - }), + isOrchestratorSession({ id: "app-control", metadata: { role: "orchestrator" } }, "app"), ).toBe(true); }); it("falls back to orchestrator naming for legacy sessions", () => { - expect(isOrchestratorSession({ id: "app-orchestrator", metadata: {} })).toBe(true); + expect(isOrchestratorSession({ id: "app-orchestrator", metadata: {} }, "app")).toBe(true); }); - it("does not classify worker sessions as orchestrators", () => { - expect(isOrchestratorSession({ id: "app-7", metadata: { role: "worker" } })).toBe(false); + it("detects numbered worktree orchestrators by prefix pattern", () => { + expect(isOrchestratorSession({ id: "app-orchestrator-1", metadata: {} }, "app")).toBe(true); + expect(isOrchestratorSession({ id: "app-orchestrator-42", metadata: {} }, "app")).toBe(true); + }); + + it("does not false-positive on worker sessions", () => { + expect(isOrchestratorSession({ id: "app-7", metadata: { role: "worker" } }, "app")).toBe(false); + }); + + it("does not false-positive when prefix ends with -orchestrator", () => { + // my-orchestrator-1 is a worker when prefix is "my-orchestrator" + expect( + isOrchestratorSession({ id: "my-orchestrator-1", metadata: {} }, "my-orchestrator"), + ).toBe(false); + // my-orchestrator-orchestrator-1 is the real worktree orchestrator + expect( + isOrchestratorSession( + { id: "my-orchestrator-orchestrator-1", metadata: {} }, + "my-orchestrator", + ), + ).toBe(true); }); }); diff --git a/packages/core/src/agent-selection.ts b/packages/core/src/agent-selection.ts index de8a76d77..158cb280e 100644 --- a/packages/core/src/agent-selection.ts +++ b/packages/core/src/agent-selection.ts @@ -20,9 +20,10 @@ export interface ResolvedAgentSelection { export function resolveSessionRole( sessionId: string, - metadata?: Record, + metadata: Record | undefined, + sessionPrefix: string, ): SessionRole { - return isOrchestratorSession({ id: sessionId, metadata }) ? "orchestrator" : "worker"; + return isOrchestratorSession({ id: sessionId, metadata }, sessionPrefix) ? "orchestrator" : "worker"; } export function resolveAgentSelection(params: { diff --git a/packages/core/src/lifecycle-manager.ts b/packages/core/src/lifecycle-manager.ts index c3f110b3d..b2fc74c8e 100644 --- a/packages/core/src/lifecycle-manager.ts +++ b/packages/core/src/lifecycle-manager.ts @@ -345,7 +345,7 @@ export function createLifecycleManager(deps: LifecycleManagerDeps): LifecycleMan if (!project) return session.status; const agentName = resolveAgentSelection({ - role: resolveSessionRole(session.id, session.metadata), + role: resolveSessionRole(session.id, session.metadata, project.sessionPrefix), project, defaults: config.defaults, persistedAgent: session.metadata["agent"], diff --git a/packages/core/src/recovery/validator.ts b/packages/core/src/recovery/validator.ts index 3a896e06f..410e0afbe 100644 --- a/packages/core/src/recovery/validator.ts +++ b/packages/core/src/recovery/validator.ts @@ -31,7 +31,7 @@ export async function validateSession( const runtimeName = project.runtime ?? config.defaults.runtime; const agentName = resolveAgentSelection({ - role: resolveSessionRole(sessionId, rawMetadata), + role: resolveSessionRole(sessionId, rawMetadata, project.sessionPrefix), project, defaults: config.defaults, persistedAgent: rawMetadata["agent"], diff --git a/packages/core/src/session-manager.ts b/packages/core/src/session-manager.ts index 3e6a67455..566fdeada 100644 --- a/packages/core/src/session-manager.ts +++ b/packages/core/src/session-manager.ts @@ -11,7 +11,7 @@ * Reference: scripts/claude-ao-session, scripts/send-to-session */ -import { statSync, existsSync, readdirSync, writeFileSync, mkdirSync, utimesSync } from "node:fs"; +import { statSync, existsSync, readdirSync, writeFileSync, mkdirSync, utimesSync, unlinkSync } from "node:fs"; import { execFile } from "node:child_process"; import { basename, join, resolve } from "node:path"; import { homedir } from "node:os"; @@ -41,7 +41,6 @@ import { type PluginRegistry, type RuntimeHandle, type Issue, - isOrchestratorSession, PR_STATE, } from "./types.js"; import { @@ -60,7 +59,6 @@ import { getWorktreesDir, getProjectBaseDir, generateTmuxName, - generateConfigHash, validateAndStoreOrigin, } from "./paths.js"; import { asValidOpenCodeSessionId } from "./opencode-session-id.js"; @@ -292,19 +290,40 @@ export function createSessionManager(deps: SessionManagerDeps): OpenCodeSessionM sourceSessionId: string; } | null { const sessionsDir = getProjectSessionsDir(project); - const orchestratorId = `${project.sessionPrefix}-orchestrator`; - const orchestratorRaw = readMetadataRaw(sessionsDir, orchestratorId); - if (!orchestratorRaw) return null; - const until = parsePauseUntil(orchestratorRaw[GLOBAL_PAUSE_UNTIL_KEY]); - if (!until) return null; - if (until.getTime() <= Date.now()) return null; + // Build the candidate list using name-pattern matching so we never read + // raw metadata for worker sessions (avoids N file reads on every hot path). + // Do not pre-seed the canonical ID — with the numbered orchestrator scheme + // ({prefix}-orchestrator-N) it will rarely exist, and pre-seeding it would + // cause an unconditional extra readMetadataRaw on every hot-path invocation. + const canonicalOrchestratorId = `${project.sessionPrefix}-orchestrator`; + const orchestratorPattern = new RegExp(`^${escapeRegex(project.sessionPrefix)}-orchestrator-(\\d+)$`); + const candidateIds = new Set(); + for (const id of listMetadata(sessionsDir)) { + if (id === canonicalOrchestratorId || orchestratorPattern.test(id)) { + candidateIds.add(id); + } + } - return { - until, - reason: orchestratorRaw[GLOBAL_PAUSE_REASON_KEY] ?? "Model rate limit reached", - sourceSessionId: orchestratorRaw[GLOBAL_PAUSE_SOURCE_KEY] ?? "unknown", - }; + let best: { until: Date; reason: string; sourceSessionId: string } | null = null; + for (const sessionId of candidateIds) { + const raw = readMetadataRaw(sessionsDir, sessionId); + if (!raw) continue; + if (!isOrchestratorSessionRecord(sessionId, raw, project.sessionPrefix)) continue; + + const until = parsePauseUntil(raw[GLOBAL_PAUSE_UNTIL_KEY]); + if (!until || until.getTime() <= Date.now()) continue; + + if (!best || until.getTime() > best.until.getTime()) { + best = { + until, + reason: raw[GLOBAL_PAUSE_REASON_KEY] ?? "Model rate limit reached", + sourceSessionId: raw[GLOBAL_PAUSE_SOURCE_KEY] ?? "unknown", + }; + } + } + + return best; } function normalizePath(path: string): string { @@ -358,9 +377,17 @@ export function createSessionManager(deps: SessionManagerDeps): OpenCodeSessionM function isOrchestratorSessionRecord( sessionId: string, raw: Record | null | undefined, + sessionPrefix?: string, ): boolean { if (!raw) return false; - return raw["role"] === "orchestrator" || sessionId.endsWith("-orchestrator"); + if (raw["role"] === "orchestrator" || sessionId.endsWith("-orchestrator")) return true; + // Check the -orchestrator-N pattern only when the prefix is known so the + // regex is anchored to the project prefix, preventing false-positives when + // the user-configured sessionPrefix itself ends with "-orchestrator". + if (sessionPrefix) { + return new RegExp(`^${escapeRegex(sessionPrefix)}-orchestrator-\\d+$`).test(sessionId); + } + return false; } function isCleanupProtectedSession( @@ -368,11 +395,7 @@ export function createSessionManager(deps: SessionManagerDeps): OpenCodeSessionM sessionId: string, metadata?: Record | null, ): boolean { - const canonicalOrchestratorId = `${project.sessionPrefix}-orchestrator`; - return ( - sessionId === canonicalOrchestratorId || - isOrchestratorSession({ id: sessionId, metadata: metadata ?? undefined }) - ); + return isOrchestratorSessionRecord(sessionId, metadata ?? {}, project.sessionPrefix); } function applyMetadataUpdatesToRaw( @@ -422,9 +445,10 @@ export function createSessionManager(deps: SessionManagerDeps): OpenCodeSessionM function repairSingleSessionMetadataOnRead( sessionsDir: string, record: ActiveSessionRecord, + sessionPrefix?: string, ): ActiveSessionRecord { const repaired = { ...record, raw: { ...record.raw } }; - if (!isOrchestratorSessionRecord(repaired.sessionName, repaired.raw)) { + if (!isOrchestratorSessionRecord(repaired.sessionName, repaired.raw, sessionPrefix)) { return repaired; } @@ -464,13 +488,14 @@ export function createSessionManager(deps: SessionManagerDeps): OpenCodeSessionM function repairSessionMetadataOnRead( sessionsDir: string, records: ActiveSessionRecord[], + sessionPrefix?: string, ): ActiveSessionRecord[] { const repaired = records.map((record) => ({ ...record, raw: { ...record.raw } })); const duplicatePRAttachments = new Map(); for (const record of repaired) { - if (isOrchestratorSessionRecord(record.sessionName, record.raw)) { - record.raw = repairSingleSessionMetadataOnRead(sessionsDir, record).raw; + if (isOrchestratorSessionRecord(record.sessionName, record.raw, sessionPrefix)) { + record.raw = repairSingleSessionMetadataOnRead(sessionsDir, record, sessionPrefix).raw; continue; } @@ -531,7 +556,7 @@ export function createSessionManager(deps: SessionManagerDeps): OpenCodeSessionM return [{ sessionName, raw, modifiedAt } satisfies ActiveSessionRecord]; }); - return repairSessionMetadataOnRead(sessionsDir, records); + return repairSessionMetadataOnRead(sessionsDir, records, project.sessionPrefix); } function markArchivedOpenCodeCleanup(sessionsDir: string, sessionId: SessionId): void { @@ -702,6 +727,66 @@ export function createSessionManager(deps: SessionManagerDeps): OpenCodeSessionM ); } + /** + * Reserve a unique orchestrator identity ({prefix}-orchestrator-N) for a worktree-based orchestrator. + * Unlike worker sessions, orchestrator IDs are assigned locally without remote branch checks. + */ + function reserveNextOrchestratorIdentity( + project: ProjectConfig, + sessionsDir: string, + ): { num: number; sessionId: string; tmuxName: string | undefined } { + const orchestratorPrefix = `${project.sessionPrefix}-orchestrator`; + const usedNumbers = new Set(); + + const orchestratorPattern = new RegExp(`^${escapeRegex(orchestratorPrefix)}-(\\d+)$`); + for (const sessionName of [ + ...listMetadata(sessionsDir), + ...listArchivedSessionIds(sessionsDir), + ]) { + const match = sessionName.match(orchestratorPattern); + if (match) { + const parsed = Number.parseInt(match[1], 10); + if (!Number.isNaN(parsed)) usedNumbers.add(parsed); + } + } + + // Build worker-ID patterns for all other projects. If another project has + // sessionPrefix === orchestratorPrefix (e.g. project B has prefix "app-orchestrator"), + // then its workers are named "app-orchestrator-1", "app-orchestrator-2", etc. — which + // would collide with our orchestrator IDs. Detect this impossible configuration early. + for (const [otherProjectId, otherProject] of Object.entries(config.projects)) { + const otherPrefix = otherProject.sessionPrefix ?? otherProjectId; + if (otherPrefix === project.sessionPrefix) continue; + if (otherPrefix === orchestratorPrefix) { + // Another project's workers are "{otherPrefix}-\d+" which equals "{orchestratorPrefix}-\d+". + // Every candidate ID we would generate collides — fail immediately with a clear message. + throw new Error( + `Cannot spawn orchestrator for project "${project.sessionPrefix}": the orchestrator ID prefix "${orchestratorPrefix}" ` + + `conflicts with the session prefix of project "${otherProjectId}" ("${otherPrefix}"). ` + + `Rename one of the project sessionPrefix values to avoid this overlap.`, + ); + } + } + + let num = 1; + for (let attempts = 0; attempts < 10_000; attempts++) { + if (!usedNumbers.has(num)) { + const sessionId = `${orchestratorPrefix}-${num}`; + const tmuxName = config.configPath + ? generateTmuxName(config.configPath, orchestratorPrefix, num) + : undefined; + if (reserveSessionId(sessionsDir, sessionId)) { + return { num, sessionId, tmuxName }; + } + } + num += 1; + } + + throw new Error( + `Failed to reserve orchestrator session ID after 10000 attempts (prefix: ${orchestratorPrefix})`, + ); + } + /** Resolve which plugins to use for a project. */ function resolvePlugins(project: ProjectConfig, agentName?: string) { const runtime = registry.get("runtime", project.runtime ?? config.defaults.runtime); @@ -725,7 +810,7 @@ export function createSessionManager(deps: SessionManagerDeps): OpenCodeSessionM metadata: Record, ) { return resolveAgentSelection({ - role: resolveSessionRole(sessionId, metadata), + role: resolveSessionRole(sessionId, metadata, project.sessionPrefix), project, defaults: config.defaults, persistedAgent: metadata["agent"], @@ -766,11 +851,11 @@ export function createSessionManager(deps: SessionManagerDeps): OpenCodeSessionM modifiedAt = undefined; } - const repaired = repairSingleSessionMetadataOnRead(sessionsDir, { - sessionName: sessionId, - raw, - modifiedAt, - }); + const repaired = repairSingleSessionMetadataOnRead( + sessionsDir, + { sessionName: sessionId, raw, modifiedAt }, + project.sessionPrefix, + ); return { raw: repaired.raw, sessionsDir, project, projectId }; } @@ -1248,29 +1333,88 @@ export function createSessionManager(deps: SessionManagerDeps): OpenCodeSessionM throw new Error(`Agent plugin '${selection.agentName}' not found`); } - const sessionId = `${project.sessionPrefix}-orchestrator`; const orchestratorSessionStrategy = normalizeOrchestratorSessionStrategy( project.orchestratorSessionStrategy, ); - // Generate tmux name if using new architecture - let tmuxName: string | undefined; - if (config.configPath) { - const hash = generateConfigHash(config.configPath); - tmuxName = `${hash}-${sessionId}`; - } - // Get the sessions directory for this project const sessionsDir = getProjectSessionsDir(project); - // Validate and store .origin file + // Validate and store .origin file before reserving any identity so that + // a validation failure does not leave an orphaned metadata entry. if (config.configPath) { validateAndStoreOrigin(config.configPath, project.path); } + // Reserve a new unique orchestrator identity (e.g. {prefix}-orchestrator-1, -2, …). + // Each spawnOrchestrator call gets its own numbered session and isolated worktree. + const identity = reserveNextOrchestratorIdentity(project, sessionsDir); + const sessionId = identity.sessionId; + const tmuxName = identity.tmuxName; + + // Each orchestrator gets an isolated worktree on its own branch. + const branch = `orchestrator/${sessionId}`; + + if (!plugins.workspace) { + try { + deleteMetadata(sessionsDir, sessionId, false); + } catch { + /* best effort */ + } + throw new Error( + `spawnOrchestrator requires a workspace plugin but none is configured for project '${orchestratorConfig.projectId}'`, + ); + } + + let workspacePath: string; + try { + const wsInfo = await plugins.workspace.create({ + projectId: orchestratorConfig.projectId, + project, + sessionId, + branch, + }); + workspacePath = wsInfo.path; + } catch (err) { + try { + deleteMetadata(sessionsDir, sessionId, false); + } catch { + /* best effort */ + } + throw err; + } + + // Helper: undo worktree + metadata if anything between workspace creation + // and a fully-written metadata record fails. + const cleanupWorktreeAndMetadata = async (promptFile?: string): Promise => { + try { + // plugins.workspace is guaranteed non-null here: we threw above if it was null + await plugins.workspace!.destroy(workspacePath); + } catch { + /* best effort */ + } + try { + deleteMetadata(sessionsDir, sessionId, false); + } catch { + /* best effort */ + } + if (promptFile) { + try { + unlinkSync(promptFile); + } catch { + /* best effort */ + } + } + }; + // Setup agent hooks for automatic metadata updates - if (plugins.agent.setupWorkspaceHooks) { - await plugins.agent.setupWorkspaceHooks(project.path, { dataDir: sessionsDir }); + try { + if (plugins.agent.setupWorkspaceHooks) { + await plugins.agent.setupWorkspaceHooks(workspacePath, { dataDir: sessionsDir }); + } + } catch (err) { + await cleanupWorktreeAndMetadata(); + throw err; } // Write system prompt to a file to avoid shell/tmux truncation. @@ -1278,93 +1422,38 @@ export function createSessionManager(deps: SessionManagerDeps): OpenCodeSessionM // via tmux send-keys or paste-buffer. File-based approach is reliable. let systemPromptFile: string | undefined; if (orchestratorConfig.systemPrompt) { - const baseDir = getProjectBaseDir(config.configPath, project.path); - mkdirSync(baseDir, { recursive: true }); - systemPromptFile = join(baseDir, "orchestrator-prompt.md"); - writeFileSync(systemPromptFile, orchestratorConfig.systemPrompt, "utf-8"); - } - - const existingRaw = readMetadataRaw(sessionsDir, sessionId); - const existingOrchestrator = existingRaw?.["runtimeHandle"] - ? metadataToSession(sessionId, existingRaw, orchestratorConfig.projectId) - : null; - if (existingOrchestrator?.runtimeHandle) { - const existingAlive = await plugins.runtime - .isAlive(existingOrchestrator.runtimeHandle) - .catch(() => false); - if (existingAlive && orchestratorSessionStrategy === "reuse") { - const persistedRaw = readMetadataRaw(sessionsDir, sessionId); - if (persistedRaw?.["runtimeHandle"]) { - const persisted = metadataToSession( - sessionId, - persistedRaw, - orchestratorConfig.projectId, - ); - persisted.metadata["orchestratorSessionReused"] = "true"; - return persisted; - } - await plugins.runtime.destroy(existingOrchestrator.runtimeHandle).catch(() => undefined); - deleteMetadata(sessionsDir, sessionId, false); - } - if (existingAlive && orchestratorSessionStrategy !== "reuse") { - await plugins.runtime.destroy(existingOrchestrator.runtimeHandle).catch(() => undefined); - // Destroy runtime and delete metadata without archive for ignore strategy - deleteMetadata(sessionsDir, sessionId, false); - } - // For dead runtime, delete metadata so reserveSessionId can succeed: - // - With reuse strategy + opencode: archive to preserve opencodeSessionId for reuse lookup - // - With non-reuse strategy: delete without archive to respawn fresh - if (!existingAlive) { - deleteMetadata(sessionsDir, sessionId, orchestratorSessionStrategy === "reuse"); + try { + const baseDir = getProjectBaseDir(config.configPath, project.path); + mkdirSync(baseDir, { recursive: true }); + systemPromptFile = join(baseDir, `orchestrator-prompt-${sessionId}.md`); + writeFileSync(systemPromptFile, orchestratorConfig.systemPrompt, "utf-8"); + } catch (err) { + await cleanupWorktreeAndMetadata(systemPromptFile); + throw err; } } - // Atomically reserve the session ID before creating any resources. - // This prevents race conditions where concurrent spawnOrchestrator calls - // both see no existing session and proceed to create duplicate runtimes. - let reserved = reserveSessionId(sessionsDir, sessionId); - if (!reserved) { - // Reservation failed - another process reserved it first. - // Check if the session now exists and is alive. - const concurrentRaw = readMetadataRaw(sessionsDir, sessionId); - const concurrentSession = concurrentRaw?.["runtimeHandle"] - ? metadataToSession(sessionId, concurrentRaw, orchestratorConfig.projectId) - : null; - if (concurrentSession?.runtimeHandle) { - const concurrentAlive = await plugins.runtime - .isAlive(concurrentSession.runtimeHandle) - .catch(() => false); - if (concurrentAlive && orchestratorSessionStrategy === "reuse") { - concurrentSession.metadata["orchestratorSessionReused"] = "true"; - return concurrentSession; - } - if (!concurrentAlive) { - deleteMetadata(sessionsDir, sessionId, orchestratorSessionStrategy === "reuse"); - reserved = reserveSessionId(sessionsDir, sessionId); - } - } else { - reserved = reserveSessionId(sessionsDir, sessionId); + let reusableOpenCodeSessionId: string | undefined; + try { + reusableOpenCodeSessionId = + plugins.agent.name === "opencode" && orchestratorSessionStrategy === "reuse" + ? await resolveOpenCodeSessionReuse({ + sessionsDir, + criteria: { sessionId }, + strategy: "reuse", + }) + : undefined; + if (plugins.agent.name === "opencode" && orchestratorSessionStrategy === "delete") { + await resolveOpenCodeSessionReuse({ + sessionsDir, + criteria: { sessionId }, + strategy: "delete", + includeTitleDiscoveryForSessionId: true, + }); } - if (!reserved) { - throw new Error(`Session ${sessionId} already exists but is not in a reusable state`); - } - } - - const reusableOpenCodeSessionId = - plugins.agent.name === "opencode" && orchestratorSessionStrategy === "reuse" - ? await resolveOpenCodeSessionReuse({ - sessionsDir, - criteria: { sessionId }, - strategy: "reuse", - }) - : undefined; - if (plugins.agent.name === "opencode" && orchestratorSessionStrategy === "delete") { - await resolveOpenCodeSessionReuse({ - sessionsDir, - criteria: { sessionId }, - strategy: "delete", - includeTitleDiscoveryForSessionId: true, - }); + } catch (err) { + await cleanupWorktreeAndMetadata(systemPromptFile); + throw err; } // Get agent launch config — uses systemPromptFile, no issue/tracker interaction. @@ -1388,22 +1477,29 @@ export function createSessionManager(deps: SessionManagerDeps): OpenCodeSessionM const launchCommand = plugins.agent.getLaunchCommand(agentLaunchConfig); const environment = plugins.agent.getEnvironment(agentLaunchConfig); - const handle = await plugins.runtime.create({ - sessionId: tmuxName ?? sessionId, - workspacePath: project.path, - launchCommand, - environment: { - ...environment, - AO_SESSION: sessionId, - AO_DATA_DIR: sessionsDir, - AO_SESSION_NAME: sessionId, - ...(tmuxName && { AO_TMUX_NAME: tmuxName }), - AO_CALLER_TYPE: "orchestrator", - AO_PROJECT_ID: orchestratorConfig.projectId, - AO_CONFIG_PATH: config.configPath, - ...(config.port !== undefined && config.port !== null && { AO_PORT: String(config.port) }), - }, - }); + // Create runtime — clean up worktree and metadata on failure + let handle: RuntimeHandle; + try { + handle = await plugins.runtime.create({ + sessionId: tmuxName ?? sessionId, + workspacePath, + launchCommand, + environment: { + ...environment, + AO_SESSION: sessionId, + AO_DATA_DIR: sessionsDir, + AO_SESSION_NAME: sessionId, + ...(tmuxName && { AO_TMUX_NAME: tmuxName }), + AO_CALLER_TYPE: "orchestrator", + AO_PROJECT_ID: orchestratorConfig.projectId, + AO_CONFIG_PATH: config.configPath, + ...(config.port !== undefined && config.port !== null && { AO_PORT: String(config.port) }), + }, + }); + } catch (err) { + await cleanupWorktreeAndMetadata(systemPromptFile); + throw err; + } // Write metadata and run post-launch setup const session: Session = { @@ -1411,10 +1507,10 @@ export function createSessionManager(deps: SessionManagerDeps): OpenCodeSessionM projectId: orchestratorConfig.projectId, status: "working", activity: "active", - branch: project.defaultBranch, + branch, issueId: null, pr: null, - workspacePath: project.path, + workspacePath, runtimeHandle: handle, agentInfo: null, createdAt: new Date(), @@ -1426,8 +1522,8 @@ export function createSessionManager(deps: SessionManagerDeps): OpenCodeSessionM try { writeMetadata(sessionsDir, sessionId, { - worktree: project.path, - branch: project.defaultBranch, + worktree: workspacePath, + branch, status: "working", role: "orchestrator", tmuxName, @@ -1466,11 +1562,7 @@ export function createSessionManager(deps: SessionManagerDeps): OpenCodeSessionM } catch { /* best effort */ } - try { - deleteMetadata(sessionsDir, sessionId, false); - } catch { - /* best effort */ - } + await cleanupWorktreeAndMetadata(systemPromptFile); throw err; } @@ -1561,11 +1653,11 @@ export function createSessionManager(deps: SessionManagerDeps): OpenCodeSessionM // If stat fails, timestamps will fall back to current time } - const repaired = repairSingleSessionMetadataOnRead(sessionsDir, { - sessionName: sessionId, - raw, - modifiedAt, - }); + const repaired = repairSingleSessionMetadataOnRead( + sessionsDir, + { sessionName: sessionId, raw, modifiedAt }, + project.sessionPrefix, + ); const session = metadataToSession(sessionId, repaired.raw, projectId, createdAt, modifiedAt); @@ -1816,8 +1908,7 @@ export function createSessionManager(deps: SessionManagerDeps): OpenCodeSessionM async function send(sessionId: SessionId, message: string): Promise { const { raw, sessionsDir, project } = requireSessionRecord(sessionId); const pause = getProjectPause(project); - const orchestratorId = `${project.sessionPrefix}-orchestrator`; - if (pause && sessionId !== orchestratorId) { + if (pause && !isOrchestratorSessionRecord(sessionId, raw, project.sessionPrefix)) { throw new Error( `Project is paused due to model rate limit until ${pause.until.toISOString()} (${pause.reason}; source: ${pause.sourceSessionId})`, ); @@ -2109,7 +2200,7 @@ export function createSessionManager(deps: SessionManagerDeps): OpenCodeSessionM if (!reference) throw new Error("PR reference is required"); const { raw, sessionsDir, project, projectId } = requireSessionRecord(sessionId); - if (isOrchestratorSessionRecord(sessionId, raw)) { + if (isOrchestratorSessionRecord(sessionId, raw, project.sessionPrefix)) { throw new Error(`Session ${sessionId} is an orchestrator session and cannot claim PRs`); } @@ -2136,7 +2227,7 @@ export function createSessionManager(deps: SessionManagerDeps): OpenCodeSessionM ); for (const { sessionName, raw: otherRaw } of activeRecords) { - if (!otherRaw || isOrchestratorSessionRecord(sessionName, otherRaw)) continue; + if (!otherRaw || isOrchestratorSessionRecord(sessionName, otherRaw, project.sessionPrefix)) continue; const samePr = otherRaw["pr"] === pr.url; const sameBranch = diff --git a/packages/core/src/types.ts b/packages/core/src/types.ts index f42ec1c63..1d6d0eee0 100644 --- a/packages/core/src/types.ts +++ b/packages/core/src/types.ts @@ -189,11 +189,37 @@ export interface Session { metadata: Record; } -export function isOrchestratorSession(session: { - id: SessionId; - metadata?: Record; -}): boolean { - return session.metadata?.["role"] === "orchestrator" || session.id.endsWith("-orchestrator"); +export function isOrchestratorSession( + session: { id: SessionId; metadata?: Record }, + sessionPrefix?: string, + allSessionPrefixes?: string[], +): boolean { + if (session.metadata?.["role"] === "orchestrator" || session.id.endsWith("-orchestrator")) { + return true; + } + if (!sessionPrefix) { + return false; + } + const escaped = sessionPrefix.replace(/[.*+?^${}()|[\]\\]/g, "\\$&"); + if (!new RegExp(`^${escaped}-orchestrator-\\d+$`).test(session.id)) { + return false; + } + // Guard against cross-project false positives: if the session ID is a plain + // numbered worker for any other known prefix (e.g. prefix "app-orchestrator" + // matches "app-orchestrator-1" as a worker), it is not an orchestrator. + if (allSessionPrefixes) { + for (const prefix of allSessionPrefixes) { + if (prefix === sessionPrefix) continue; + if ( + new RegExp( + `^${prefix.replace(/[.*+?^${}()|[\]\\]/g, "\\$&")}-\\d+$`, + ).test(session.id) + ) { + return false; + } + } + } + return true; } /** Config for creating a new session */ diff --git a/packages/web/src/app/api/sessions/route.ts b/packages/web/src/app/api/sessions/route.ts index 3485e7231..7fb406e90 100644 --- a/packages/web/src/app/api/sessions/route.ts +++ b/packages/web/src/app/api/sessions/route.ts @@ -75,7 +75,17 @@ export async function GET(request: Request) { const allSessions = requestedProjectId ? await sessionManager.list() : coreSessions; - let workerSessions = visibleSessions.filter((session) => !isOrchestratorSession(session)); + const allSessionPrefixes = Object.entries(config.projects).map( + ([projectId, p]) => p.sessionPrefix ?? projectId, + ); + let workerSessions = visibleSessions.filter( + (session) => + !isOrchestratorSession( + session, + config.projects[session.projectId]?.sessionPrefix ?? session.projectId, + allSessionPrefixes, + ), + ); // Convert to dashboard format let dashboardSessions = workerSessions.map(sessionToDashboard); @@ -134,7 +144,7 @@ export async function GET(request: Request) { stats: computeStats(dashboardSessions), orchestratorId, orchestrators, - globalPause: resolveGlobalPause(allSessions), + globalPause: resolveGlobalPause(allSessions, config.projects), }, { status: 200 }, correlationId, diff --git a/packages/web/src/app/sessions/[id]/page.tsx b/packages/web/src/app/sessions/[id]/page.tsx index fc8c33b00..cf0bf2fe3 100644 --- a/packages/web/src/app/sessions/[id]/page.tsx +++ b/packages/web/src/app/sessions/[id]/page.tsx @@ -6,6 +6,7 @@ import { isOrchestratorSession } from "@composio/ao-core/types"; import { SessionDetail } from "@/components/SessionDetail"; import { type DashboardSession, type ActivityState, getAttentionLevel, type AttentionLevel } from "@/lib/types"; import { activityIcon } from "@/lib/activity-icons"; +import type { ProjectInfo } from "@/lib/project-name"; import { useSSESessionActivity } from "@/hooks/useSSESessionActivity"; function truncate(s: string, max: number): string { @@ -15,12 +16,14 @@ function truncate(s: string, max: number): string { /** Build a descriptive tab title from session data. */ function buildSessionTitle( session: DashboardSession, + prefixByProject: Map, activityOverride?: ActivityState | null, ): string { const id = session.id; const activity = activityOverride !== undefined ? activityOverride : session.activity; const emoji = activity ? (activityIcon[activity] ?? "") : ""; - const isOrchestrator = isOrchestratorSession(session); + const allPrefixes = [...prefixByProject.values()]; + const isOrchestrator = isOrchestratorSession(session, prefixByProject.get(session.projectId), allPrefixes); let detail: string; @@ -61,11 +64,35 @@ export default function SessionPage() { const [projectOrchestratorId, setProjectOrchestratorId] = useState(undefined); const [loading, setLoading] = useState(true); const [error, setError] = useState(null); + const [prefixByProject, setPrefixByProject] = useState>(new Map()); const sessionProjectId = session?.projectId ?? null; - const sessionIsOrchestrator = session ? isOrchestratorSession(session) : false; + const allPrefixes = [...prefixByProject.values()]; + const sessionIsOrchestrator = session + ? isOrchestratorSession(session, prefixByProject.get(session.projectId), allPrefixes) + : false; const sessionProjectIdRef = useRef(null); const sessionIsOrchestratorRef = useRef(false); const resolvedProjectSessionsKeyRef = useRef(null); + const prefixByProjectRef = useRef>(new Map()); + + // Keep prefixByProjectRef in sync so fetchProjectSessions (stable [] dep) reads latest map + useEffect(() => { + prefixByProjectRef.current = prefixByProject; + }, [prefixByProject]); + + // Fetch project prefix map once on mount so isOrchestratorSession can use the correct prefix + useEffect(() => { + fetch("/api/projects") + .then((res) => res.ok ? res.json() : null) + .then((data: { projects?: ProjectInfo[] } | null) => { + if (data?.projects) { + setPrefixByProject( + new Map(data.projects.map((p) => [p.id, p.sessionPrefix ?? p.id])), + ); + } + }) + .catch(() => {/* non-critical — falls back to role metadata check */}); + }, []); // Subscribe to SSE for real-time activity updates (title emoji) const sseActivity = useSSESessionActivity(id); @@ -73,11 +100,11 @@ export default function SessionPage() { // Update document title based on session data + SSE activity override useEffect(() => { if (session) { - document.title = buildSessionTitle(session, sseActivity?.activity); + document.title = buildSessionTitle(session, prefixByProject, sseActivity?.activity); } else { document.title = `${id} | Session Detail`; } - }, [session, id, sseActivity]); + }, [session, id, prefixByProject, sseActivity]); useEffect(() => { sessionProjectIdRef.current = sessionProjectId; @@ -141,8 +168,9 @@ export default function SessionPage() { working: 0, done: 0, }; + const allPrefixes = [...prefixByProjectRef.current.values()]; for (const s of sessions) { - if (!isOrchestratorSession(s)) { + if (!isOrchestratorSession(s, prefixByProjectRef.current.get(s.projectId), allPrefixes)) { counts[getAttentionLevel(s) as AttentionLevel]++; } } diff --git a/packages/web/src/components/ProjectSidebar.tsx b/packages/web/src/components/ProjectSidebar.tsx index f2f5b5fed..036e73d75 100644 --- a/packages/web/src/components/ProjectSidebar.tsx +++ b/packages/web/src/components/ProjectSidebar.tsx @@ -21,8 +21,14 @@ interface ProjectSidebarProps { type ProjectHealth = "red" | "yellow" | "green" | "gray"; -function computeProjectHealth(sessions: DashboardSession[]): ProjectHealth { - const workers = sessions.filter((s) => !isOrchestratorSession(s)); +function computeProjectHealth( + sessions: DashboardSession[], + prefixByProject: Map, + allPrefixes: string[], +): ProjectHealth { + const workers = sessions.filter( + (s) => !isOrchestratorSession(s, prefixByProject.get(s.projectId), allPrefixes), + ); if (workers.length === 0) return "gray"; for (const s of workers) { if (getAttentionLevel(s) === "respond") return "red"; @@ -131,6 +137,16 @@ function ProjectSidebarInner({ router.push(pathname + `?project=${encodeURIComponent(projectId)}`); }; + const prefixByProject = useMemo( + () => new Map(projects.map((p) => [p.id, p.sessionPrefix ?? p.id])), + [projects], + ); + + const allPrefixes = useMemo( + () => projects.map((p) => p.sessionPrefix ?? p.id), + [projects], + ); + const sessionsByProject = useMemo(() => { const map = new Map(); let totalWorkers = 0; @@ -144,7 +160,7 @@ function ProjectSidebarInner({ map.set(s.projectId, entry); } entry.all.push(s); - if (!isOrchestratorSession(s)) { + if (!isOrchestratorSession(s, prefixByProject.get(s.projectId), allPrefixes)) { entry.workers.push(s); totalWorkers++; } @@ -154,7 +170,7 @@ function ProjectSidebarInner({ } return { map, totalWorkers, needsInput, reviewLoad }; - }, [sessions]); + }, [sessions, prefixByProject, allPrefixes]); const { totalWorkers: totalWorkerSessions, needsInput: needsInputCount, reviewLoad: reviewLoadCount } = sessionsByProject; @@ -168,7 +184,7 @@ function ProjectSidebarInner({
{projects.map((project) => { const entry = sessionsByProject.map.get(project.id); - const health = entry ? computeProjectHealth(entry.all) : ("gray" as ProjectHealth); + const health = entry ? computeProjectHealth(entry.all, prefixByProject, allPrefixes) : ("gray" as ProjectHealth); const isActive = activeProjectId === project.id; const initial = project.name.charAt(0).toUpperCase(); return ( @@ -287,7 +303,7 @@ function ProjectSidebarInner({ const entry = sessionsByProject.map.get(project.id); const projectSessions = entry?.all ?? []; const workerSessions = entry?.workers ?? []; - const health = computeProjectHealth(projectSessions); + const health = computeProjectHealth(projectSessions, prefixByProject, allPrefixes); const isExpanded = expandedProjects.has(project.id); const isActive = activeProjectId === project.id; diff --git a/packages/web/src/hooks/useSessionEvents.ts b/packages/web/src/hooks/useSessionEvents.ts index 7eae5e8c7..e9b531f72 100644 --- a/packages/web/src/hooks/useSessionEvents.ts +++ b/packages/web/src/hooks/useSessionEvents.ts @@ -1,11 +1,12 @@ "use client"; import { useEffect, useReducer, useRef } from "react"; -import type { - AttentionLevel, - DashboardSession, - GlobalPauseState, - SSESnapshotEvent, +import { + getAttentionLevel, + type AttentionLevel, + type DashboardSession, + type GlobalPauseState, + type SSESnapshotEvent, } from "@/lib/types"; /** Debounce before fetching full session list after membership change. */ @@ -176,10 +177,14 @@ export function useSessionEvents( if (disposed || refreshController.signal.aborted || !updated?.sessions) return; lastRefreshAtRef.current = Date.now(); + const sseAttentionLevels = Object.fromEntries( + updated.sessions.map((s) => [s.id, getAttentionLevel(s)]), + ) as SSEAttentionMap; dispatch({ type: "reset", sessions: updated.sessions, globalPause: updated.globalPause ?? null, + sseAttentionLevels, }); }, ) diff --git a/packages/web/src/lib/dashboard-page-data.ts b/packages/web/src/lib/dashboard-page-data.ts index 830b6e168..896df0a3c 100644 --- a/packages/web/src/lib/dashboard-page-data.ts +++ b/packages/web/src/lib/dashboard-page-data.ts @@ -58,7 +58,7 @@ export const getDashboardPageData = cache(async function getDashboardPageData(pr const { config, registry, sessionManager } = await getServices(); const allSessions = await sessionManager.list(); - pageData.globalPause = resolveGlobalPause(allSessions); + pageData.globalPause = resolveGlobalPause(allSessions, config.projects); const visibleSessions = filterProjectSessions(allSessions, projectFilter, config.projects); pageData.orchestrators = listDashboardOrchestrators(visibleSessions, config.projects); diff --git a/packages/web/src/lib/global-pause.ts b/packages/web/src/lib/global-pause.ts index 2811aada6..2c683bc1b 100644 --- a/packages/web/src/lib/global-pause.ts +++ b/packages/web/src/lib/global-pause.ts @@ -13,19 +13,27 @@ export interface GlobalPauseState { } export function resolveGlobalPause( - sessions: Array<{ id: string; metadata: Record }>, + sessions: Array<{ id: string; projectId: string; metadata: Record }>, + projects: Record, ): GlobalPauseState | null { + const allSessionPrefixes = Object.entries(projects).map( + ([projectId, p]) => p.sessionPrefix ?? projectId, + ); + let best: { pausedUntil: string; reason: string; sourceSessionId: string | null } | null = null; for (const session of sessions) { - if (!isOrchestratorSession(session)) continue; + const sessionPrefix = projects[session.projectId]?.sessionPrefix ?? session.projectId; + if (!isOrchestratorSession(session, sessionPrefix, allSessionPrefixes)) continue; const parsed = parsePauseUntil(session.metadata[GLOBAL_PAUSE_UNTIL_KEY]); if (!parsed || parsed.getTime() <= Date.now()) continue; - return { - pausedUntil: parsed.toISOString(), - reason: session.metadata[GLOBAL_PAUSE_REASON_KEY] ?? "Model rate limit reached", - sourceSessionId: session.metadata[GLOBAL_PAUSE_SOURCE_KEY] ?? null, - }; + if (!best || parsed.getTime() > new Date(best.pausedUntil).getTime()) { + best = { + pausedUntil: parsed.toISOString(), + reason: session.metadata[GLOBAL_PAUSE_REASON_KEY] ?? "Model rate limit reached", + sourceSessionId: session.metadata[GLOBAL_PAUSE_SOURCE_KEY] ?? null, + }; + } } - return null; + return best; } diff --git a/packages/web/src/lib/project-name.ts b/packages/web/src/lib/project-name.ts index d6fdae75a..b410ec102 100644 --- a/packages/web/src/lib/project-name.ts +++ b/packages/web/src/lib/project-name.ts @@ -4,6 +4,7 @@ import { loadConfig } from "@composio/ao-core"; export interface ProjectInfo { id: string; name: string; + sessionPrefix?: string; } export const getProjectName = cache((): string => { @@ -37,6 +38,7 @@ export const getAllProjects = cache((): ProjectInfo[] => { return Object.entries(config.projects).map(([id, project]) => ({ id, name: project.name ?? id, + sessionPrefix: project.sessionPrefix ?? id, })); } catch { return []; diff --git a/packages/web/src/lib/project-utils.ts b/packages/web/src/lib/project-utils.ts index 179d00eec..1807034a9 100644 --- a/packages/web/src/lib/project-utils.ts +++ b/packages/web/src/lib/project-utils.ts @@ -44,6 +44,16 @@ export function filterWorkerSessions( projectFilter: string | null | undefined, projects: Record, ): T[] { - const workers = sessions.filter((s) => !isOrchestratorSession(s)); + const allSessionPrefixes = Object.entries(projects).map( + ([projectId, p]) => p.sessionPrefix ?? projectId, + ); + const workers = sessions.filter( + (s) => + !isOrchestratorSession( + s, + projects[s.projectId]?.sessionPrefix ?? s.projectId, + allSessionPrefixes, + ), + ); return filterProjectSessions(workers, projectFilter, projects); } diff --git a/packages/web/src/lib/serialize.ts b/packages/web/src/lib/serialize.ts index 6cce47e10..70fafca6a 100644 --- a/packages/web/src/lib/serialize.ts +++ b/packages/web/src/lib/serialize.ts @@ -73,8 +73,17 @@ export function listDashboardOrchestrators( sessions: Session[], projects: Record, ): DashboardOrchestratorLink[] { + const allSessionPrefixes = Object.entries(projects).map( + ([projectId, p]) => p.sessionPrefix ?? projectId, + ); return sessions - .filter((session) => isOrchestratorSession(session)) + .filter((session) => + isOrchestratorSession( + session, + projects[session.projectId]?.sessionPrefix ?? session.projectId, + allSessionPrefixes, + ), + ) .map((session) => ({ id: session.id, projectId: session.projectId, diff --git a/packages/web/src/lib/services.ts b/packages/web/src/lib/services.ts index e6a274b92..31897c277 100644 --- a/packages/web/src/lib/services.ts +++ b/packages/web/src/lib/services.ts @@ -225,8 +225,16 @@ export async function pollBacklog(): Promise { // Detect reopened issues: open state + agent:done label → relabel as agent:backlog await relabelReopenedIssues(config, registry); + const allSessionPrefixes = Object.entries(config.projects).map( + ([id, p]) => p.sessionPrefix ?? id, + ); const workerSessions = allSessions.filter( - (session) => !isOrchestratorSession(session) && !TERMINAL_STATUSES.has(session.status), + (session) => + !isOrchestratorSession( + session, + config.projects[session.projectId]?.sessionPrefix ?? session.projectId, + allSessionPrefixes, + ) && !TERMINAL_STATUSES.has(session.status), ); const activeIssueIds = new Set( workerSessions