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

fix: decouple PR truth from worker session policy

This commit is contained in:
harshitsinghbhandari 2026-04-17 10:44:35 +05:30
parent f2b74a8fc8
commit a45eb322bb
15 changed files with 191 additions and 55 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
.changeset/stage4-pr-policy.md Normal file
View File

@ -0,0 +1,6 @@
---
"@aoagents/ao-core": patch
"@aoagents/ao-web": patch
---
Decouple canonical session state from PR state so workers stay idle while waiting on reviews or merged/closed PR decisions, stop cleanup from auto-killing merged PR sessions, and make the dashboard/rendered labels follow canonical PR truth instead of inferring it from legacy lifecycle aliases.

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

@ -375,7 +375,7 @@ describe("check (single session)", () => {
runtime: plugins.runtime,
agent: plugins.agent,
});
vi.mocked(registryWithoutAgent.get).mockImplementation((slot: string, name?: string) => {
vi.mocked(registryWithoutAgent.get).mockImplementation((slot: string, _name?: string) => {
if (slot === "runtime") return plugins.runtime;
if (slot === "agent") return null;
return null;
@ -584,6 +584,38 @@ describe("check (single session)", () => {
expect(lm.getStates().get("app-1")).toBe("ci_failed");
});
it("keeps canonical session state idle while waiting on external review", async () => {
const mockSCM = createMockSCM({ getReviewDecision: vi.fn().mockResolvedValue("pending") });
const registry = createMockRegistry({
runtime: plugins.runtime,
agent: plugins.agent,
scm: mockSCM,
});
const session = makeSession({ status: "pr_open", pr: makePR() });
vi.mocked(mockSessionManager.get).mockResolvedValue(session);
writeMetadata(env.sessionsDir, "app-1", {
worktree: "/tmp",
branch: session.branch ?? "main",
status: session.status,
project: "my-app",
pr: session.pr?.url,
runtimeHandle: session.runtimeHandle ? JSON.stringify(session.runtimeHandle) : undefined,
});
const lm = createLifecycleManager({
config,
registry,
sessionManager: mockSessionManager,
});
await lm.check("app-1");
expect(lm.getStates().get("app-1")).toBe("review_pending");
expect(session.lifecycle.session.state).toBe("idle");
expect(session.lifecycle.session.reason).toBe("awaiting_external_review");
});
it("skips PR auto-detection when metadata disables it", async () => {
const mockSCM = createMockSCM({ detectPR: vi.fn().mockResolvedValue(makePR()) });
const registry = createMockRegistry({
@ -702,23 +734,36 @@ describe("check (single session)", () => {
expect(lm.getStates().get("app-1")).toBe("merged");
});
it("treats closed PRs as done when the canonical lifecycle marks them complete", async () => {
it("keeps closed PR sessions idle and emits a PR-closed notification", async () => {
const mockSCM = createMockSCM({ getPRState: vi.fn().mockResolvedValue("closed") });
const notifier = createMockNotifier();
const registry = createMockRegistry({
runtime: plugins.runtime,
agent: plugins.agent,
scm: mockSCM,
notifier,
});
const session = makeSession({ status: "pr_open", pr: makePR() });
const lm = setupCheck("app-1", {
session: makeSession({ status: "pr_open", pr: makePR() }),
session,
registry,
configOverride: {
...config,
notificationRouting: {
...config.notificationRouting,
info: ["desktop"],
},
},
});
await lm.check("app-1");
expect(lm.getStates().get("app-1")).toBe("done");
expect(lm.getStates().get("app-1")).toBe("idle");
const meta = readMetadataRaw(env.sessionsDir, "app-1");
expect(meta?.["status"]).toBe("done");
expect(meta?.["status"]).toBe("idle");
expect(meta?.["statePayload"]).toContain('"state":"closed"');
expect(meta?.["statePayload"]).toContain('"reason":"pr_closed_waiting_decision"');
expect(notifier.notify).toHaveBeenCalledWith(expect.objectContaining({ type: "pr.closed" }));
});
it("detects mergeable when approved + CI green", async () => {

4
packages/core/src/__tests__/plugin-integration.test.ts
View File

@ -392,7 +392,7 @@ describe("plugin integration", () => {
// -------------------------------------------------------------------------
describe("SessionManager + SCM", () => {
it("cleanup() calls scm-github getPRState() and kills merged PR sessions", async () => {
it("cleanup() calls scm-github getPRState() but keeps merged PR sessions alive", async () => {
const registry = createTestRegistry();
const sm = createSessionManager({ config, registry });
@ -412,7 +412,7 @@ describe("plugin integration", () => {
const result = await sm.cleanup("my-app");
expect(result.killed).toContain("app-1");
expect(result.skipped).toContain("app-1");
// Verify gh CLI was called for PR state check
expect(ghMock).toHaveBeenCalledWith(
"gh",

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

@ -229,11 +229,11 @@ describe("kill", () => {
});
describe("cleanup", () => {
it("kills sessions with merged PRs", async () => {
it("kills sessions with closed PRs", async () => {
const mockSCM: SCM = {
name: "mock-scm",
detectPR: vi.fn(),
getPRState: vi.fn().mockResolvedValue("merged"),
getPRState: vi.fn().mockResolvedValue("closed"),
mergePR: vi.fn(),
closePR: vi.fn(),
getCIChecks: vi.fn(),
@ -272,7 +272,7 @@ describe("cleanup", () => {
expect(result.skipped).toHaveLength(0);
});
it("deletes mapped OpenCode session during cleanup", async () => {
it("deletes mapped OpenCode session during cleanup for closed PRs", async () => {
const deleteLogPath = join(tmpDir, "opencode-delete.log");
const mockBin = installMockOpencode(tmpDir, "[]", deleteLogPath);
process.env.PATH = `${mockBin}:${originalPath ?? ""}`;
@ -280,7 +280,7 @@ describe("cleanup", () => {
const mockSCM: SCM = {
name: "mock-scm",
detectPR: vi.fn(),
getPRState: vi.fn().mockResolvedValue("merged"),
getPRState: vi.fn().mockResolvedValue("closed"),
mergePR: vi.fn(),
closePR: vi.fn(),
getCIChecks: vi.fn(),
@ -322,14 +322,14 @@ describe("cleanup", () => {
expect(deleteLog).toContain("session delete ses_cleanup");
});
it("treats missing mapped OpenCode session as already cleaned", async () => {
it("treats missing mapped OpenCode session as already cleaned for closed PRs", async () => {
const mockBin = installMockOpencodeWithNotFoundDelete(tmpDir, "[]");
process.env.PATH = `${mockBin}:${originalPath ?? ""}`;
const mockSCM: SCM = {
name: "mock-scm",
detectPR: vi.fn(),
getPRState: vi.fn().mockResolvedValue("merged"),
getPRState: vi.fn().mockResolvedValue("closed"),
mergePR: vi.fn(),
closePR: vi.fn(),
getCIChecks: vi.fn(),
@ -392,7 +392,7 @@ describe("cleanup", () => {
expect(result.killed).toContain("app-6");
const deleteLog = readFileSync(deleteLogPath, "utf-8");
expect(deleteLog).toContain("session delete ses_archived");
});
}, 15_000);
it("does not skip archived cleanup for matching session IDs in other projects", async () => {
const deleteLogPath = join(tmpDir, "opencode-delete-archived-cross-project.log");

4
packages/core/src/__tests__/session-manager/spawn.test.ts
View File

@ -576,7 +576,7 @@ describe("spawn", () => {
const metadata = readMetadataRaw(sessionsDir, session.id);
expect(metadata?.["opencodeSessionId"]).toBeUndefined();
});
}, 15_000);
it("throws for unknown project", async () => {
const sm = createSessionManager({ config, registry: mockRegistry });
@ -1865,7 +1865,7 @@ describe("spawn", () => {
}),
}),
);
});
}, 15_000);
it("throws for unknown project", async () => {
const sm = createSessionManager({ config, registry: mockRegistry });

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

@ -142,6 +142,19 @@ function statusToEventType(_from: SessionStatus | undefined, to: SessionStatus):
}
}
function prStateToEventType(
from: Session["lifecycle"]["pr"]["state"],
to: Session["lifecycle"]["pr"]["state"],
): EventType | null {
if (from === to) return null;
switch (to) {
case "closed":
return "pr.closed";
default:
return null;
}
}
/** Map event type to reaction config key. */
function eventToReactionKey(eventType: EventType): string | null {
switch (eventType) {
@ -683,8 +696,8 @@ export function createLifecycleManager(deps: LifecycleManagerDeps): LifecycleMan
if (cachedData.state === PR_STATE.CLOSED) {
lifecycle.pr.state = "closed";
lifecycle.pr.reason = "closed_unmerged";
setSessionState("done", "research_complete");
return commit(SESSION_STATUS.DONE, "pr_closed", 0);
setSessionState("idle", "pr_closed_waiting_decision");
return commit(SESSION_STATUS.IDLE, "pr_closed", 0);
}
lifecycle.pr.state = "open";
@ -701,18 +714,18 @@ export function createLifecycleManager(deps: LifecycleManagerDeps): LifecycleMan
if (cachedData.reviewDecision === "approved" || cachedData.reviewDecision === "none") {
if (cachedData.mergeable) {
lifecycle.pr.reason = "merge_ready";
setSessionState("working", "awaiting_external_review");
setSessionState("idle", "awaiting_external_review");
return commit(SESSION_STATUS.MERGEABLE, "merge_ready", 0);
}
if (cachedData.reviewDecision === "approved") {
lifecycle.pr.reason = "approved";
setSessionState("working", "awaiting_external_review");
setSessionState("idle", "awaiting_external_review");
return commit(SESSION_STATUS.APPROVED, "review_approved", 0);
}
}
if (cachedData.reviewDecision === "pending") {
lifecycle.pr.reason = "review_pending";
setSessionState("working", "awaiting_external_review");
setSessionState("idle", "awaiting_external_review");
return commit(SESSION_STATUS.REVIEW_PENDING, "review_pending", 0);
}
@ -723,7 +736,7 @@ export function createLifecycleManager(deps: LifecycleManagerDeps): LifecycleMan
}
lifecycle.pr.reason = "in_progress";
setSessionState("working", "pr_created");
setSessionState("idle", "pr_created");
return commit(SESSION_STATUS.PR_OPEN, "pr_open", 0);
}
@ -737,8 +750,8 @@ export function createLifecycleManager(deps: LifecycleManagerDeps): LifecycleMan
if (prState === PR_STATE.CLOSED) {
lifecycle.pr.state = "closed";
lifecycle.pr.reason = "closed_unmerged";
setSessionState("done", "research_complete");
return commit(SESSION_STATUS.DONE, "pr_closed", 0);
setSessionState("idle", "pr_closed_waiting_decision");
return commit(SESSION_STATUS.IDLE, "pr_closed", 0);
}
lifecycle.pr.state = "open";
@ -759,18 +772,18 @@ export function createLifecycleManager(deps: LifecycleManagerDeps): LifecycleMan
const mergeReady = await scm.getMergeability(session.pr);
if (mergeReady.mergeable) {
lifecycle.pr.reason = "merge_ready";
setSessionState("working", "awaiting_external_review");
setSessionState("idle", "awaiting_external_review");
return commit(SESSION_STATUS.MERGEABLE, "merge_ready", 0);
}
if (reviewDecision === "approved") {
lifecycle.pr.reason = "approved";
setSessionState("working", "awaiting_external_review");
setSessionState("idle", "awaiting_external_review");
return commit(SESSION_STATUS.APPROVED, "review_approved", 0);
}
}
if (reviewDecision === "pending") {
lifecycle.pr.reason = "review_pending";
setSessionState("working", "awaiting_external_review");
setSessionState("idle", "awaiting_external_review");
return commit(SESSION_STATUS.REVIEW_PENDING, "review_pending", 0);
}
@ -781,7 +794,7 @@ export function createLifecycleManager(deps: LifecycleManagerDeps): LifecycleMan
}
lifecycle.pr.reason = "in_progress";
setSessionState("working", "pr_created");
setSessionState("idle", "pr_created");
return commit(SESSION_STATUS.PR_OPEN, "pr_open", 0);
} catch {
// Keep current status on SCM failure.
@ -1463,6 +1476,7 @@ export function createLifecycleManager(deps: LifecycleManagerDeps): LifecycleMan
const tracked = states.get(session.id);
const oldStatus =
tracked ?? ((session.metadata?.["status"] as SessionStatus | undefined) || session.status);
const previousPRState = session.lifecycle.pr.state;
const assessment = await determineStatus(session);
const newStatus = assessment.status;
const lifecycleChanged = session.metadata["statePayload"] !== JSON.stringify(session.lifecycle);
@ -1572,6 +1586,22 @@ export function createLifecycleManager(deps: LifecycleManagerDeps): LifecycleMan
}
}
const prEventType = prStateToEventType(previousPRState, session.lifecycle.pr.state);
if (prEventType) {
const prEvent = createEvent(prEventType, {
sessionId: session.id,
projectId: session.projectId,
message: `${session.id}: PR ${previousPRState} → ${session.lifecycle.pr.state}`,
data: {
oldPRState: previousPRState,
newPRState: session.lifecycle.pr.state,
prNumber: session.lifecycle.pr.number,
prUrl: session.lifecycle.pr.url,
},
});
await notifyHuman(prEvent, inferPriority(prEventType));
}
// Pin first quality summary for title stability
if (
session.agentInfo?.summary &&

22
packages/core/src/lifecycle-state.ts
View File

@ -250,8 +250,20 @@ export function parseCanonicalLifecycle(
export function deriveLegacyStatus(
lifecycle: CanonicalSessionLifecycle,
previousStatus: SessionStatus = "working",
_previousStatus: SessionStatus = "working",
): SessionStatus {
if (lifecycle.pr.state === "merged") {
return "merged";
}
if (lifecycle.pr.state === "open") {
if (lifecycle.pr.reason === "ci_failing") return "ci_failed";
if (lifecycle.pr.reason === "changes_requested") return "changes_requested";
if (lifecycle.pr.reason === "review_pending") return "review_pending";
if (lifecycle.pr.reason === "approved") return "approved";
if (lifecycle.pr.reason === "merge_ready") return "mergeable";
return "pr_open";
}
switch (lifecycle.session.state) {
case "not_started":
return "spawning";
@ -264,16 +276,10 @@ export function deriveLegacyStatus(
case "terminated":
return "terminated";
case "idle":
return lifecycle.pr.state === "merged" ? "merged" : "idle";
return "idle";
case "detecting":
return "detecting";
case "working":
if (lifecycle.pr.reason === "ci_failing") return "ci_failed";
if (lifecycle.pr.reason === "changes_requested") return "changes_requested";
if (lifecycle.pr.reason === "review_pending") return "review_pending";
if (lifecycle.pr.reason === "approved") return "approved";
if (lifecycle.pr.reason === "merge_ready") return "mergeable";
if (lifecycle.pr.state === "open") return "pr_open";
return "working";
}
}

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

@ -1876,7 +1876,7 @@ export function createSessionManager(deps: SessionManagerDeps): OpenCodeSessionM
if (session.pr && plugins.scm) {
try {
const prState = await plugins.scm.getPRState(session.pr);
if (prState === PR_STATE.MERGED || prState === PR_STATE.CLOSED) {
if (prState === PR_STATE.CLOSED) {
shouldKill = true;
}
} catch {

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

@ -41,6 +41,7 @@ export type CanonicalSessionReason =
| "agent_acknowledged"
| "task_in_progress"
| "pr_created"
| "pr_closed_waiting_decision"
| "fixing_ci"
| "resolving_review_comments"
| "awaiting_user_input"

19
packages/web/src/components/SessionCard.tsx
View File

@ -53,6 +53,25 @@ function getDoneStatusInfo(session: DashboardSession): {
};
}
if (prState === "closed") {
return {
label: "closed",
pillClass: "done-status-pill--exited",
icon: (
<svg
fill="none"
stroke="currentColor"
strokeWidth="2"
viewBox="0 0 24 24"
className="h-3 w-3"
>
<circle cx="12" cy="12" r="9" />
<path d="M9 12h6" />
</svg>
),
};
}
if (status === "killed" || status === "terminated") {
return {
label: status,

10
packages/web/src/lib/__tests__/dashboard-page-data.fast-path.test.ts
View File

@ -60,15 +60,15 @@ describe("getDashboardPageData fast path", () => {
hoisted.enrichSessionsMetadataFastMock.mockResolvedValue(undefined);
});
it("runs fast enrichment, uses cache-only PR hydration, and infers merged/closed state for terminal cache misses even without SCM", async () => {
it("runs fast enrichment, uses cache-only PR hydration, and preserves canonical PR state on cache misses even without SCM", async () => {
const noPrCore = { id: "session-no-pr", status: "working", pr: null };
const closedCore = { id: "session-closed", status: "killed", pr: { number: 2 } };
const mergedCore = { id: "session-merged", status: "merged", pr: { number: 3 } };
const closedCore = { id: "session-closed", status: "idle", pr: { number: 2 } };
const mergedCore = { id: "session-merged", status: "idle", pr: { number: 3 } };
const allSessions = [noPrCore, closedCore, mergedCore];
const dashboardNoPr = { id: "session-no-pr", pr: null };
const dashboardClosed = { id: "session-closed", pr: { state: "open", enriched: false } };
const dashboardMerged = { id: "session-merged", pr: { state: "open", enriched: false } };
const dashboardClosed = { id: "session-closed", pr: { state: "closed", enriched: false } };
const dashboardMerged = { id: "session-merged", pr: { state: "merged", enriched: false } };
hoisted.getServicesMock.mockResolvedValue({
config: { projects: { docs: { id: "docs" } } },

21
packages/web/src/lib/__tests__/serialize.test.ts
View File

@ -149,6 +149,27 @@ describe("sessionToDashboard", () => {
});
});
it("should seed dashboard PR state from canonical lifecycle truth", () => {
const lifecycle = createInitialCanonicalLifecycle("worker", new Date("2025-01-01T00:00:00Z"));
lifecycle.session.state = "idle";
lifecycle.session.reason = "merged_waiting_decision";
lifecycle.session.startedAt = lifecycle.session.lastTransitionAt;
lifecycle.pr.state = "merged";
lifecycle.pr.reason = "merged";
lifecycle.runtime.state = "alive";
lifecycle.runtime.reason = "process_running";
const coreSession = createCoreSession({
status: "idle",
lifecycle,
pr: createPRInfo(),
});
const dashboard = sessionToDashboard(coreSession);
expect(dashboard.pr?.state).toBe("merged");
});
it("should use agentInfo summary with summaryIsFallback false", () => {
const coreSession = createCoreSession({
agentInfo: {

12
packages/web/src/lib/dashboard-page-data.ts
View File

@ -1,7 +1,7 @@
import "server-only";
import { cache } from "react";
import { TERMINAL_STATUSES, type DashboardSession, type DashboardOrchestratorLink } from "@/lib/types";
import { type DashboardSession, type DashboardOrchestratorLink } from "@/lib/types";
import { getServices, getSCM } from "@/lib/services";
import {
sessionToDashboard,
@ -84,16 +84,6 @@ export const getDashboardPageData = cache(async function getDashboardPageData(pr
await enrichSessionPR(pageData.sessions[i], scm, core.pr, { cacheOnly: true });
}
// For cache-miss PRs, infer terminal PR state from lifecycle status
// to avoid showing merged/closed PRs as "open" until client refresh.
const sessionPR = pageData.sessions[i].pr;
if (sessionPR && !sessionPR.enriched && TERMINAL_STATUSES.has(core.status)) {
if (core.status === "merged") {
sessionPR.state = "merged";
} else if (core.status === "killed") {
sessionPR.state = "closed";
}
}
}
} catch {
pageData.sessions = [];

18
packages/web/src/lib/serialize.ts
View File

@ -75,7 +75,12 @@ export function sessionToDashboard(session: Session): DashboardSession {
summaryIsFallback: agentSummary ? (session.agentInfo?.summaryIsFallback ?? false) : false,
createdAt: session.createdAt.toISOString(),
lastActivityAt: session.lastActivityAt.toISOString(),
pr: session.pr ? basicPRToDashboard(session.pr) : null,
pr: session.pr
? {
...basicPRToDashboard(session.pr),
state: normalizeDashboardPRState(session.lifecycle.pr.state),
}
: null,
metadata: session.metadata,
};
}
@ -137,6 +142,17 @@ function basicPRToDashboard(pr: PRInfo): DashboardPR {
};
}
function normalizeDashboardPRState(state: Session["lifecycle"]["pr"]["state"]): DashboardPR["state"] {
switch (state) {
case "merged":
return "merged";
case "closed":
return "closed";
default:
return "open";
}
}
/**
* Enrich a DashboardSession's PR with live data from the SCM plugin.
* Uses cache to reduce API calls and handles rate limit errors gracefully.

4
packages/web/src/lib/services.ts
View File

@ -129,7 +129,9 @@ async function labelIssuesForVerification(
): Promise<void> {
const mergedSessions = sessions.filter(
(s) =>
s.status === "merged" && s.issueId && !processedIssues.has(`${s.projectId}:${s.issueId}`),
s.lifecycle.pr.state === "merged" &&
s.issueId &&
!processedIssues.has(`${s.projectId}:${s.issueId}`),
);
for (const session of mergedSessions) {