fix: prevent orchestrator sessions from owning PRs (#432)
* fix: repair orchestrator PR metadata and session branch allocation Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-opencode) Co-authored-by: Sisyphus <clio-agent@sisyphuslabs.ai> * fix: harden orchestrator prompt delegation rules Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-opencode) Co-authored-by: Sisyphus <clio-agent@sisyphuslabs.ai> * fix: hide orchestrator PR ownership in status Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-opencode) Co-authored-by: Sisyphus <clio-agent@sisyphuslabs.ai> * fix: restore session suffix parsing for branch allocation Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-opencode) Co-authored-by: Sisyphus <clio-agent@sisyphuslabs.ai> * fix: avoid dynamic delete in metadata repair path Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-opencode) Co-authored-by: Sisyphus <clio-agent@sisyphuslabs.ai> * fix: reduce read-time repair overhead and preserve activity timestamps Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-opencode) Co-authored-by: Sisyphus <clio-agent@sisyphuslabs.ai> * refactor: share orchestrator read-repair logic Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-opencode) Co-authored-by: Sisyphus <clio-agent@sisyphuslabs.ai> --------- Co-authored-by: Sisyphus <clio-agent@sisyphuslabs.ai>
This commit is contained in:
Harsh Batheja
GitHub
parent
9794291319
commit
c1e8c1e839
|
|
@ -692,4 +692,42 @@ describe("status command", () => {
|
|||
const parsed = JSON.parse(jsonCalls);
|
||||
expect(parsed[0].activity).toBe("exited");
|
||||
});
|
||||
|
||||
it("suppresses orchestrator PR ownership in status output", async () => {
|
||||
writeFileSync(
|
||||
join(sessionsDir, "app-orchestrator"),
|
||||
[
|
||||
"worktree=/tmp/wt",
|
||||
"branch=main",
|
||||
"status=working",
|
||||
"role=orchestrator",
|
||||
"pr=https://github.com/org/repo/pull/77",
|
||||
].join("\n"),
|
||||
);
|
||||
|
||||
mockTmux.mockImplementation(async (...args: string[]) => {
|
||||
if (args[0] === "list-sessions") return "app-orchestrator";
|
||||
if (args[0] === "display-message") return String(Math.floor(Date.now() / 1000));
|
||||
return null;
|
||||
});
|
||||
mockGit.mockResolvedValue("main");
|
||||
mockDetectPR.mockResolvedValue({
|
||||
number: 77,
|
||||
url: "https://github.com/org/repo/pull/77",
|
||||
title: "Orchestrator should not own this",
|
||||
owner: "org",
|
||||
repo: "repo",
|
||||
branch: "main",
|
||||
baseBranch: "main",
|
||||
isDraft: false,
|
||||
});
|
||||
|
||||
await program.parseAsync(["node", "test", "status", "--json"]);
|
||||
|
||||
const parsed = JSON.parse(consoleSpy.mock.calls.map((c) => c[0]).join(""));
|
||||
expect(parsed[0].name).toBe("app-orchestrator");
|
||||
expect(parsed[0].pr).toBeNull();
|
||||
expect(parsed[0].prNumber).toBeNull();
|
||||
expect(mockDetectPR).not.toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
|
|
|
|||
|
|
@ -42,16 +42,21 @@ interface SessionInfo {
|
|||
activity: ActivityState | null;
|
||||
}
|
||||
|
||||
function isOrchestratorSession(session: Session): boolean {
|
||||
return session.metadata["role"] === "orchestrator" || session.id.endsWith("-orchestrator");
|
||||
}
|
||||
|
||||
async function gatherSessionInfo(
|
||||
session: Session,
|
||||
agent: Agent,
|
||||
scm: SCM,
|
||||
projectConfig: ReturnType<typeof loadConfig>,
|
||||
): Promise<SessionInfo> {
|
||||
const suppressPROwnership = isOrchestratorSession(session);
|
||||
let branch = session.branch;
|
||||
const status = session.status;
|
||||
const summary = session.metadata["summary"] ?? null;
|
||||
const prUrl = session.metadata["pr"] ?? null;
|
||||
const prUrl = suppressPROwnership ? null : (session.metadata["pr"] ?? null);
|
||||
const issue = session.issueId;
|
||||
|
||||
// Get live branch from worktree if available
|
||||
|
|
@ -91,7 +96,7 @@ async function gatherSessionInfo(
|
|||
}
|
||||
}
|
||||
|
||||
if (branch) {
|
||||
if (branch && !suppressPROwnership) {
|
||||
try {
|
||||
const project = projectConfig.projects[session.projectId];
|
||||
if (project) {
|
||||
|
|
|
|||
|
|
@ -0,0 +1,57 @@
|
|||
import { describe, expect, it } from "vitest";
|
||||
import { generateOrchestratorPrompt } from "../orchestrator-prompt.js";
|
||||
import type { OrchestratorConfig } from "../types.js";
|
||||
|
||||
const config: OrchestratorConfig = {
|
||||
configPath: "/tmp/agent-orchestrator.yaml",
|
||||
port: 3000,
|
||||
defaults: {
|
||||
runtime: "tmux",
|
||||
agent: "claude-code",
|
||||
workspace: "worktree",
|
||||
notifiers: ["desktop"],
|
||||
},
|
||||
projects: {
|
||||
"my-app": {
|
||||
name: "My App",
|
||||
repo: "org/my-app",
|
||||
path: "/tmp/my-app",
|
||||
defaultBranch: "main",
|
||||
sessionPrefix: "app",
|
||||
},
|
||||
},
|
||||
notifiers: {},
|
||||
notificationRouting: {
|
||||
urgent: ["desktop"],
|
||||
action: ["desktop"],
|
||||
warning: [],
|
||||
info: [],
|
||||
},
|
||||
reactions: {},
|
||||
readyThresholdMs: 300_000,
|
||||
};
|
||||
|
||||
describe("generateOrchestratorPrompt", () => {
|
||||
it("requires read-only investigation from the orchestrator session", () => {
|
||||
const prompt = generateOrchestratorPrompt({
|
||||
config,
|
||||
projectId: "my-app",
|
||||
project: config.projects["my-app"]!,
|
||||
});
|
||||
|
||||
expect(prompt).toContain("Investigations from the orchestrator session are **read-only**");
|
||||
expect(prompt).toContain("do not edit repository files or implement fixes");
|
||||
});
|
||||
|
||||
it("pushes implementation and PR claiming into worker sessions", () => {
|
||||
const prompt = generateOrchestratorPrompt({
|
||||
config,
|
||||
projectId: "my-app",
|
||||
project: config.projects["my-app"]!,
|
||||
});
|
||||
|
||||
expect(prompt).toContain("must be delegated to a **worker session**");
|
||||
expect(prompt).toContain("Never claim a PR into `app-orchestrator`");
|
||||
expect(prompt).toContain("Delegate implementation, test execution, or PR claiming");
|
||||
});
|
||||
});
|
||||
|
|
@ -1,5 +1,13 @@
|
|||
import { describe, it, expect, beforeEach, afterEach, vi } from "vitest";
|
||||
import { chmodSync, mkdirSync, readFileSync, rmSync, writeFileSync, existsSync } from "node:fs";
|
||||
import {
|
||||
chmodSync,
|
||||
mkdirSync,
|
||||
readFileSync,
|
||||
rmSync,
|
||||
writeFileSync,
|
||||
existsSync,
|
||||
utimesSync,
|
||||
} from "node:fs";
|
||||
import { join } from "node:path";
|
||||
import { homedir, tmpdir } from "node:os";
|
||||
import { randomUUID } from "node:crypto";
|
||||
|
|
@ -100,6 +108,32 @@ function installMockOpencodeWithNotFoundDelete(sessionListJson: string): string
|
|||
return binDir;
|
||||
}
|
||||
|
||||
function installMockGit(remoteBranches: string[]): string {
|
||||
const binDir = join(tmpDir, "mock-git-bin");
|
||||
mkdirSync(binDir, { recursive: true });
|
||||
const scriptPath = join(binDir, "git");
|
||||
const refs = remoteBranches
|
||||
.map((branch) => `deadbeef\trefs/heads/${branch}`)
|
||||
.join("\\n")
|
||||
.replace(/'/g, "'\\''");
|
||||
writeFileSync(
|
||||
scriptPath,
|
||||
[
|
||||
"#!/usr/bin/env bash",
|
||||
"set -euo pipefail",
|
||||
'if [[ "$1" == "ls-remote" && "$2" == "--heads" && "$3" == "origin" ]]; then',
|
||||
` printf '%b\\n' '${refs}'`,
|
||||
" exit 0",
|
||||
"fi",
|
||||
"exit 1",
|
||||
"",
|
||||
].join("\n"),
|
||||
"utf-8",
|
||||
);
|
||||
chmodSync(scriptPath, 0o755);
|
||||
return binDir;
|
||||
}
|
||||
|
||||
beforeEach(() => {
|
||||
originalPath = process.env.PATH;
|
||||
tmpDir = join(tmpdir(), `ao-test-session-mgr-${randomUUID()}`);
|
||||
|
|
@ -365,6 +399,32 @@ describe("spawn", () => {
|
|||
expect(session.id).toBe("app-8");
|
||||
});
|
||||
|
||||
it("does not reuse a killed session branch on recreate", async () => {
|
||||
const sm = createSessionManager({ config, registry: mockRegistry });
|
||||
|
||||
const first = await sm.spawn({ projectId: "my-app" });
|
||||
expect(first.id).toBe("app-1");
|
||||
expect(first.branch).toBe("session/app-1");
|
||||
|
||||
await sm.kill(first.id);
|
||||
|
||||
const second = await sm.spawn({ projectId: "my-app" });
|
||||
expect(second.id).toBe("app-2");
|
||||
expect(second.branch).toBe("session/app-2");
|
||||
});
|
||||
|
||||
it("skips remote session branches when allocating a fresh session id", async () => {
|
||||
const mockGitBin = installMockGit(["session/app-22"]);
|
||||
process.env.PATH = `${mockGitBin}:${originalPath ?? ""}`;
|
||||
mkdirSync(config.projects["my-app"]!.path, { recursive: true });
|
||||
|
||||
const sm = createSessionManager({ config, registry: mockRegistry });
|
||||
const session = await sm.spawn({ projectId: "my-app" });
|
||||
|
||||
expect(session.id).toBe("app-23");
|
||||
expect(session.branch).toBe("session/app-23");
|
||||
});
|
||||
|
||||
it("writes metadata file", async () => {
|
||||
const sm = createSessionManager({ config, registry: mockRegistry });
|
||||
await sm.spawn({ projectId: "my-app", issueId: "INT-42" });
|
||||
|
|
@ -1080,6 +1140,32 @@ describe("list", () => {
|
|||
expect(sessions.map((s) => s.id).sort()).toEqual(["app-1", "app-2"]);
|
||||
});
|
||||
|
||||
it("preserves lastActivityAt when read-time repair rewrites metadata", async () => {
|
||||
writeMetadata(sessionsDir, "app-orchestrator", {
|
||||
worktree: config.projects["my-app"]!.path,
|
||||
branch: "main",
|
||||
status: "merged",
|
||||
project: "my-app",
|
||||
pr: "https://github.com/org/my-app/pull/42",
|
||||
runtimeHandle: JSON.stringify(makeHandle("rt-orch")),
|
||||
});
|
||||
|
||||
const oldTime = new Date("2026-01-01T00:00:00.000Z");
|
||||
utimesSync(join(sessionsDir, "app-orchestrator"), oldTime, oldTime);
|
||||
|
||||
const sm = createSessionManager({ config, registry: mockRegistry });
|
||||
const sessions = await sm.list("my-app");
|
||||
const orchestrator = sessions.find((session) => session.id === "app-orchestrator");
|
||||
|
||||
expect(orchestrator).toBeDefined();
|
||||
expect(orchestrator!.lastActivityAt.getTime()).toBe(oldTime.getTime());
|
||||
|
||||
const repaired = readMetadataRaw(sessionsDir, "app-orchestrator");
|
||||
expect(repaired!["pr"]).toBeUndefined();
|
||||
expect(repaired!["prAutoDetect"]).toBe("off");
|
||||
expect(repaired!["status"]).toBe("working");
|
||||
});
|
||||
|
||||
it("filters by project ID", async () => {
|
||||
// In hash-based architecture, each project has its own directory
|
||||
// so filtering is implicit. This test verifies list(projectId) only
|
||||
|
|
@ -3869,6 +3955,169 @@ describe("claimPR", () => {
|
|||
expect(raw!["prAutoDetect"]).toBeUndefined();
|
||||
});
|
||||
|
||||
it("consolidates ownership by disabling PR auto-detect on the previous session", async () => {
|
||||
const mockSCM = makeSCM();
|
||||
|
||||
writeMetadata(sessionsDir, "app-1", {
|
||||
worktree: "/tmp/ws-app-1",
|
||||
branch: "feat/existing-pr",
|
||||
status: "review_pending",
|
||||
project: "my-app",
|
||||
pr: "https://github.com/org/my-app/pull/42",
|
||||
runtimeHandle: JSON.stringify(makeHandle("rt-1")),
|
||||
});
|
||||
|
||||
writeMetadata(sessionsDir, "app-2", {
|
||||
worktree: "/tmp/ws-app-2",
|
||||
branch: "feat/other-work",
|
||||
status: "working",
|
||||
project: "my-app",
|
||||
runtimeHandle: JSON.stringify(makeHandle("rt-2")),
|
||||
});
|
||||
|
||||
const sm = createSessionManager({ config, registry: registryWithSCM(mockSCM) });
|
||||
const result = await sm.claimPR("app-2", "42");
|
||||
|
||||
expect(result.takenOverFrom).toEqual(["app-1"]);
|
||||
|
||||
const previous = readMetadataRaw(sessionsDir, "app-1");
|
||||
expect(previous!["pr"]).toBeUndefined();
|
||||
expect(previous!["prAutoDetect"]).toBe("off");
|
||||
expect(previous!["status"]).toBe("working");
|
||||
});
|
||||
|
||||
it("ignores legacy orchestrator metadata when claiming a PR", async () => {
|
||||
const mockSCM = makeSCM();
|
||||
|
||||
writeMetadata(sessionsDir, "app-orchestrator", {
|
||||
worktree: config.projects["my-app"]!.path,
|
||||
branch: "feat/existing-pr",
|
||||
status: "pr_open",
|
||||
project: "my-app",
|
||||
pr: "https://github.com/org/my-app/pull/42",
|
||||
runtimeHandle: JSON.stringify(makeHandle("rt-orch")),
|
||||
});
|
||||
|
||||
writeMetadata(sessionsDir, "app-2", {
|
||||
worktree: "/tmp/ws-app-2",
|
||||
branch: "feat/other-work",
|
||||
status: "working",
|
||||
project: "my-app",
|
||||
runtimeHandle: JSON.stringify(makeHandle("rt-2")),
|
||||
});
|
||||
|
||||
const sm = createSessionManager({ config, registry: registryWithSCM(mockSCM) });
|
||||
const result = await sm.claimPR("app-2", "42");
|
||||
|
||||
expect(result.takenOverFrom).toEqual([]);
|
||||
expect(readMetadataRaw(sessionsDir, "app-2")!["pr"]).toBe(
|
||||
"https://github.com/org/my-app/pull/42",
|
||||
);
|
||||
});
|
||||
|
||||
it("repairs legacy orchestrator PR metadata and stale duplicate PR attachments on read", async () => {
|
||||
writeMetadata(sessionsDir, "app-orchestrator", {
|
||||
worktree: config.projects["my-app"]!.path,
|
||||
branch: "main",
|
||||
status: "merged",
|
||||
project: "my-app",
|
||||
pr: "https://github.com/org/my-app/pull/42",
|
||||
runtimeHandle: JSON.stringify(makeHandle("rt-orch")),
|
||||
});
|
||||
|
||||
writeMetadata(sessionsDir, "app-1", {
|
||||
worktree: "/tmp/ws-app-1",
|
||||
branch: "feat/existing-pr",
|
||||
status: "review_pending",
|
||||
project: "my-app",
|
||||
pr: "https://github.com/org/my-app/pull/42",
|
||||
runtimeHandle: JSON.stringify(makeHandle("rt-1")),
|
||||
});
|
||||
|
||||
writeMetadata(sessionsDir, "app-2", {
|
||||
worktree: "/tmp/ws-app-2",
|
||||
branch: "feat/existing-pr",
|
||||
status: "pr_open",
|
||||
project: "my-app",
|
||||
pr: "https://github.com/org/my-app/pull/42",
|
||||
runtimeHandle: JSON.stringify(makeHandle("rt-2")),
|
||||
});
|
||||
|
||||
const staleTime = new Date("2026-01-01T00:00:00.000Z");
|
||||
const freshTime = new Date("2026-01-02T00:00:00.000Z");
|
||||
utimesSync(join(sessionsDir, "app-1"), staleTime, staleTime);
|
||||
utimesSync(join(sessionsDir, "app-2"), freshTime, freshTime);
|
||||
|
||||
const sm = createSessionManager({ config, registry: mockRegistry });
|
||||
const sessions = await sm.list();
|
||||
|
||||
expect(sessions).toHaveLength(3);
|
||||
|
||||
const orchestrator = readMetadataRaw(sessionsDir, "app-orchestrator");
|
||||
expect(orchestrator!["role"]).toBe("orchestrator");
|
||||
expect(orchestrator!["pr"]).toBeUndefined();
|
||||
expect(orchestrator!["prAutoDetect"]).toBe("off");
|
||||
expect(orchestrator!["status"]).toBe("working");
|
||||
|
||||
const staleWorker = readMetadataRaw(sessionsDir, "app-1");
|
||||
expect(staleWorker!["pr"]).toBeUndefined();
|
||||
expect(staleWorker!["prAutoDetect"]).toBe("off");
|
||||
expect(staleWorker!["status"]).toBe("working");
|
||||
|
||||
const activeWorker = readMetadataRaw(sessionsDir, "app-2");
|
||||
expect(activeWorker!["pr"]).toBe("https://github.com/org/my-app/pull/42");
|
||||
expect(activeWorker!["status"]).toBe("pr_open");
|
||||
});
|
||||
|
||||
it("repairs stale duplicate PR attachments before claim conflict checks", async () => {
|
||||
const mockSCM = makeSCM();
|
||||
|
||||
writeMetadata(sessionsDir, "app-1", {
|
||||
worktree: "/tmp/ws-app-1",
|
||||
branch: "feat/existing-pr",
|
||||
status: "review_pending",
|
||||
project: "my-app",
|
||||
pr: "https://github.com/org/my-app/pull/42",
|
||||
runtimeHandle: JSON.stringify(makeHandle("rt-1")),
|
||||
});
|
||||
writeMetadata(sessionsDir, "app-2", {
|
||||
worktree: "/tmp/ws-app-2",
|
||||
branch: "feat/existing-pr",
|
||||
status: "pr_open",
|
||||
project: "my-app",
|
||||
pr: "https://github.com/org/my-app/pull/42",
|
||||
runtimeHandle: JSON.stringify(makeHandle("rt-2")),
|
||||
});
|
||||
writeMetadata(sessionsDir, "app-3", {
|
||||
worktree: "/tmp/ws-app-3",
|
||||
branch: "feat/other-work",
|
||||
status: "working",
|
||||
project: "my-app",
|
||||
runtimeHandle: JSON.stringify(makeHandle("rt-3")),
|
||||
});
|
||||
|
||||
const staleTime = new Date("2026-01-01T00:00:00.000Z");
|
||||
const freshTime = new Date("2026-01-02T00:00:00.000Z");
|
||||
utimesSync(join(sessionsDir, "app-1"), staleTime, staleTime);
|
||||
utimesSync(join(sessionsDir, "app-2"), freshTime, freshTime);
|
||||
|
||||
const sm = createSessionManager({ config, registry: registryWithSCM(mockSCM) });
|
||||
const result = await sm.claimPR("app-3", "42");
|
||||
|
||||
expect(result.takenOverFrom).toEqual(["app-2"]);
|
||||
|
||||
const staleWorker = readMetadataRaw(sessionsDir, "app-1");
|
||||
expect(staleWorker!["pr"]).toBeUndefined();
|
||||
expect(staleWorker!["prAutoDetect"]).toBe("off");
|
||||
|
||||
const activeWorker = readMetadataRaw(sessionsDir, "app-2");
|
||||
expect(activeWorker!["pr"]).toBeUndefined();
|
||||
expect(activeWorker!["prAutoDetect"]).toBe("off");
|
||||
|
||||
const claimant = readMetadataRaw(sessionsDir, "app-3");
|
||||
expect(claimant!["pr"]).toBe("https://github.com/org/my-app/pull/42");
|
||||
});
|
||||
|
||||
it("automatically consolidates ownership when another session tracks the PR", async () => {
|
||||
const mockSCM = makeSCM();
|
||||
|
||||
|
|
|
|||
|
|
@ -29,6 +29,13 @@ You are the **orchestrator agent** for the ${project.name} project.
|
|||
|
||||
Your role is to coordinate and manage worker agent sessions. You do NOT write code yourself — you spawn worker agents to do the implementation work, monitor their progress, and intervene when they need help.`);
|
||||
|
||||
sections.push(`## Non-Negotiable Rules
|
||||
|
||||
- Investigations from the orchestrator session are **read-only**. Inspect status, logs, metadata, PR state, and worker output, but do not edit repository files or implement fixes from the orchestrator session.
|
||||
- Any code change, test run tied to implementation, git branch work, or PR takeover must be delegated to a **worker session**.
|
||||
- The orchestrator session must never own a PR. Never claim a PR into the orchestrator session, and never treat the orchestrator as the worker responsible for implementation.
|
||||
- If an investigation discovers follow-up work, either spawn a worker session or direct an existing worker session with clear instructions.`);
|
||||
|
||||
// Project Info
|
||||
sections.push(`## Project Info
|
||||
|
||||
|
|
@ -57,7 +64,7 @@ ao session ls -p ${projectId}
|
|||
# Send message to a session
|
||||
ao send ${project.sessionPrefix}-1 "Your message here"
|
||||
|
||||
# Claim an existing PR for a session
|
||||
# Claim an existing PR for a worker session
|
||||
ao session claim-pr 123 ${project.sessionPrefix}-1
|
||||
|
||||
# Kill a session
|
||||
|
|
@ -76,7 +83,7 @@ ao open ${projectId}
|
|||
| \`ao spawn <project> [issue] [--claim-pr <pr>]\` | Spawn a worker session, optionally attached to an existing PR |
|
||||
| \`ao batch-spawn <project> <issues...>\` | Spawn multiple sessions in parallel |
|
||||
| \`ao session ls [-p project]\` | List all sessions (optionally filter by project) |
|
||||
| \`ao session claim-pr <pr> [session]\` | Attach an existing PR to a session |
|
||||
| \`ao session claim-pr <pr> [session]\` | Attach an existing PR to a worker session |
|
||||
| \`ao session attach <session>\` | Attach to a session's tmux window |
|
||||
| \`ao session kill <session>\` | Kill a specific session |
|
||||
| \`ao session cleanup [-p project]\` | Kill completed/merged sessions |
|
||||
|
|
@ -114,14 +121,24 @@ ao send ${project.sessionPrefix}-1 "Please address the review comments on your P
|
|||
|
||||
### PR Takeover
|
||||
|
||||
If a session needs to continue work on an existing PR:
|
||||
If a worker session needs to continue work on an existing PR:
|
||||
\`\`\`bash
|
||||
ao session claim-pr 123 ${project.sessionPrefix}-1
|
||||
# or do it at spawn time
|
||||
ao spawn ${projectId} --claim-pr 123
|
||||
\`\`\`
|
||||
|
||||
This updates AO metadata, switches the worktree onto the PR branch, and lets lifecycle reactions keep routing CI and review feedback to that session.
|
||||
This updates AO metadata, switches the worker worktree onto the PR branch, and lets lifecycle reactions keep routing CI and review feedback to that worker session.
|
||||
|
||||
Never claim a PR into \`${project.sessionPrefix}-orchestrator\`. If a PR needs implementation or takeover, delegate it to a worker session instead.
|
||||
|
||||
### Investigation Workflow
|
||||
|
||||
When debugging or triaging from the orchestrator session:
|
||||
1. Inspect with read-only commands such as \`ao status\`, \`ao session ls\`, \`ao session attach\`, and SCM/tracker lookups.
|
||||
2. Decide whether a worker already owns the work or a new worker is needed.
|
||||
3. Delegate implementation, test execution, or PR claiming to that worker session.
|
||||
4. Return to monitoring and coordination once the worker has the task.
|
||||
|
||||
### Cleanup
|
||||
|
||||
|
|
@ -195,7 +212,7 @@ When an agent needs human judgment:
|
|||
2. Check the dashboard or \`ao status\` for details
|
||||
3. Attach to the session if needed: \`ao session attach <session>\`
|
||||
4. Send instructions: \`ao send <session> '...'\`
|
||||
5. Or handle it yourself (merge PR, close issue, etc.)`);
|
||||
5. Or handle the human-only action yourself (merge PR, close issue, etc.) while keeping implementation in worker sessions.`);
|
||||
|
||||
// Tips
|
||||
sections.push(`## Tips
|
||||
|
|
|
|||
|
|
@ -11,7 +11,7 @@
|
|||
* Reference: scripts/claude-ao-session, scripts/send-to-session
|
||||
*/
|
||||
|
||||
import { statSync, existsSync, readdirSync, writeFileSync, mkdirSync } from "node:fs";
|
||||
import { statSync, existsSync, readdirSync, writeFileSync, mkdirSync, utimesSync } from "node:fs";
|
||||
import { execFile } from "node:child_process";
|
||||
import { basename, join, resolve } from "node:path";
|
||||
import { homedir } from "node:os";
|
||||
|
|
@ -192,6 +192,14 @@ function getNextSessionNumber(existingSessions: string[], prefix: string): numbe
|
|||
return max + 1;
|
||||
}
|
||||
|
||||
function getSessionNumber(sessionId: string, prefix: string): number | undefined {
|
||||
const match = sessionId.match(new RegExp(`^${escapeRegex(prefix)}-(\\d+)$`));
|
||||
if (!match) return undefined;
|
||||
|
||||
const parsed = Number.parseInt(match[1], 10);
|
||||
return Number.isNaN(parsed) ? undefined : parsed;
|
||||
}
|
||||
|
||||
const PR_TRACKING_STATUSES: ReadonlySet<string> = new Set([
|
||||
"pr_open",
|
||||
"ci_failed",
|
||||
|
|
@ -201,6 +209,11 @@ const PR_TRACKING_STATUSES: ReadonlySet<string> = new Set([
|
|||
"mergeable",
|
||||
]);
|
||||
|
||||
const STALE_PR_OWNERSHIP_STATUSES: ReadonlySet<string> = new Set([
|
||||
...PR_TRACKING_STATUSES,
|
||||
"merged",
|
||||
]);
|
||||
|
||||
const SEND_RESTORE_READY_TIMEOUT_MS = 5_000;
|
||||
const SEND_RESTORE_READY_POLL_MS = 500;
|
||||
const SEND_CONFIRMATION_ATTEMPTS = 3;
|
||||
|
|
@ -240,6 +253,12 @@ export function createSessionManager(deps: SessionManagerDeps): OpenCodeSessionM
|
|||
projectId: string;
|
||||
}
|
||||
|
||||
interface ActiveSessionRecord {
|
||||
sessionName: string;
|
||||
raw: Record<string, string>;
|
||||
modifiedAt?: Date;
|
||||
}
|
||||
|
||||
/**
|
||||
* Get the sessions directory for a project.
|
||||
*/
|
||||
|
|
@ -305,43 +324,6 @@ export function createSessionManager(deps: SessionManagerDeps): OpenCodeSessionM
|
|||
return roots.some((root) => isPathInside(workspacePath, root));
|
||||
}
|
||||
|
||||
/**
|
||||
* List all session files across all projects (or filtered by projectId).
|
||||
* Scans project-specific directories under ~/.agent-orchestrator/{hash}-{projectId}/sessions/
|
||||
*
|
||||
* Note: projectId is the config key (e.g., "test-project"), not the path basename.
|
||||
*/
|
||||
function listAllSessions(projectIdFilter?: string): { sessionName: string; projectId: string }[] {
|
||||
const results: { sessionName: string; projectId: string }[] = [];
|
||||
|
||||
// Scan each project's sessions directory
|
||||
for (const [projectKey, project] of Object.entries(config.projects)) {
|
||||
// Use config key as projectId for consistency with metadata
|
||||
const projectId = projectKey;
|
||||
|
||||
// Filter by project if specified
|
||||
if (projectIdFilter && projectId !== projectIdFilter) continue;
|
||||
|
||||
const sessionsDir = getSessionsDir(config.configPath, project.path);
|
||||
if (!existsSync(sessionsDir)) continue;
|
||||
|
||||
const files = readdirSync(sessionsDir);
|
||||
for (const file of files) {
|
||||
if (file === "archive" || file.startsWith(".")) continue;
|
||||
const fullPath = join(sessionsDir, file);
|
||||
try {
|
||||
if (statSync(fullPath).isFile()) {
|
||||
results.push({ sessionName: file, projectId });
|
||||
}
|
||||
} catch {
|
||||
// Skip files that can't be stat'd
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return results;
|
||||
}
|
||||
|
||||
function listArchivedSessionIds(sessionsDir: string): string[] {
|
||||
const archiveDir = join(sessionsDir, "archive");
|
||||
if (!existsSync(archiveDir)) return [];
|
||||
|
|
@ -353,6 +335,173 @@ export function createSessionManager(deps: SessionManagerDeps): OpenCodeSessionM
|
|||
return [...ids];
|
||||
}
|
||||
|
||||
function isOrchestratorSessionRecord(
|
||||
sessionId: string,
|
||||
raw: Record<string, string> | null | undefined,
|
||||
): boolean {
|
||||
if (!raw) return false;
|
||||
return raw["role"] === "orchestrator" || sessionId.endsWith("-orchestrator");
|
||||
}
|
||||
|
||||
function applyMetadataUpdatesToRaw(
|
||||
raw: Record<string, string>,
|
||||
updates: Partial<Record<string, string>>,
|
||||
): Record<string, string> {
|
||||
let next = { ...raw };
|
||||
for (const [key, value] of Object.entries(updates)) {
|
||||
if (value === undefined) continue;
|
||||
if (value === "") {
|
||||
const { [key]: _removed, ...rest } = next;
|
||||
void _removed;
|
||||
next = rest;
|
||||
continue;
|
||||
}
|
||||
next[key] = value;
|
||||
}
|
||||
return next;
|
||||
}
|
||||
|
||||
function updateMetadataPreservingMtime(
|
||||
sessionsDir: string,
|
||||
sessionName: string,
|
||||
updates: Partial<Record<string, string>>,
|
||||
modifiedAt?: Date,
|
||||
): void {
|
||||
const metaPath = join(sessionsDir, sessionName);
|
||||
let preservedMtime = modifiedAt;
|
||||
if (!preservedMtime) {
|
||||
try {
|
||||
preservedMtime = statSync(metaPath).mtime;
|
||||
} catch {
|
||||
preservedMtime = undefined;
|
||||
}
|
||||
}
|
||||
|
||||
updateMetadata(sessionsDir, sessionName, updates);
|
||||
|
||||
if (!preservedMtime) return;
|
||||
try {
|
||||
utimesSync(metaPath, preservedMtime, preservedMtime);
|
||||
} catch {
|
||||
void 0;
|
||||
}
|
||||
}
|
||||
|
||||
function repairSingleSessionMetadataOnRead(
|
||||
sessionsDir: string,
|
||||
record: ActiveSessionRecord,
|
||||
): ActiveSessionRecord {
|
||||
const repaired = { ...record, raw: { ...record.raw } };
|
||||
if (!isOrchestratorSessionRecord(repaired.sessionName, repaired.raw)) {
|
||||
return repaired;
|
||||
}
|
||||
|
||||
const updates: Partial<Record<string, string>> = {};
|
||||
if (repaired.raw["role"] !== "orchestrator") {
|
||||
updates["role"] = "orchestrator";
|
||||
}
|
||||
if (repaired.raw["pr"]) {
|
||||
updates["pr"] = "";
|
||||
}
|
||||
if (repaired.raw["prAutoDetect"] !== "off") {
|
||||
updates["prAutoDetect"] = "off";
|
||||
}
|
||||
if (STALE_PR_OWNERSHIP_STATUSES.has(repaired.raw["status"] ?? "")) {
|
||||
updates["status"] = "working";
|
||||
}
|
||||
|
||||
if (Object.keys(updates).length > 0) {
|
||||
updateMetadataPreservingMtime(
|
||||
sessionsDir,
|
||||
repaired.sessionName,
|
||||
updates,
|
||||
repaired.modifiedAt,
|
||||
);
|
||||
repaired.raw = applyMetadataUpdatesToRaw(repaired.raw, updates);
|
||||
}
|
||||
|
||||
return repaired;
|
||||
}
|
||||
|
||||
function sessionMetadataTimestamp(record: ActiveSessionRecord): number {
|
||||
const metadataTimestamp = Date.parse(record.raw["restoredAt"] ?? record.raw["createdAt"] ?? "");
|
||||
if (record.modifiedAt) return record.modifiedAt.getTime();
|
||||
return Number.isNaN(metadataTimestamp) ? 0 : metadataTimestamp;
|
||||
}
|
||||
|
||||
function repairSessionMetadataOnRead(
|
||||
sessionsDir: string,
|
||||
records: ActiveSessionRecord[],
|
||||
): ActiveSessionRecord[] {
|
||||
const repaired = records.map((record) => ({ ...record, raw: { ...record.raw } }));
|
||||
const duplicatePRAttachments = new Map<string, ActiveSessionRecord[]>();
|
||||
|
||||
for (const record of repaired) {
|
||||
if (isOrchestratorSessionRecord(record.sessionName, record.raw)) {
|
||||
record.raw = repairSingleSessionMetadataOnRead(sessionsDir, record).raw;
|
||||
continue;
|
||||
}
|
||||
|
||||
const prUrl = record.raw["pr"];
|
||||
if (!prUrl) continue;
|
||||
|
||||
const attached = duplicatePRAttachments.get(prUrl) ?? [];
|
||||
attached.push(record);
|
||||
duplicatePRAttachments.set(prUrl, attached);
|
||||
}
|
||||
|
||||
for (const attachedRecords of duplicatePRAttachments.values()) {
|
||||
if (attachedRecords.length < 2) continue;
|
||||
|
||||
const [owner, ...staleRecords] = [...attachedRecords].sort((a, b) => {
|
||||
const trackingDiff =
|
||||
Number(PR_TRACKING_STATUSES.has(b.raw["status"] ?? "")) -
|
||||
Number(PR_TRACKING_STATUSES.has(a.raw["status"] ?? ""));
|
||||
if (trackingDiff !== 0) return trackingDiff;
|
||||
|
||||
const timestampDiff = sessionMetadataTimestamp(b) - sessionMetadataTimestamp(a);
|
||||
if (timestampDiff !== 0) return timestampDiff;
|
||||
|
||||
return b.sessionName.localeCompare(a.sessionName);
|
||||
});
|
||||
|
||||
void owner;
|
||||
|
||||
for (const record of staleRecords) {
|
||||
const updates: Partial<Record<string, string>> = {
|
||||
pr: "",
|
||||
prAutoDetect: "off",
|
||||
...(PR_TRACKING_STATUSES.has(record.raw["status"] ?? "") ? { status: "working" } : {}),
|
||||
};
|
||||
updateMetadataPreservingMtime(sessionsDir, record.sessionName, updates, record.modifiedAt);
|
||||
record.raw = applyMetadataUpdatesToRaw(record.raw, updates);
|
||||
}
|
||||
}
|
||||
|
||||
return repaired;
|
||||
}
|
||||
|
||||
function loadActiveSessionRecords(project: ProjectConfig): ActiveSessionRecord[] {
|
||||
const sessionsDir = getProjectSessionsDir(project);
|
||||
if (!existsSync(sessionsDir)) return [];
|
||||
|
||||
const records = listMetadata(sessionsDir).flatMap((sessionName) => {
|
||||
const raw = readMetadataRaw(sessionsDir, sessionName);
|
||||
if (!raw) return [];
|
||||
|
||||
let modifiedAt: Date | undefined;
|
||||
try {
|
||||
modifiedAt = statSync(join(sessionsDir, sessionName)).mtime;
|
||||
} catch {
|
||||
void 0;
|
||||
}
|
||||
|
||||
return [{ sessionName, raw, modifiedAt } satisfies ActiveSessionRecord];
|
||||
});
|
||||
|
||||
return repairSessionMetadataOnRead(sessionsDir, records);
|
||||
}
|
||||
|
||||
function markArchivedOpenCodeCleanup(sessionsDir: string, sessionId: SessionId): void {
|
||||
updateArchivedMetadata(sessionsDir, sessionId, {
|
||||
opencodeSessionId: "",
|
||||
|
|
@ -442,6 +591,81 @@ export function createSessionManager(deps: SessionManagerDeps): OpenCodeSessionM
|
|||
return candidateIds[0];
|
||||
}
|
||||
|
||||
async function listRemoteSessionNumbers(project: ProjectConfig): Promise<number[]> {
|
||||
try {
|
||||
const { stdout } = await execFileAsync(
|
||||
"git",
|
||||
["ls-remote", "--heads", "origin", `session/${project.sessionPrefix}-*`],
|
||||
{
|
||||
cwd: project.path,
|
||||
timeout: 5_000,
|
||||
},
|
||||
);
|
||||
|
||||
return stdout
|
||||
.split("\n")
|
||||
.flatMap((line: string) => {
|
||||
const trimmed = line.trim();
|
||||
if (!trimmed) return [];
|
||||
|
||||
const ref = trimmed.split(/\s+/)[1] ?? "";
|
||||
const match = ref.match(
|
||||
new RegExp(`refs/heads/session/${escapeRegex(project.sessionPrefix)}-(\\d+)$`),
|
||||
);
|
||||
if (!match) return [];
|
||||
|
||||
const parsed = Number.parseInt(match[1], 10);
|
||||
return Number.isNaN(parsed) ? [] : [parsed];
|
||||
})
|
||||
.filter((num: number, index: number, values: number[]) => values.indexOf(num) === index);
|
||||
} catch {
|
||||
return [];
|
||||
}
|
||||
}
|
||||
|
||||
async function reserveNextSessionIdentity(
|
||||
project: ProjectConfig,
|
||||
sessionsDir: string,
|
||||
): Promise<{
|
||||
num: number;
|
||||
sessionId: string;
|
||||
tmuxName: string | undefined;
|
||||
}> {
|
||||
const usedNumbers = new Set<number>();
|
||||
for (const sessionName of [
|
||||
...listMetadata(sessionsDir),
|
||||
...listArchivedSessionIds(sessionsDir),
|
||||
]) {
|
||||
const num = getSessionNumber(sessionName, project.sessionPrefix);
|
||||
if (num !== undefined) usedNumbers.add(num);
|
||||
}
|
||||
for (const num of await listRemoteSessionNumbers(project)) {
|
||||
usedNumbers.add(num);
|
||||
}
|
||||
|
||||
let num = getNextSessionNumber(
|
||||
[...usedNumbers].map((value) => `${project.sessionPrefix}-${value}`),
|
||||
project.sessionPrefix,
|
||||
);
|
||||
for (let attempts = 0; attempts < 10_000; attempts++) {
|
||||
const sessionId = `${project.sessionPrefix}-${num}`;
|
||||
const tmuxName = config.configPath
|
||||
? generateTmuxName(config.configPath, project.sessionPrefix, num)
|
||||
: undefined;
|
||||
|
||||
if (!usedNumbers.has(num) && reserveSessionId(sessionsDir, sessionId)) {
|
||||
return { num, sessionId, tmuxName };
|
||||
}
|
||||
|
||||
usedNumbers.add(num);
|
||||
num += 1;
|
||||
}
|
||||
|
||||
throw new Error(
|
||||
`Failed to reserve session ID after 10000 attempts (prefix: ${project.sessionPrefix})`,
|
||||
);
|
||||
}
|
||||
|
||||
/** Resolve which plugins to use for a project. */
|
||||
function resolvePlugins(project: ProjectConfig, agentOverride?: string) {
|
||||
const runtime = registry.get<Runtime>("runtime", project.runtime ?? config.defaults.runtime);
|
||||
|
|
@ -487,7 +711,21 @@ export function createSessionManager(deps: SessionManagerDeps): OpenCodeSessionM
|
|||
const sessionsDir = getProjectSessionsDir(project);
|
||||
const raw = readMetadataRaw(sessionsDir, sessionId);
|
||||
if (!raw) continue;
|
||||
return { raw, sessionsDir, project, projectId };
|
||||
|
||||
let modifiedAt: Date | undefined;
|
||||
try {
|
||||
modifiedAt = statSync(join(sessionsDir, sessionId)).mtime;
|
||||
} catch {
|
||||
modifiedAt = undefined;
|
||||
}
|
||||
|
||||
const repaired = repairSingleSessionMetadataOnRead(sessionsDir, {
|
||||
sessionName: sessionId,
|
||||
raw,
|
||||
modifiedAt,
|
||||
});
|
||||
|
||||
return { raw: repaired.raw, sessionsDir, project, projectId };
|
||||
}
|
||||
|
||||
return null;
|
||||
|
|
@ -655,29 +893,7 @@ export function createSessionManager(deps: SessionManagerDeps): OpenCodeSessionM
|
|||
}
|
||||
|
||||
// Determine session ID — atomically reserve to prevent concurrent collisions
|
||||
const existingSessions = listMetadata(sessionsDir);
|
||||
let num = getNextSessionNumber(existingSessions, project.sessionPrefix);
|
||||
let sessionId: string;
|
||||
let tmuxName: string | undefined;
|
||||
for (let attempts = 0; attempts < 10; attempts++) {
|
||||
sessionId = `${project.sessionPrefix}-${num}`;
|
||||
// Generate tmux name if using new architecture
|
||||
if (config.configPath) {
|
||||
tmuxName = generateTmuxName(config.configPath, project.sessionPrefix, num);
|
||||
}
|
||||
if (reserveSessionId(sessionsDir, sessionId)) break;
|
||||
num++;
|
||||
if (attempts === 9) {
|
||||
throw new Error(
|
||||
`Failed to reserve session ID after 10 attempts (prefix: ${project.sessionPrefix})`,
|
||||
);
|
||||
}
|
||||
}
|
||||
// Reassign to satisfy TypeScript's flow analysis (not redundant from compiler's perspective)
|
||||
sessionId = `${project.sessionPrefix}-${num}`;
|
||||
if (config.configPath) {
|
||||
tmuxName = generateTmuxName(config.configPath, project.sessionPrefix, num);
|
||||
}
|
||||
const { sessionId, tmuxName } = await reserveNextSessionIdentity(project, sessionsDir);
|
||||
|
||||
// Determine branch name — explicit branch always takes priority
|
||||
let branch: string;
|
||||
|
|
@ -1173,16 +1389,21 @@ export function createSessionManager(deps: SessionManagerDeps): OpenCodeSessionM
|
|||
}
|
||||
|
||||
async function list(projectId?: string): Promise<Session[]> {
|
||||
const allSessions = listAllSessions(projectId);
|
||||
const allSessions = Object.entries(config.projects).flatMap(([entryProjectId, project]) => {
|
||||
if (projectId && entryProjectId !== projectId) return [];
|
||||
return loadActiveSessionRecords(project).map((record) => ({
|
||||
sessionName: record.sessionName,
|
||||
projectId: entryProjectId,
|
||||
raw: record.raw,
|
||||
}));
|
||||
});
|
||||
let openCodeSessionListPromise: Promise<OpenCodeSessionListEntry[]> | undefined;
|
||||
|
||||
const tasks = allSessions.map(async ({ sessionName, projectId: sessionProjectId }) => {
|
||||
const tasks = allSessions.map(async ({ sessionName, projectId: sessionProjectId, raw }) => {
|
||||
const project = config.projects[sessionProjectId];
|
||||
if (!project) return null;
|
||||
|
||||
const sessionsDir = getProjectSessionsDir(project);
|
||||
const raw = readMetadataRaw(sessionsDir, sessionName);
|
||||
if (!raw) return null;
|
||||
|
||||
let createdAt: Date | undefined;
|
||||
let modifiedAt: Date | undefined;
|
||||
|
|
@ -1251,9 +1472,15 @@ export function createSessionManager(deps: SessionManagerDeps): OpenCodeSessionM
|
|||
// If stat fails, timestamps will fall back to current time
|
||||
}
|
||||
|
||||
const session = metadataToSession(sessionId, raw, createdAt, modifiedAt);
|
||||
const repaired = repairSingleSessionMetadataOnRead(sessionsDir, {
|
||||
sessionName: sessionId,
|
||||
raw,
|
||||
modifiedAt,
|
||||
});
|
||||
|
||||
const selectedAgentName = raw["agent"];
|
||||
const session = metadataToSession(sessionId, repaired.raw, createdAt, modifiedAt);
|
||||
|
||||
const selectedAgentName = repaired.raw["agent"];
|
||||
const effectiveAgentName = selectedAgentName ?? project.agent ?? config.defaults.agent;
|
||||
const plugins = resolvePlugins(project, effectiveAgentName);
|
||||
await ensureHandleAndEnrich(
|
||||
|
|
@ -1713,7 +1940,7 @@ export function createSessionManager(deps: SessionManagerDeps): OpenCodeSessionM
|
|||
if (!reference) throw new Error("PR reference is required");
|
||||
|
||||
const { raw, sessionsDir, project, projectId } = requireSessionRecord(sessionId);
|
||||
if (raw["role"] === "orchestrator") {
|
||||
if (isOrchestratorSessionRecord(sessionId, raw)) {
|
||||
throw new Error(`Session ${sessionId} is an orchestrator session and cannot claim PRs`);
|
||||
}
|
||||
|
||||
|
|
@ -1732,11 +1959,12 @@ export function createSessionManager(deps: SessionManagerDeps): OpenCodeSessionM
|
|||
}
|
||||
|
||||
const conflictingSessions = new Set<SessionId>();
|
||||
for (const { sessionName } of listAllSessions(projectId)) {
|
||||
if (sessionName === sessionId) continue;
|
||||
const activeRecords = loadActiveSessionRecords(project).filter(
|
||||
(record) => record.sessionName !== sessionId,
|
||||
);
|
||||
|
||||
const otherRaw = readMetadataRaw(sessionsDir, sessionName);
|
||||
if (!otherRaw || otherRaw["role"] === "orchestrator") continue;
|
||||
for (const { sessionName, raw: otherRaw } of activeRecords) {
|
||||
if (!otherRaw || isOrchestratorSessionRecord(sessionName, otherRaw)) continue;
|
||||
|
||||
const samePr = otherRaw["pr"] === pr.url;
|
||||
const sameBranch =
|
||||
|
|
|
|||
Loading…
Reference in New Issue