From 50ccf4e4185fcbe05ce824237c5e973c936dcd33 Mon Sep 17 00:00:00 2001 From: -Puter <22245429+puterhimself@users.noreply.github.com> Date: Sat, 25 Jul 2026 01:53:16 +0530 Subject: [PATCH] feat(agents/orb): wire project-manager orchestration over merged Orb runtime Add a thin orchestration layer that drives the existing Orb runtime to progress issue-scoped work. OrbProjectManager creates/resumes Orb runs, assembles context packs, projects OrbEvents into durable project events, detects needs-input and work-complete conditions, forwards follow-up messages, and runs the Git/Gitea lifecycle on completion. Port interfaces (OrbAdapter, OrbRunPort, GitLifecyclePort) make the orchestrator testable with fakes. The real adapters wrap OrbRuntime and runPostRunGiteaLifecycle without reimplementing Docker, AgentOS, or OpenCode. Idempotency: duplicate starts reuse active runs, duplicate completion returns cached results, cancel is idempotent. Never auto-merges. 88 tests pass (19 orchestration + 12 event-mapping + orb suite), 2 live tests skipped; lint and type checks clean. --- packages/agents/src/orb/context-pack.ts | 109 +++ packages/agents/src/orb/git-adapter.ts | 81 ++ packages/agents/src/orb/index.ts | 6 + packages/agents/src/orb/orb-adapter.ts | 132 ++++ .../src/orb/orb-project-manager.live.test.ts | 132 ++++ .../src/orb/orb-project-manager.test.ts | 709 ++++++++++++++++++ .../agents/src/orb/orb-project-manager.ts | 636 ++++++++++++++++ packages/agents/src/orb/ports.ts | 136 ++++ .../agents/src/orb/project-events.test.ts | 123 +++ packages/agents/src/orb/project-events.ts | 171 +++++ 10 files changed, 2235 insertions(+) create mode 100644 packages/agents/src/orb/context-pack.ts create mode 100644 packages/agents/src/orb/git-adapter.ts create mode 100644 packages/agents/src/orb/orb-adapter.ts create mode 100644 packages/agents/src/orb/orb-project-manager.live.test.ts create mode 100644 packages/agents/src/orb/orb-project-manager.test.ts create mode 100644 packages/agents/src/orb/orb-project-manager.ts create mode 100644 packages/agents/src/orb/ports.ts create mode 100644 packages/agents/src/orb/project-events.test.ts create mode 100644 packages/agents/src/orb/project-events.ts diff --git a/packages/agents/src/orb/context-pack.ts b/packages/agents/src/orb/context-pack.ts new file mode 100644 index 0000000..a6e61c5 --- /dev/null +++ b/packages/agents/src/orb/context-pack.ts @@ -0,0 +1,109 @@ +import { Schema } from "effect"; + +// --------------------------------------------------------------------------- +// Context pack — assembled from ProjectIssue, evidence, project docs, artifacts +// --------------------------------------------------------------------------- + +export const ContextFile = Schema.Struct({ + content: Schema.String, + label: Schema.String, +}); +export type ContextFile = typeof ContextFile.Type; + +export const ContextPackInput = Schema.Struct({ + artifacts: Schema.Array( + Schema.Struct({ + content: Schema.String, + path: Schema.String, + }) + ), + contextFiles: Schema.Array(ContextFile), + evidence: Schema.Array( + Schema.Struct({ + content: Schema.String, + source: Schema.String, + }) + ), + issueBody: Schema.String, + issueNumber: Schema.Int, + issueTitle: Schema.String, + repositoryMetadata: Schema.Struct({ + baseBranch: Schema.String, + repositoryName: Schema.String, + repositoryUrl: Schema.String, + }), +}); +export type ContextPackInput = typeof ContextPackInput.Type; + +const section = (heading: string, lines: readonly string[]): string => { + if (lines.length === 0) { + return ""; + } + return `## ${heading}\n\n${lines.join("\n")}`; +}; + +const OPERATIONAL_INSTRUCTIONS = `## Operational Instructions + +You are working inside an isolated sandbox on a dedicated work branch. The repository checkout is your working directory. + +1. Inspect existing source before changing anything. +2. Implement the complete issue scope. Run the project's test or verification command. +3. If you need human input to proceed safely, emit exactly one line starting with \`NEEDS_INPUT:\` followed by your question, then stop. +4. When implementation and verification are complete, emit exactly one line starting with \`WORK_COMPLETE:\` followed by a one-sentence summary, then stop. +5. Do not merge, deploy, or push. The orchestrator handles the Git lifecycle after you signal completion. +6. Never claim a change you did not observe. Preserve command evidence.`; + +/** + * Build a concise context pack prompt for the OpenCode session. The pack is + * sent as the first task and includes: issue details, project docs, evidence, + * prior artifacts, repository metadata, and operational instructions with the + * needs-input / work-complete marker protocol. + */ +export const buildContextPack = (input: ContextPackInput): string => { + const parts: string[] = [ + `# Issue #${input.issueNumber}: ${input.issueTitle}\n\n${input.issueBody}`, + ]; + + if (input.contextFiles.length > 0) { + parts.push( + section( + "Project Context", + input.contextFiles.map( + (file) => `### ${file.label}\n\n\`\`\`\n${file.content}\n\`\`\`` + ) + ) + ); + } + + if (input.evidence.length > 0) { + parts.push( + section( + "Supporting Evidence", + input.evidence.map((item) => `- **${item.source}**: ${item.content}`) + ) + ); + } + + if (input.artifacts.length > 0) { + parts.push( + section( + "Previous Work Artifacts", + input.artifacts.map( + (artifact) => + `### ${artifact.path}\n\n\`\`\`\n${artifact.content}\n\`\`\`` + ) + ) + ); + } + + parts.push( + section("Repository", [ + `- Name: ${input.repositoryMetadata.repositoryName}`, + `- URL: ${input.repositoryMetadata.repositoryUrl}`, + `- Base branch: ${input.repositoryMetadata.baseBranch}`, + ]), + OPERATIONAL_INSTRUCTIONS + ); + + return parts.filter((part) => part.length > 0).join("\n\n---\n\n"); +}; diff --git a/packages/agents/src/orb/git-adapter.ts b/packages/agents/src/orb/git-adapter.ts new file mode 100644 index 0000000..8e53d8d --- /dev/null +++ b/packages/agents/src/orb/git-adapter.ts @@ -0,0 +1,81 @@ +import { runPostRunGiteaLifecycle } from "../git/gitea"; +import type { GitCommandRunner, GiteaTransport } from "../git/gitea"; +import type { + GitLifecyclePort, + GitLifecycleResult, + GitPublishInput, + OrbRunPort, +} from "./ports"; + +// --------------------------------------------------------------------------- +// OrbGitCommandRunner — runs git commands inside the Orb sandbox +// +// Adapts the OrbRunPort.executeCommand interface to the GitCommandRunner +// shape expected by runPostRunGiteaLifecycle. Commands execute inside the +// Docker sandbox where OpenCode made its changes. +// --------------------------------------------------------------------------- + +const createOrbGitRunner = (orb: OrbRunPort): GitCommandRunner => ({ + run( + command: string, + options?: { readonly cwd?: string; readonly env?: Record } + ) { + const cwd = options?.cwd ?? "/mnt/sandbox/repository"; + return orb.executeCommand(command, cwd); + }, +}); + +// --------------------------------------------------------------------------- +// Git lifecycle adapter factory +// +// Creates a GitLifecyclePort bound to one Orb run. The command runner executes +// git inside that Orb's sandbox; the Gitea transport creates PRs via HTTP. +// This is the application-layer entry point — no interactive Flue shell needed. +// --------------------------------------------------------------------------- + +export const createGitLifecyclePort = ( + orb: OrbRunPort, + transport: GiteaTransport +): GitLifecyclePort => { + const runner = createOrbGitRunner(orb); + + return { + async publish(input: GitPublishInput): Promise { + const result = await runPostRunGiteaLifecycle({ + baseBranch: input.baseBranch, + body: `Verified changes for project issue #${input.issueNumber}. Merge remains a manual review action.`, + ...(input.commitMessage === undefined + ? {} + : { commitMessage: input.commitMessage }), + issueNumber: input.issueNumber, + issueTitle: input.issueTitle, + repositoryPath: input.repositoryPath, + runner, + title: `Issue #${input.issueNumber}: ${input.issueTitle}`, + transport, + verification: input.verification, + workspace: input.workspace, + }); + + return { + baseBranch: result.baseBranch, + branch: result.branch, + ...(result.commitSha === undefined + ? {} + : { commitSha: result.commitSha }), + ...(result.pullRequest === undefined + ? {} + : { + pullRequest: { + baseBranch: result.pullRequest.baseBranch, + branch: result.pullRequest.branch, + number: result.pullRequest.number, + status: result.pullRequest.status, + url: result.pullRequest.url, + }, + }), + status: result.status, + }; + }, + }; +}; diff --git a/packages/agents/src/orb/index.ts b/packages/agents/src/orb/index.ts index 046a564..dd47506 100644 --- a/packages/agents/src/orb/index.ts +++ b/packages/agents/src/orb/index.ts @@ -5,3 +5,9 @@ export * from "./docker-sandbox"; export * from "./opencode-config"; export * from "./permission-policy"; export * from "./runtime"; +export * from "./ports"; +export * from "./project-events"; +export * from "./context-pack"; +export * from "./orb-project-manager"; +export * from "./orb-adapter"; +export * from "./git-adapter"; diff --git a/packages/agents/src/orb/orb-adapter.ts b/packages/agents/src/orb/orb-adapter.ts new file mode 100644 index 0000000..97bfcbb --- /dev/null +++ b/packages/agents/src/orb/orb-adapter.ts @@ -0,0 +1,132 @@ +import { Effect } from "effect"; + +import type { OrbEvent } from "./events"; +import type { + CommandResult, + OrbAdapter, + OrbCreatePortInput, + OrbRunPort, + PrepareRepoInput, +} from "./ports"; +import { OrbRuntime } from "./runtime"; +import type { OrbEnv, OrbHandle } from "./runtime"; + +// --------------------------------------------------------------------------- +// RealOrbRun — wraps an OrbHandle behind the OrbRunPort interface +// +// Effect-based Orb methods are run to Promise here so the orchestrator and +// tests work with plain async/await. Effect failures surface as rejections. +// --------------------------------------------------------------------------- + +class RealOrbRun implements OrbRunPort { + private readonly listeners = new Set<(event: OrbEvent) => void>(); + private unsubscribe: (() => void) | null = null; + private readonly handle: OrbHandle; + + constructor(handle: OrbHandle) { + this.handle = handle; + } + + /** Called once after the handle-level listener is wired. */ + _setUnsubscribe(fn: () => void): void { + this.unsubscribe = fn; + } + + get orbId(): string { + return this.handle.id; + } + + get runId(): string { + return this.handle.runId; + } + + get sessionId(): string | undefined { + return this.handle.currentSessionId; + } + + get state(): string { + return this.handle.state; + } + + onEvent = (listener: (event: OrbEvent) => void): (() => void) => { + this.listeners.add(listener); + return () => { + this.listeners.delete(listener); + }; + }; + + /** Forward an OrbHandle event to all port-level listeners. */ + forwardEvent = (event: OrbEvent): void => { + for (const listener of this.listeners) { + listener(event); + } + }; + + prepareRepository = async (input: PrepareRepoInput): Promise => { + await Effect.runPromise(this.handle.prepareRepository(input)); + }; + + openSession = async (): Promise => + (await Effect.runPromise(this.handle.openSession())) as string; + + sendTask = async (prompt: string): Promise => + await Effect.runPromise(this.handle.sendTask(prompt)); + + executeCommand = async ( + command: string, + cwd?: string + ): Promise => { + const result = (await Effect.runPromise( + this.handle.executeCommand({ + command, + ...(cwd === undefined ? {} : { cwd }), + }) + )) as { exitCode: number | null; stderr: string; stdout: string }; + return { + exitCode: result.exitCode ?? -1, + stderr: result.stderr, + stdout: result.stdout, + }; + }; + + cancel = async (): Promise => { + await Effect.runPromise(this.handle.cancel().pipe(Effect.ignore)); + }; + + dispose = async (): Promise => { + this.unsubscribe?.(); + this.listeners.clear(); + await Effect.runPromise(this.handle.dispose().pipe(Effect.ignore)); + }; +} + +// --------------------------------------------------------------------------- +// Real Orb adapter — wraps OrbRuntime.createOrb behind the OrbAdapter port +// --------------------------------------------------------------------------- + +export const createOrbAdapter = (env?: OrbEnv): OrbAdapter => { + const runtime = new OrbRuntime(env); + + return { + createOrb: async (input: OrbCreatePortInput): Promise => { + const handle = await Effect.runPromise( + runtime.createOrb({ + context: input.context, + docker: input.docker, + gateway: input.gateway, + identity: input.identity, + }) + ); + + const run = new RealOrbRun(handle); + + // Bridge OrbHandle events to port listeners via a single forwarding point. + const unsubscribe = handle.onEvent((event) => { + run.forwardEvent(event); + }); + run._setUnsubscribe(unsubscribe); + + return run; + }, + }; +}; diff --git a/packages/agents/src/orb/orb-project-manager.live.test.ts b/packages/agents/src/orb/orb-project-manager.live.test.ts new file mode 100644 index 0000000..a9d5395 --- /dev/null +++ b/packages/agents/src/orb/orb-project-manager.live.test.ts @@ -0,0 +1,132 @@ +/* eslint-disable no-console -- integration test is a CLI-style probe */ +/** + * Opt-in live integration test for the Orb-wired project-manager. + * + * Only runs when ORB_PM_INTEGRATION=1 is set. Requires: + * - Docker daemon + * - Model gateway credentials (ORB_GATEWAY_* or AGENT_MODEL_*) + * - A local Gitea instance with a test repo (optional — set GITEA_*) + * + * This test exercises the full flow: Orb creation, repo preparation, session + * open, model turn, Git lifecycle (or skip if Gitea is not configured). + * + * Usage: + * ORB_PM_INTEGRATION=1 \ + * ORB_GATEWAY_API_KEY=... \ + * ORB_GATEWAY_BASE_URL=https://ai.example.com/v1 \ + * ORB_GATEWAY_MODEL=glm-5.2 \ + * ORB_GATEWAY_PROVIDER=cheaptricks \ + * bun test packages/agents/src/orb/orb-project-manager.live.test.ts + */ +import { describe, expect, it } from "vitest"; + +import { createGiteaHttpTransport } from "../git/gitea"; +import { createGitLifecyclePort } from "./git-adapter"; +import { createOrbAdapter } from "./orb-adapter"; +import { OrbProjectManager } from "./orb-project-manager"; + +const env = (key: string): string | undefined => process.env[key]; +const isLive = env("ORB_PM_INTEGRATION") === "1"; + +const resolveGateway = () => { + const apiKey = env("ORB_GATEWAY_API_KEY") ?? env("AGENT_MODEL_API_KEY"); + const baseUrl = env("ORB_GATEWAY_BASE_URL") ?? env("AGENT_MODEL_BASE_URL"); + const model = env("ORB_GATEWAY_MODEL") ?? env("AGENT_MODEL_NAME"); + const provider = env("ORB_GATEWAY_PROVIDER") ?? env("AGENT_MODEL_PROVIDER"); + if (!apiKey || !baseUrl || !model || !provider) { + return null; + } + return { apiKey, baseUrl, model, provider }; +}; + +const hasGitea = () => + env("GITEA_URL") !== undefined && env("GITEA_TOKEN") !== undefined; + +describe.skipIf(!isLive)("OrbProjectManager live integration", () => { + it("creates an Orb, prepares repo, sends a model turn, and projects events", async () => { + const gateway = resolveGateway(); + expect(gateway, "Gateway credentials required").not.toBeNull(); + + const adapter = createOrbAdapter(); + const events: { type: string; text?: string }[] = []; + + const pm = new OrbProjectManager({ + createGitLifecycle: (orb) => + createGitLifecyclePort( + orb, + createGiteaHttpTransport({ + baseUrl: env("GITEA_URL") ?? "http://localhost:3000", + token: env("GITEA_TOKEN") ?? "", + }) + ), + onProjectEvent: (e) => events.push({ text: e.text, type: e.type }), + orbAdapter: adapter, + }); + + const result = await pm.startIssue({ + baseBranch: "main", + branchName: `work/orb-pm-test-${Date.now()}`, + context: { + artifacts: [], + contextFiles: [], + issueBody: "Reply with WORK_COMPLETE: hello world test passed", + issueTitle: "Integration: model echo", + repositoryUrl: undefined, + }, + contextPack: { + artifacts: [], + contextFiles: [], + evidence: [], + issueBody: "Reply with WORK_COMPLETE: hello world test passed", + issueNumber: 1, + issueTitle: "Integration: model echo", + repositoryMetadata: { + baseBranch: "main", + repositoryName: "orb-pm-test", + repositoryUrl: "local", + }, + }, + docker: { + hostWorkspacePath: `/tmp/orb-pm-test-${Date.now()}`, + }, + gateway: gateway ?? { + apiKey: "", + baseUrl: "", + model: "", + provider: "", + }, + issueId: `orb-pm-integration-${Date.now()}`, + issueNumber: 1, + issueTitle: "Integration: model echo", + projectId: "orb-pm-test", + runId: `run-${Date.now()}`, + }); + + expect(result.orbId).toBeDefined(); + expect(result.sessionId).toBeDefined(); + + const eventTypes = events.map((e) => e.type); + expect(eventTypes).toContain("run.started"); + expect(eventTypes).toContain("run.repository_prepared"); + expect(eventTypes).toContain("run.session_opened"); + + // If Gitea is configured, attempt the full Git lifecycle. + if (hasGitea()) { + console.log("[orb-pm] Gitea configured — attempting Git lifecycle..."); + try { + const gitResult = await pm.complete({ + commitMessage: "test: orb project-manager integration", + issueId: result.issueId, + }); + console.log(`[orb-pm] Git lifecycle result: ${gitResult.status}`); + } catch (error) { + console.log( + `[orb-pm] Git lifecycle failed (expected in CI): ${error instanceof Error ? error.message : String(error)}` + ); + } + } + + await pm.cancel(result.issueId); + console.log(`[orb-pm] Events: ${eventTypes.join(", ")}`); + }, 120_000); +}); diff --git a/packages/agents/src/orb/orb-project-manager.test.ts b/packages/agents/src/orb/orb-project-manager.test.ts new file mode 100644 index 0000000..f2bb869 --- /dev/null +++ b/packages/agents/src/orb/orb-project-manager.test.ts @@ -0,0 +1,709 @@ +/* eslint-disable no-non-null-assertion -- test assertions on controlled fakes */ +import { describe, expect, it } from "vitest"; + +import { makeOrbEvent } from "./events"; +import type { OrbEvent } from "./events"; +import { OrbProjectManager, ProjectManagerError } from "./orb-project-manager"; +import type { + GitLifecyclePort, + GitLifecycleResult, + GitPublishInput, + OrbAdapter, + OrbCreatePortInput, + OrbRunPort, + PrepareRepoInput, + ProjectArtifact, +} from "./ports"; +import type { ProjectRunEvent } from "./project-events"; + +// --------------------------------------------------------------------------- +// Fake Orb run — records calls and emits configurable events on sendTask +// --------------------------------------------------------------------------- + +interface FakeOrbConfig { + /** Events to emit when sendTask resolves, mapped by call index. */ + readonly eventsByTurn?: readonly (readonly OrbEvent[])[]; + /** Default events to emit on every sendTask if no per-turn mapping. */ + readonly defaultEvents?: readonly OrbEvent[]; + /** If true, sendTask rejects on the first call. */ + readonly failOnFirstSend?: boolean; + /** If true, prepareRepository rejects. */ + readonly failOnPrepare?: boolean; +} + +class FakeOrbRun implements OrbRunPort { + readonly orbId: string; + readonly runId: string; + sessionId: string | undefined; + state = "running"; + private readonly config: FakeOrbConfig; + + readonly sentTasks: string[] = []; + readonly prepareCalls: PrepareRepoInput[] = []; + cancelCalled = false; + disposeCalled = false; + + private listeners = new Set<(event: OrbEvent) => void>(); + private sendCallCount = 0; + + constructor(orbId: string, runId: string, config: FakeOrbConfig) { + this.orbId = orbId; + this.runId = runId; + this.config = config; + } + + onEvent(listener: (event: OrbEvent) => void): () => void { + this.listeners.add(listener); + return () => { + this.listeners.delete(listener); + }; + } + + private emit(events: readonly OrbEvent[]): void { + for (const event of events) { + for (const listener of this.listeners) { + listener(event); + } + } + } + + prepareRepository(input: PrepareRepoInput): Promise { + this.prepareCalls.push(input); + if (this.config.failOnPrepare) { + return Promise.reject(new Error("Fake: prepareRepository failed")); + } + this.state = "prepared"; + return Promise.resolve(); + } + + openSession(): Promise { + this.sessionId = `session-${this.orbId}`; + this.state = "running"; + return Promise.resolve(this.sessionId); + } + + sendTask(prompt: string): Promise { + this.sentTasks.push(prompt); + const callIndex = this.sendCallCount; + this.sendCallCount += 1; + + if (callIndex === 0 && this.config.failOnFirstSend) { + return Promise.reject(new Error("Fake: sendTask failed on first call")); + } + + const events = + this.config.eventsByTurn?.[callIndex] ?? this.config.defaultEvents ?? []; + this.emit(events); + return Promise.resolve({ ok: true }); + } + + executeCommand( + command: string, + _cwd?: string + ): Promise<{ exitCode: number; stderr: string; stdout: string }> { + void this.config; + return Promise.resolve({ + exitCode: 0, + stderr: "", + stdout: `fake: ${command}`, + }); + } + + cancel(): Promise { + this.cancelCalled = true; + this.state = "cancelled"; + return Promise.resolve(); + } + + dispose(): Promise { + this.disposeCalled = true; + this.state = "disposed"; + return Promise.resolve(); + } +} + +// --------------------------------------------------------------------------- +// Fake Orb adapter +// --------------------------------------------------------------------------- + +const createFakeOrbAdapter = ( + config?: FakeOrbConfig, + counter?: { value: number } +): { adapter: OrbAdapter; runs: FakeOrbRun[] } => { + const runs: FakeOrbRun[] = []; + const cfg = config ?? {}; + const cnt = counter ?? { value: 0 }; + const adapter: OrbAdapter = { + createOrb(_input: OrbCreatePortInput): Promise { + cnt.value += 1; + const orbId = `orb-fake-${cnt.value}`; + const runId = `run-fake-${cnt.value}`; + const run = new FakeOrbRun(orbId, runId, cfg); + runs.push(run); + return Promise.resolve(run); + }, + }; + return { adapter, runs }; +}; + +// --------------------------------------------------------------------------- +// Fake Git lifecycle adapter +// --------------------------------------------------------------------------- + +interface FakeGitConfig { + readonly result?: GitLifecycleResult; + readonly failWith?: Error; + readonly failOnFirstAttempt?: boolean; +} + +const createFakeGitLifecycle = ( + config?: FakeGitConfig +): { port: GitLifecyclePort; publishCalls: GitPublishInput[] } => { + const cfg = config ?? {}; + const publishCalls: GitPublishInput[] = []; + let attemptCount = 0; + + const defaultResult: GitLifecycleResult = { + baseBranch: "main", + branch: "work/issue-42", + commitSha: "abc123def", + pullRequest: { + baseBranch: "main", + branch: "work/issue-42", + number: 7, + status: "open", + url: "https://git.example.com/repo/pulls/7", + }, + status: "pull_request_open", + }; + + const port: GitLifecyclePort = { + publish(input: GitPublishInput): Promise { + publishCalls.push(input); + attemptCount += 1; + + if (cfg.failWith && attemptCount === 1) { + return Promise.reject(cfg.failWith); + } + if (cfg.failOnFirstAttempt && attemptCount === 1) { + return Promise.reject( + new Error("Fake: PR creation failed on first attempt") + ); + } + + return Promise.resolve(cfg.result ?? defaultResult); + }, + }; + + return { port, publishCalls }; +}; + +// --------------------------------------------------------------------------- +// Shared fixtures +// --------------------------------------------------------------------------- + +const baseStartInput = ( + overrides?: Partial +): StartIssueInputTest => ({ + baseBranch: "main", + branchName: "work/issue-42", + context: { + artifacts: [], + contextFiles: [], + issueBody: "Add a hello world endpoint", + issueTitle: "Add hello endpoint", + repositoryUrl: undefined, + }, + contextPack: { + artifacts: [], + contextFiles: [], + evidence: [], + issueBody: "Add a hello world endpoint", + issueNumber: 42, + issueTitle: "Add hello endpoint", + repositoryMetadata: { + baseBranch: "main", + repositoryName: "test-repo", + repositoryUrl: "https://git.example.com/repo", + }, + }, + docker: { hostWorkspacePath: "/tmp/test" }, + gateway: { + apiKey: "test-key", + baseUrl: "https://gw.example.com/v1", + model: "m1", + provider: "p1", + }, + issueId: "issue-42", + issueNumber: 42, + issueTitle: "Add hello endpoint", + projectId: "prj-1", + repositoryPath: "org/repo", + runId: "run-42", + ...overrides, +}); + +type StartIssueInputTest = Parameters[0]; + +const makeMessageEvent = (text: string, sequence: number): OrbEvent => + makeOrbEvent(sequence, "agent_message_completed", { text }); + +const makeCommandEvent = ( + command: string, + exitCode: number, + sequence: number +): OrbEvent => + makeOrbEvent(sequence, "command_executed", { command, exitCode }); + +// =========================================================================== +// TESTS +// =========================================================================== + +describe("OrbProjectManager", () => { + // ----------------------------------------------------------------------- + // 1. First-message start + // ----------------------------------------------------------------------- + + describe("first-message start", () => { + it("creates an Orb run, prepares the repo, opens a session, and sends the context pack", async () => { + const { adapter, runs } = createFakeOrbAdapter({ + defaultEvents: [makeMessageEvent("WORK_COMPLETE: done", 1)], + }); + const events: ProjectRunEvent[] = []; + const pm = new OrbProjectManager({ + createGitLifecycle: () => createFakeGitLifecycle().port, + onProjectEvent: (e) => events.push(e), + orbAdapter: adapter, + }); + + const result = await pm.startIssue(baseStartInput()); + + expect(result.issueId).toBe("issue-42"); + expect(result.status).toBe("completing"); + expect(result.sessionId).toBeDefined(); + expect(runs).toHaveLength(1); + expect(runs[0]!.prepareCalls).toHaveLength(1); + expect(runs[0]!.prepareCalls[0]?.branchName).toBe("work/issue-42"); + expect(runs[0]!.sentTasks).toHaveLength(1); + expect(runs[0]!.sentTasks[0]).toContain("Add hello endpoint"); + + const eventTypes = events.map((e) => e.type); + expect(eventTypes).toContain("run.started"); + expect(eventTypes).toContain("run.repository_prepared"); + expect(eventTypes).toContain("run.session_opened"); + }); + + it("maps OrbEvents into durable project events and forwards them to the sink", async () => { + const { adapter } = createFakeOrbAdapter({ + defaultEvents: [ + makeOrbEvent(1, "tool_call_started", { toolName: "edit_file" }), + makeCommandEvent("npm test", 0, 2), + makeMessageEvent("WORK_COMPLETE: all tests pass", 3), + ], + }); + const events: ProjectRunEvent[] = []; + const pm = new OrbProjectManager({ + createGitLifecycle: () => createFakeGitLifecycle().port, + onProjectEvent: (e) => events.push(e), + orbAdapter: adapter, + }); + + await pm.startIssue(baseStartInput()); + + const types = events.map((e) => e.type); + expect(types).toContain("run.agent_progress"); + expect(types).toContain("run.command_executed"); + expect(types).toContain("run.agent_message"); + + const cmdEvent = events.find((e) => e.type === "run.command_executed"); + expect(cmdEvent?.text).toBe("npm test"); + expect(cmdEvent?.exitCode).toBe(0); + }); + }); + + // ----------------------------------------------------------------------- + // 2. Follow-up forwarding + // ----------------------------------------------------------------------- + + describe("follow-up forwarding", () => { + it("forwards a contextual message to the same OpenCode session", async () => { + const { adapter, runs } = createFakeOrbAdapter({ + eventsByTurn: [ + [makeMessageEvent("NEEDS_INPUT: what name?", 1)], + [makeMessageEvent("WORK_COMPLETE: done", 2)], + ], + }); + const pm = new OrbProjectManager({ + createGitLifecycle: () => createFakeGitLifecycle().port, + orbAdapter: adapter, + }); + + const start = await pm.startIssue(baseStartInput()); + expect(start.status).toBe("needs-input"); + expect(start.needsInputQuestion).toBe("what name?"); + + const followUp = await pm.sendMessage("issue-42", "Name it /hello"); + expect(followUp.status).toBe("completing"); + expect(runs[0]!.sentTasks).toHaveLength(2); + expect(runs[0]!.sentTasks[1]).toBe("Name it /hello"); + }); + + it("rejects a follow-up for a non-existent run", async () => { + const { adapter } = createFakeOrbAdapter(); + const pm = new OrbProjectManager({ + createGitLifecycle: () => createFakeGitLifecycle().port, + orbAdapter: adapter, + }); + + await expect(pm.sendMessage("nope", "hello")).rejects.toThrow(); + try { + await pm.sendMessage("nope", "hello"); + } catch (error) { + expect(error).toBeInstanceOf(ProjectManagerError); + expect((error as ProjectManagerError).reason).toBe("RunNotFound"); + } + }); + + it("rejects a follow-up after cancellation", async () => { + const { adapter } = createFakeOrbAdapter({ + defaultEvents: [makeMessageEvent("NEEDS_INPUT: hmm", 1)], + }); + const pm = new OrbProjectManager({ + createGitLifecycle: () => createFakeGitLifecycle().port, + orbAdapter: adapter, + }); + + await pm.startIssue(baseStartInput()); + await pm.cancel("issue-42"); + + await expect(pm.sendMessage("issue-42", "hello")).rejects.toThrow(); + try { + await pm.sendMessage("issue-42", "hello"); + } catch (error) { + expect(error).toBeInstanceOf(ProjectManagerError); + expect((error as ProjectManagerError).reason).toBe("RunTerminal"); + } + }); + }); + + // ----------------------------------------------------------------------- + // 3. Needs-input handling + // ----------------------------------------------------------------------- + + describe("needs-input handling", () => { + it("detects the needs-input marker and surfaces the question", async () => { + const { adapter } = createFakeOrbAdapter({ + defaultEvents: [ + makeMessageEvent("NEEDS_INPUT: Should I use GET or POST?", 1), + ], + }); + const pm = new OrbProjectManager({ + createGitLifecycle: () => createFakeGitLifecycle().port, + orbAdapter: adapter, + }); + + const result = await pm.startIssue(baseStartInput()); + + expect(result.status).toBe("needs-input"); + expect(result.needsInputQuestion).toBe("Should I use GET or POST?"); + + const events = pm.getRunEvents("issue-42"); + expect(events.some((e) => e.type === "run.needs_input")).toBe(true); + }); + + it("clears the needs-input condition when a follow-up is sent", async () => { + const { adapter } = createFakeOrbAdapter({ + eventsByTurn: [ + [makeMessageEvent("NEEDS_INPUT: clarify?", 1)], + [makeMessageEvent("WORK_COMPLETE: done", 2)], + ], + }); + const pm = new OrbProjectManager({ + createGitLifecycle: () => createFakeGitLifecycle().port, + orbAdapter: adapter, + }); + + await pm.startIssue(baseStartInput()); + expect(pm.getRunStatus("issue-42")).toBe("needs-input"); + + const result = await pm.sendMessage("issue-42", "Use POST"); + expect(result.status).toBe("completing"); + expect(result.needsInputQuestion).toBeUndefined(); + }); + }); + + // ----------------------------------------------------------------------- + // 4. Cancellation + // ----------------------------------------------------------------------- + + describe("cancellation", () => { + it("cancels and disposes the Orb run", async () => { + const { adapter, runs } = createFakeOrbAdapter({ + defaultEvents: [makeMessageEvent("NEEDS_INPUT: wait", 1)], + }); + const pm = new OrbProjectManager({ + createGitLifecycle: () => createFakeGitLifecycle().port, + orbAdapter: adapter, + }); + + await pm.startIssue(baseStartInput()); + await pm.cancel("issue-42"); + + expect(runs[0]!.cancelCalled).toBe(true); + expect(runs[0]!.disposeCalled).toBe(true); + expect(pm.getRunStatus("issue-42")).toBe("cancelled"); + }); + + it("is idempotent — cancelling a non-existent run is a no-op", async () => { + const { adapter } = createFakeOrbAdapter(); + const pm = new OrbProjectManager({ + createGitLifecycle: () => createFakeGitLifecycle().port, + orbAdapter: adapter, + }); + + await expect(pm.cancel("nonexistent")).resolves.toBeUndefined(); + }); + + it("is idempotent — cancelling twice does not error", async () => { + const { adapter } = createFakeOrbAdapter({ + defaultEvents: [makeMessageEvent("working...", 1)], + }); + const pm = new OrbProjectManager({ + createGitLifecycle: () => createFakeGitLifecycle().port, + orbAdapter: adapter, + }); + + await pm.startIssue(baseStartInput()); + await pm.cancel("issue-42"); + await pm.cancel("issue-42"); + + expect(pm.getRunStatus("issue-42")).toBe("cancelled"); + }); + }); + + // ----------------------------------------------------------------------- + // 5. Duplicate-start prevention + // ----------------------------------------------------------------------- + + describe("duplicate-start prevention", () => { + it("does not create a second Orb run for an active issue", async () => { + const { adapter, runs } = createFakeOrbAdapter({ + defaultEvents: [makeMessageEvent("NEEDS_INPUT: hmm", 1)], + }); + const pm = new OrbProjectManager({ + createGitLifecycle: () => createFakeGitLifecycle().port, + orbAdapter: adapter, + }); + + const first = await pm.startIssue(baseStartInput()); + const second = await pm.startIssue(baseStartInput()); + + expect(runs).toHaveLength(1); + expect(second.orbId).toBe(first.orbId); + expect(second.status).toBe("needs-input"); + }); + + it("allows starting a new run after the previous one completed", async () => { + const { adapter, runs } = createFakeOrbAdapter({ + defaultEvents: [makeMessageEvent("WORK_COMPLETE: done", 1)], + }); + const { port: gitPort } = createFakeGitLifecycle(); + const pm = new OrbProjectManager({ + createGitLifecycle: () => gitPort, + orbAdapter: adapter, + }); + + await pm.startIssue(baseStartInput()); + await pm.complete({ issueId: "issue-42" }); + expect(pm.getRunStatus("issue-42")).toBe("completed"); + + // A terminal run allows re-creating via startIssue. + const second = await pm.startIssue(baseStartInput()); + expect(runs.length).toBeGreaterThanOrEqual(2); + expect(second.orbId).not.toBe(runs[0]!.orbId); + }); + }); + + // ----------------------------------------------------------------------- + // 6. Successful PR completion + // ----------------------------------------------------------------------- + + describe("successful PR completion", () => { + it("runs the Git lifecycle, creates a PR, stores artifacts, and marks completed", async () => { + const { adapter } = createFakeOrbAdapter({ + defaultEvents: [makeMessageEvent("WORK_COMPLETE: implemented", 1)], + }); + const artifacts: ProjectArtifact[] = []; + const { port: gitPort, publishCalls } = createFakeGitLifecycle(); + const pm = new OrbProjectManager({ + createGitLifecycle: () => gitPort, + onArtifact: (a) => artifacts.push(a), + orbAdapter: adapter, + }); + + await pm.startIssue(baseStartInput()); + const result = await pm.complete({ issueId: "issue-42" }); + + expect(result.status).toBe("pull_request_open"); + expect(result.pullRequest?.number).toBe(7); + expect(pm.getRunStatus("issue-42")).toBe("completed"); + expect(publishCalls).toHaveLength(1); + expect(publishCalls[0]?.branchName).toBe("work/issue-42"); + + const artifactTypes = artifacts.map((a) => a.type); + expect(artifactTypes).toContain("branch"); + expect(artifactTypes).toContain("commit"); + expect(artifactTypes).toContain("pull_request"); + expect(artifactTypes).toContain("agent_summary"); + }); + + it("marks completed with no_changes when the Git lifecycle reports no changes", async () => { + const { adapter } = createFakeOrbAdapter({ + defaultEvents: [makeMessageEvent("WORK_COMPLETE: nothing to do", 1)], + }); + const { port: gitPort } = createFakeGitLifecycle({ + result: { + baseBranch: "main", + branch: "work/issue-42", + status: "no_changes", + }, + }); + const pm = new OrbProjectManager({ + createGitLifecycle: () => gitPort, + orbAdapter: adapter, + }); + + await pm.startIssue(baseStartInput()); + const result = await pm.complete({ issueId: "issue-42" }); + + expect(result.status).toBe("no_changes"); + expect(pm.getRunStatus("issue-42")).toBe("completed"); + }); + }); + + // ----------------------------------------------------------------------- + // 7. Failed PR creation recovery + // ----------------------------------------------------------------------- + + describe("failed PR creation recovery", () => { + it("marks the run as failed and raises a tagged error when PR creation fails", async () => { + const { adapter } = createFakeOrbAdapter({ + defaultEvents: [makeMessageEvent("WORK_COMPLETE: done", 1)], + }); + const { port: gitPort } = createFakeGitLifecycle({ + failWith: new Error("Remote rejected push"), + }); + const pm = new OrbProjectManager({ + createGitLifecycle: () => gitPort, + orbAdapter: adapter, + }); + + await pm.startIssue(baseStartInput()); + + try { + await pm.complete({ issueId: "issue-42" }); + expect.fail("Should have thrown"); + } catch (error) { + expect(error).toBeInstanceOf(ProjectManagerError); + expect((error as ProjectManagerError).reason).toBe("GitRejection"); + } + expect(pm.getRunStatus("issue-42")).toBe("failed"); + }); + + it("allows retrying completion after a failed attempt (idempotent commit+push)", async () => { + const { adapter } = createFakeOrbAdapter({ + defaultEvents: [makeMessageEvent("WORK_COMPLETE: done", 1)], + }); + const { port: gitPort, publishCalls } = createFakeGitLifecycle({ + failOnFirstAttempt: true, + }); + const pm = new OrbProjectManager({ + createGitLifecycle: () => gitPort, + orbAdapter: adapter, + }); + + await pm.startIssue(baseStartInput()); + + try { + await pm.complete({ issueId: "issue-42" }); + } catch { + // expected first-attempt failure + } + expect(pm.getRunStatus("issue-42")).toBe("failed"); + + // Retry: the Orb run still exists, the branch is preserved. + const result = await pm.complete({ issueId: "issue-42" }); + expect(result.status).toBe("pull_request_open"); + expect(pm.getRunStatus("issue-42")).toBe("completed"); + expect(publishCalls).toHaveLength(2); + }); + }); + + // ----------------------------------------------------------------------- + // 8. Infrastructure failure mapping + // ----------------------------------------------------------------------- + + describe("infrastructure failure mapping", () => { + it("maps a failed Orb creation to InfrastructureFailure", async () => { + const failingAdapter: OrbAdapter = { + createOrb: () => Promise.reject(new Error("Docker daemon unavailable")), + }; + const pm = new OrbProjectManager({ + createGitLifecycle: () => createFakeGitLifecycle().port, + orbAdapter: failingAdapter, + }); + + try { + await pm.startIssue(baseStartInput()); + expect.fail("Should have thrown"); + } catch (error) { + expect(error).toBeInstanceOf(ProjectManagerError); + expect((error as ProjectManagerError).reason).toBe( + "InfrastructureFailure" + ); + } + }); + + it("maps a failed prepareRepository to InfrastructureFailure", async () => { + const { adapter } = createFakeOrbAdapter({ + failOnPrepare: true, + }); + const pm = new OrbProjectManager({ + createGitLifecycle: () => createFakeGitLifecycle().port, + orbAdapter: adapter, + }); + + try { + await pm.startIssue(baseStartInput()); + expect.fail("Should have thrown"); + } catch (error) { + expect(error).toBeInstanceOf(ProjectManagerError); + expect((error as ProjectManagerError).reason).toBe( + "InfrastructureFailure" + ); + } + }); + + it("does not create duplicate events from idempotent completion", async () => { + const { adapter } = createFakeOrbAdapter({ + defaultEvents: [makeMessageEvent("WORK_COMPLETE: done", 1)], + }); + const { port: gitPort } = createFakeGitLifecycle(); + const pm = new OrbProjectManager({ + createGitLifecycle: () => gitPort, + orbAdapter: adapter, + }); + + await pm.startIssue(baseStartInput()); + await pm.complete({ issueId: "issue-42" }); + + const eventsBefore = pm.getRunEvents("issue-42").length; + const result = await pm.complete({ issueId: "issue-42" }); + const eventsAfter = pm.getRunEvents("issue-42").length; + + // Idempotent: returns cached result without duplicating events. + expect(result.status).toBe("pull_request_open"); + expect(eventsAfter).toBe(eventsBefore); + }); + }); +}); diff --git a/packages/agents/src/orb/orb-project-manager.ts b/packages/agents/src/orb/orb-project-manager.ts new file mode 100644 index 0000000..0e87432 --- /dev/null +++ b/packages/agents/src/orb/orb-project-manager.ts @@ -0,0 +1,636 @@ +/* eslint-disable max-classes-per-file -- domain errors are grouped by concern. */ +import { Schema } from "effect"; + +import { buildContextPack } from "./context-pack"; +import type { ContextPackInput } from "./context-pack"; +import type { OrbEvent } from "./events"; +import type { + GitLifecyclePort, + GitLifecycleResult, + OrbAdapter, + OrbCreatePortInput, + OrbRunPort, + ProjectArtifact, + RunStatus, +} from "./ports"; +import { isWorkComplete, mapOrbEvent } from "./project-events"; +import type { ProjectRunEvent } from "./project-events"; + +// --------------------------------------------------------------------------- +// Tagged errors — the failure-mapping surface +// --------------------------------------------------------------------------- + +export const ProjectManagerErrorReason = Schema.Literals([ + "RunNotFound", + "SessionNotReady", + "RunTerminal", + "DuplicateActiveRun", + "InfrastructureFailure", + "NeedsInput", + "GitRejection", + "PullRequestFailure", + "Cancelled", + "UnrecoverableFailure", +]); +export type ProjectManagerErrorReason = typeof ProjectManagerErrorReason.Type; + +export class ProjectManagerError extends Schema.TaggedErrorClass()( + "ProjectManagerError", + { + issueId: Schema.String, + message: Schema.String, + reason: ProjectManagerErrorReason, + } +) {} + +// --------------------------------------------------------------------------- +// Active run record — one per managed issue +// --------------------------------------------------------------------------- + +interface ActiveRun { + readonly issueId: string; + readonly orbId: string; + readonly runId: string; + readonly orb: OrbRunPort; + sessionId: string | undefined; + status: RunStatus; + readonly projectEvents: ProjectRunEvent[]; + result: GitLifecycleResult | undefined; + needsInputQuestion: string | undefined; + readonly contextPack: string; + lastTurnEventIndex: number; + readonly baseBranch: string; + readonly branchName: string; + readonly issueNumber: number; + readonly issueTitle: string; + readonly repositoryPath: string; + readonly workspacePath: string; +} + +// --------------------------------------------------------------------------- +// Dependencies injected into the orchestrator +// --------------------------------------------------------------------------- + +export interface OrbProjectManagerDeps { + readonly orbAdapter: OrbAdapter; + readonly createGitLifecycle: (orb: OrbRunPort) => GitLifecyclePort; + readonly onProjectEvent?: (event: ProjectRunEvent) => void; + readonly onArtifact?: (artifact: ProjectArtifact) => void; +} + +// --------------------------------------------------------------------------- +// Input types for the orchestration API +// --------------------------------------------------------------------------- + +export interface StartIssueInput { + readonly issueId: string; + readonly projectId: string; + readonly runId: string; + readonly context: OrbCreatePortInput["context"]; + readonly gateway: OrbCreatePortInput["gateway"]; + readonly docker: OrbCreatePortInput["docker"]; + readonly baseBranch: string; + readonly branchName: string; + readonly contextPack: ContextPackInput; + readonly workspacePath?: string; + readonly repositoryPath?: string; + readonly issueNumber?: number; + readonly issueTitle?: string; +} + +export interface StartIssueResult { + readonly issueId: string; + readonly orbId: string; + readonly runId: string; + readonly sessionId: string | undefined; + readonly status: RunStatus; + readonly needsInputQuestion?: string; +} + +export interface SendMessageResult { + readonly issueId: string; + readonly status: RunStatus; + readonly needsInputQuestion?: string; +} + +export interface CompleteInput { + readonly issueId: string; + readonly verification?: "passed" | "failed" | "not-run"; + readonly commitMessage?: string; +} + +// --------------------------------------------------------------------------- +// Helpers +// --------------------------------------------------------------------------- + +const isTerminal = (status: RunStatus): boolean => + status === "completed" || status === "failed" || status === "cancelled"; + +const isSessionValid = (run: ActiveRun): boolean => + run.sessionId !== undefined && !isTerminal(run.status); + +const timestamp = () => new Date().toISOString(); + +const wrapError = ( + error: unknown, + issueId: string, + reason: ProjectManagerErrorReason, + fallback: string +): ProjectManagerError => { + const message = error instanceof Error ? error.message : String(error); + return new ProjectManagerError({ + issueId, + message: message.length > 0 ? message : fallback, + reason, + }); +}; + +const isGitRejection = (error: unknown): boolean => { + if (!(error instanceof Error)) { + return false; + } + const message = error.message.toLowerCase(); + return ( + message.includes("rejected") || + message.includes("authentication") || + message.includes("permission denied") || + message.includes("remote") + ); +}; + +// --------------------------------------------------------------------------- +// OrbProjectManager — thin orchestration agent over the merged Orb runtime +// +// Responsibilities: +// - Create or resume an Orb run per issue (idempotent) +// - Assemble a context pack and send it as the implementation objective +// - Project OrbEvents into durable ProjectRunEvents (no parallel event system) +// - Detect needs-input and work-complete conditions +// - Forward follow-up messages to the same OpenCode session +// - Drive the Git publish lifecycle on completion (never auto-merge) +// - Store branch/commit/diff/PR/summary as project artifacts +// --------------------------------------------------------------------------- + +export class OrbProjectManager { + private readonly activeRuns = new Map(); + + private readonly deps: OrbProjectManagerDeps; + + constructor(deps: OrbProjectManagerDeps) { + this.deps = deps; + } + + // ----------------------------------------------------------------------- + // Public state queries + // ----------------------------------------------------------------------- + + getRunStatus(issueId: string): RunStatus | undefined { + return this.activeRuns.get(issueId)?.status; + } + + getRunEvents(issueId: string): readonly ProjectRunEvent[] { + return this.activeRuns.get(issueId)?.projectEvents ?? []; + } + + getRunResult(issueId: string): GitLifecycleResult | undefined { + return this.activeRuns.get(issueId)?.result; + } + + // ----------------------------------------------------------------------- + // Start or resume an Orb run for an issue + // ----------------------------------------------------------------------- + + async startIssue(input: StartIssueInput): Promise { + const existing = this.activeRuns.get(input.issueId); + + // Idempotency: a still-active run is reused, never duplicated. + if (existing && !isTerminal(existing.status)) { + return { + issueId: input.issueId, + needsInputQuestion: existing.needsInputQuestion, + orbId: existing.orbId, + runId: existing.runId, + sessionId: existing.sessionId, + status: existing.status, + }; + } + + let orb: OrbRunPort; + try { + orb = await this.deps.orbAdapter.createOrb({ + context: input.context, + docker: input.docker, + gateway: input.gateway, + identity: { + projectId: input.projectId, + runId: input.runId, + workUnitId: input.issueId, + }, + }); + } catch (error) { + throw wrapError( + error, + input.issueId, + "InfrastructureFailure", + "Failed to create Orb run" + ); + } + + const contextPack = buildContextPack(input.contextPack); + + const run: ActiveRun = { + baseBranch: input.baseBranch, + branchName: input.branchName, + contextPack, + issueId: input.issueId, + issueNumber: input.issueNumber ?? 0, + issueTitle: input.issueTitle ?? input.issueId, + lastTurnEventIndex: 0, + needsInputQuestion: undefined, + orb, + orbId: orb.orbId, + projectEvents: [], + repositoryPath: input.repositoryPath ?? "", + result: undefined, + runId: orb.runId, + sessionId: undefined, + status: "starting", + workspacePath: input.workspacePath ?? "/mnt/sandbox/repository", + }; + this.activeRuns.set(input.issueId, run); + + // Subscribe to OrbEvents and project them into durable ProjectRunEvents. + orb.onEvent((orbEvent: OrbEvent) => { + this.processOrbEvent(orbEvent, run); + }); + + this.emitProjectEvent(run, "run.started", { + text: `Started Orb run for issue ${input.issueId}`, + }); + + try { + await orb.prepareRepository({ + baseBranch: input.baseBranch, + branchName: input.branchName, + }); + this.emitProjectEvent(run, "run.repository_prepared", { + text: `Repository prepared on branch ${input.branchName}`, + }); + + const sessionId = await orb.openSession(); + run.sessionId = sessionId; + run.status = "working"; + this.emitProjectEvent(run, "run.session_opened", { + text: `Session ${sessionId} opened`, + }); + + // Record turn boundary before sending the implementation objective. + run.lastTurnEventIndex = run.projectEvents.length; + + // Send the implementation objective (the assembled context pack). + await orb.sendTask(contextPack); + + // After the turn, check for needs-input or work-complete signals. + OrbProjectManager.evaluateTurnOutcome(run); + } catch (error) { + if (error instanceof ProjectManagerError) { + throw error; + } + throw wrapError( + error, + input.issueId, + "InfrastructureFailure", + "Orb run failed during startup" + ); + } + + return { + issueId: input.issueId, + needsInputQuestion: run.needsInputQuestion, + orbId: run.orbId, + runId: run.runId, + sessionId: run.sessionId, + status: run.status, + }; + } + + // ----------------------------------------------------------------------- + // Forward a follow-up message to the same OpenCode session + // ----------------------------------------------------------------------- + + async sendMessage( + issueId: string, + message: string + ): Promise { + const run = this.activeRuns.get(issueId); + if (!run) { + throw new ProjectManagerError({ + issueId, + message: `No active run for issue ${issueId}`, + reason: "RunNotFound", + }); + } + if (isTerminal(run.status)) { + throw new ProjectManagerError({ + issueId, + message: `Run for issue ${issueId} is in terminal state ${run.status}`, + reason: "RunTerminal", + }); + } + if (!isSessionValid(run)) { + throw new ProjectManagerError({ + issueId, + message: `Session is not open for issue ${issueId}`, + reason: "SessionNotReady", + }); + } + + // Clear any prior needs-input condition and record turn boundary. + run.needsInputQuestion = undefined; + run.status = "working"; + run.lastTurnEventIndex = run.projectEvents.length; + + try { + await run.orb.sendTask(message); + OrbProjectManager.evaluateTurnOutcome(run); + } catch (error) { + throw wrapError( + error, + issueId, + "InfrastructureFailure", + "Failed to forward message to OpenCode session" + ); + } + + return { + issueId, + needsInputQuestion: run.needsInputQuestion, + status: run.status, + }; + } + + // ----------------------------------------------------------------------- + // Cancel — terminate OpenCode and sandbox, then dispose + // ----------------------------------------------------------------------- + + async cancel(issueId: string): Promise { + const run = this.activeRuns.get(issueId); + if (!run) { + // Idempotent cancel: a non-existent run is already "cancelled". + return; + } + if (run.status === "cancelled") { + return; + } + + try { + await run.orb.cancel(); + } catch { + // best-effort: proceed to dispose even if cancel failed + } + try { + await run.orb.dispose(); + } catch { + // best-effort cleanup + } + + run.status = "cancelled"; + this.emitProjectEvent(run, "run.cancelled", { + text: "Run cancelled by user", + }); + } + + // ----------------------------------------------------------------------- + // Complete — run Git publish lifecycle, store artifacts, mark completed + // ----------------------------------------------------------------------- + + async complete(input: CompleteInput): Promise { + const run = this.activeRuns.get(input.issueId); + if (!run) { + throw new ProjectManagerError({ + issueId: input.issueId, + message: `No active run for issue ${input.issueId}`, + reason: "RunNotFound", + }); + } + + // Idempotency: a completed run with an existing result is returned as-is. + if (run.status === "completed" && run.result !== undefined) { + return run.result; + } + + if (run.status === "cancelled") { + throw new ProjectManagerError({ + issueId: input.issueId, + message: "Cannot complete a cancelled run", + reason: "Cancelled", + }); + } + + const git = this.deps.createGitLifecycle(run.orb); + const verification = input.verification ?? "passed"; + + run.status = "completing"; + + let gitResult: GitLifecycleResult; + try { + gitResult = await git.publish({ + baseBranch: run.baseBranch, + branchName: run.branchName, + commitMessage: input.commitMessage, + issueNumber: run.issueNumber, + issueTitle: run.issueTitle, + repositoryPath: run.repositoryPath, + verification, + workspace: run.workspacePath, + }); + } catch (error) { + run.status = "failed"; + const reason = isGitRejection(error) + ? "GitRejection" + : "PullRequestFailure"; + this.emitProjectEvent(run, "run.failed", { + text: error instanceof Error ? error.message : String(error), + }); + throw wrapError( + error, + input.issueId, + reason, + "Git publish lifecycle failed" + ); + } + + run.result = gitResult; + this.storeArtifacts(run, gitResult); + + // Mark completed only when a PR exists or a verified no-change result. + if ( + (gitResult.status === "pull_request_open" && gitResult.pullRequest) || + gitResult.status === "no_changes" + ) { + run.status = "completed"; + this.emitProjectEvent(run, "run.completed", { + text: gitResult.pullRequest + ? `PR #${gitResult.pullRequest.number} created: ${gitResult.pullRequest.url}` + : "No changes to publish", + }); + } else { + run.status = "failed"; + this.emitProjectEvent(run, "run.failed", { + text: `Git lifecycle stopped at ${gitResult.status} without a pull request`, + }); + throw new ProjectManagerError({ + issueId: input.issueId, + message: `Git lifecycle did not produce a pull request (status: ${gitResult.status})`, + reason: "PullRequestFailure", + }); + } + + return gitResult; + } + + // ----------------------------------------------------------------------- + // Dispose all runs (for graceful shutdown) + // ----------------------------------------------------------------------- + + async disposeAll(): Promise { + const issues = [...this.activeRuns.keys()]; + await Promise.allSettled( + issues.map(async (issueId) => { + const run = this.activeRuns.get(issueId); + if (run && !isTerminal(run.status)) { + try { + await run.orb.dispose(); + } catch { + // best-effort + } + } + }) + ); + } + + // ----------------------------------------------------------------------- + // Internal: OrbEvent processing + // ----------------------------------------------------------------------- + + private processOrbEvent(orbEvent: OrbEvent, run: ActiveRun): void { + const projectEvent = mapOrbEvent(orbEvent, run.issueId, run.runId); + if (projectEvent !== undefined) { + run.projectEvents.push(projectEvent); + this.deps.onProjectEvent?.(projectEvent); + + if ( + projectEvent.type === "run.needs_input" && + run.needsInputQuestion === undefined + ) { + run.needsInputQuestion = projectEvent.text; + } + } + } + + // ----------------------------------------------------------------------- + // Internal: evaluate the outcome of a completed model turn + // ----------------------------------------------------------------------- + + private static evaluateTurnOutcome(run: ActiveRun): void { + // Only scan events from the current turn (after the last turn boundary). + const turnEvents = run.projectEvents.slice(run.lastTurnEventIndex); + + // Check for needs-input: the mapOrbEvent step already extracted the marker + // into a run.needs_input event with the question text. A run.needs_input + // event IS the signal — no need to re-extract the marker from its text. + const needsInputEvent = [...turnEvents] + .toReversed() + .find((event) => event.type === "run.needs_input"); + + if (needsInputEvent !== undefined) { + run.status = "needs-input"; + run.needsInputQuestion = needsInputEvent.text ?? "Agent requires input"; + return; + } + + // Check for work-complete marker in agent messages from this turn. + const turnMessages = turnEvents.filter( + (event) => event.type === "run.agent_message" + ); + + const hasWorkComplete = turnMessages.some( + (event) => event.text !== undefined && isWorkComplete(event.text) + ); + + if (hasWorkComplete && run.status === "working") { + run.status = "completing"; + } + } + + // ----------------------------------------------------------------------- + // Internal: emit a synthetic project event (not derived from an OrbEvent) + // ----------------------------------------------------------------------- + + private emitProjectEvent( + run: ActiveRun, + type: ProjectRunEvent["type"], + fields: { text?: string; exitCode?: number; toolName?: string } + ): void { + const event: ProjectRunEvent = { + exitCode: fields.exitCode, + issueId: run.issueId, + runId: run.runId, + sequence: run.projectEvents.length + 1, + text: fields.text, + timestamp: timestamp(), + toolName: fields.toolName, + type, + }; + run.projectEvents.push(event); + this.deps.onProjectEvent?.(event); + } + + // ----------------------------------------------------------------------- + // Internal: store artifacts from the Git lifecycle result + // ----------------------------------------------------------------------- + + private storeArtifacts(run: ActiveRun, result: GitLifecycleResult): void { + const ts = timestamp(); + const base = { issueId: run.issueId, runId: run.runId }; + + const emitArtifact = ( + type: ProjectArtifact["type"], + path: string, + content: string + ): void => { + const artifact: ProjectArtifact = { + ...base, + content, + path, + timestamp: ts, + type, + }; + this.deps.onArtifact?.(artifact); + }; + + emitArtifact("branch", "branch.txt", result.branch); + + if (result.commitSha) { + emitArtifact("commit", "commit.txt", result.commitSha); + } + + if (result.pullRequest) { + emitArtifact( + "pull_request", + "pull_request.json", + JSON.stringify(result.pullRequest, null, 2) + ); + } + + const lastMessage = [...run.projectEvents] + .toReversed() + .find( + (event) => + event.type === "run.agent_message" || event.type === "run.needs_input" + ); + if (lastMessage?.text) { + emitArtifact("agent_summary", "summary.md", lastMessage.text); + } + } +} diff --git a/packages/agents/src/orb/ports.ts b/packages/agents/src/orb/ports.ts new file mode 100644 index 0000000..aff98fd --- /dev/null +++ b/packages/agents/src/orb/ports.ts @@ -0,0 +1,136 @@ +/* eslint-disable max-classes-per-file -- domain errors are grouped by concern. */ +import { Schema } from "effect"; + +import type { + OrbIdentity, + OrbModelGatewayConfig, + OrbProjectContext, +} from "./domain"; +import type { OrbEvent } from "./events"; + +// --------------------------------------------------------------------------- +// Command result — shared shape for sandbox command execution +// --------------------------------------------------------------------------- + +export interface CommandResult { + readonly exitCode: number; + readonly stderr: string; + readonly stdout: string; +} + +// --------------------------------------------------------------------------- +// Orb port — abstracts OrbRuntime/OrbHandle for testability +// --------------------------------------------------------------------------- + +export interface PrepareRepoInput { + readonly baseBranch?: string; + readonly branchName?: string; +} + +export interface OrbRunPort { + readonly orbId: string; + readonly runId: string; + readonly sessionId: string | undefined; + readonly state: string; + readonly onEvent: (listener: (event: OrbEvent) => void) => () => void; + readonly prepareRepository: (input: PrepareRepoInput) => Promise; + readonly openSession: () => Promise; + readonly sendTask: (prompt: string) => Promise; + readonly executeCommand: ( + command: string, + cwd?: string + ) => Promise; + readonly cancel: () => Promise; + readonly dispose: () => Promise; +} + +export interface OrbCreatePortInput { + readonly context: OrbProjectContext; + readonly gateway: OrbModelGatewayConfig; + readonly identity: OrbIdentity; + readonly docker: { + readonly hostWorkspacePath: string; + readonly containerName?: string; + readonly image?: string; + }; +} + +export interface OrbAdapter { + readonly createOrb: (input: OrbCreatePortInput) => Promise; +} + +// --------------------------------------------------------------------------- +// Git lifecycle port — abstracts the Gitea lifecycle for testability +// --------------------------------------------------------------------------- + +export interface GitPublishInput { + readonly workspace: string; + readonly baseBranch: string; + readonly branchName: string; + readonly issueNumber: number; + readonly issueTitle: string; + readonly repositoryPath: string; + readonly commitMessage?: string; + readonly verification: "passed" | "failed" | "not-run"; +} + +export interface GitPullRequestMeta { + readonly baseBranch: string; + readonly branch: string; + readonly number: number; + readonly status: "open" | "closed" | "merged"; + readonly url: string; +} + +export interface GitLifecycleResult { + readonly baseBranch: string; + readonly branch: string; + readonly commitSha?: string; + readonly pullRequest?: GitPullRequestMeta; + readonly status: + | "no_changes" + | "committed" + | "pushed" + | "pull_request_open" + | "failed"; +} + +export interface GitLifecyclePort { + readonly publish: (input: GitPublishInput) => Promise; +} + +// --------------------------------------------------------------------------- +// Project artifact — durable output stored after a run +// --------------------------------------------------------------------------- + +export const ProjectArtifactSchema = Schema.Struct({ + content: Schema.String, + issueId: Schema.String, + path: Schema.String, + runId: Schema.String, + timestamp: Schema.String, + type: Schema.Literals([ + "branch", + "commit", + "diff", + "verification_report", + "pull_request", + "agent_summary", + ]), +}); +export type ProjectArtifact = typeof ProjectArtifactSchema.Type; + +// --------------------------------------------------------------------------- +// Orchestration status for a managed issue run +// --------------------------------------------------------------------------- + +export const RunStatus = Schema.Literals([ + "starting", + "working", + "needs-input", + "completing", + "completed", + "failed", + "cancelled", +]); +export type RunStatus = typeof RunStatus.Type; diff --git a/packages/agents/src/orb/project-events.test.ts b/packages/agents/src/orb/project-events.test.ts new file mode 100644 index 0000000..0724e26 --- /dev/null +++ b/packages/agents/src/orb/project-events.test.ts @@ -0,0 +1,123 @@ +import { describe, expect, it } from "vitest"; + +import { makeOrbEvent } from "./events"; +import { + extractNeedsInputQuestion, + isWorkComplete, + mapOrbEvent, + NEEDS_INPUT_MARKER, + WORK_COMPLETE_MARKER, +} from "./project-events"; + +describe("marker detection", () => { + it("extracts the needs-input question from a marker message", () => { + const text = `Some preamble\n${NEEDS_INPUT_MARKER} Which port should I use?`; + expect(extractNeedsInputQuestion(text)).toBe("Which port should I use?"); + }); + + it("returns undefined for a message without the marker", () => { + expect(extractNeedsInputQuestion("just working")).toBeUndefined(); + }); + + it("returns a default question when marker has no text after it", () => { + expect(extractNeedsInputQuestion(NEEDS_INPUT_MARKER)).toBe( + "Agent requires input" + ); + }); + + it("detects the work-complete marker", () => { + expect(isWorkComplete(`${WORK_COMPLETE_MARKER} all good`)).toBe(true); + expect(isWorkComplete("still working")).toBe(false); + }); +}); + +describe("mapOrbEvent", () => { + const issueId = "issue-1"; + const runId = "run-1"; + + it("maps session_opened to run.session_opened", () => { + const event = mapOrbEvent( + makeOrbEvent(1, "session_opened", { text: "Session abc opened" }), + issueId, + runId + ); + expect(event?.type).toBe("run.session_opened"); + expect(event?.text).toBe("Session abc opened"); + expect(event?.issueId).toBe(issueId); + expect(event?.runId).toBe(runId); + }); + + it("maps agent_message_completed with needs-input marker to run.needs_input", () => { + const event = mapOrbEvent( + makeOrbEvent(2, "agent_message_completed", { + text: `${NEEDS_INPUT_MARKER} What name?`, + }), + issueId, + runId + ); + expect(event?.type).toBe("run.needs_input"); + expect(event?.text).toBe("What name?"); + }); + + it("maps agent_message_completed without marker to run.agent_message", () => { + const event = mapOrbEvent( + makeOrbEvent(3, "agent_message_completed", { text: "I fixed the bug" }), + issueId, + runId + ); + expect(event?.type).toBe("run.agent_message"); + expect(event?.text).toBe("I fixed the bug"); + }); + + it("maps command_executed to run.command_executed with exit code", () => { + const event = mapOrbEvent( + makeOrbEvent(4, "command_executed", { command: "npm test", exitCode: 0 }), + issueId, + runId + ); + expect(event?.type).toBe("run.command_executed"); + expect(event?.exitCode).toBe(0); + expect(event?.text).toBe("npm test"); + }); + + it("maps tool_call events to run.agent_progress", () => { + const started = mapOrbEvent( + makeOrbEvent(5, "tool_call_started", { toolName: "edit_file" }), + issueId, + runId + ); + expect(started?.type).toBe("run.agent_progress"); + expect(started?.toolName).toBe("edit_file"); + }); + + it("maps session_failed to run.failed", () => { + const event = mapOrbEvent( + makeOrbEvent(6, "session_failed", { text: "crashed" }), + issueId, + runId + ); + expect(event?.type).toBe("run.failed"); + }); + + it("returns undefined for chunked events and vm_booted", () => { + expect( + mapOrbEvent( + makeOrbEvent(7, "agent_message_chunk", { text: "partial" }), + issueId, + runId + ) + ).toBeUndefined(); + expect( + mapOrbEvent(makeOrbEvent(8, "vm_booted"), issueId, runId) + ).toBeUndefined(); + }); + + it("maps permission_requested to run.permission_requested", () => { + const event = mapOrbEvent( + makeOrbEvent(9, "permission_requested", { text: "needs approval" }), + issueId, + runId + ); + expect(event?.type).toBe("run.permission_requested"); + }); +}); diff --git a/packages/agents/src/orb/project-events.ts b/packages/agents/src/orb/project-events.ts new file mode 100644 index 0000000..abac101 --- /dev/null +++ b/packages/agents/src/orb/project-events.ts @@ -0,0 +1,171 @@ +import { Schema } from "effect"; + +import type { OrbEvent } from "./events"; + +// --------------------------------------------------------------------------- +// Durable project events — human-meaningful projections of Orb execution +// +// These are NOT raw OpenCode events. Each variant maps to a product-level +// concept the UI and work-graph understand. Raw OrbEvents are preserved for +// audit; this is the durable projection layer. +// --------------------------------------------------------------------------- + +export const ProjectRunEventVariant = Schema.Literals([ + "run.started", + "run.repository_prepared", + "run.session_opened", + "run.agent_message", + "run.agent_progress", + "run.command_executed", + "run.needs_input", + "run.permission_requested", + "run.completed", + "run.failed", + "run.cancelled", + "run.session_closed", +]); +export type ProjectRunEventVariant = typeof ProjectRunEventVariant.Type; + +export const ProjectRunEvent = Schema.Struct({ + exitCode: Schema.UndefinedOr(Schema.Int), + issueId: Schema.String, + runId: Schema.String, + sequence: Schema.Number, + text: Schema.UndefinedOr(Schema.String), + timestamp: Schema.String, + toolName: Schema.UndefinedOr(Schema.String), + type: ProjectRunEventVariant, +}); +export type ProjectRunEvent = typeof ProjectRunEvent.Type; + +// --------------------------------------------------------------------------- +// Marker detection — OpenCode signals product-level conditions via markers +// --------------------------------------------------------------------------- + +/** Prefix the agent emits when it cannot proceed without human input. */ +export const NEEDS_INPUT_MARKER = "NEEDS_INPUT:"; + +/** + * Prefix the agent emits when implementation and verification are complete + * and the orchestrator should proceed to the Git publish lifecycle. + */ +export const WORK_COMPLETE_MARKER = "WORK_COMPLETE:"; + +export const extractNeedsInputQuestion = (text: string): string | undefined => { + const index = text.indexOf(NEEDS_INPUT_MARKER); + if (index === -1) { + return undefined; + } + const after = text.slice(index + NEEDS_INPUT_MARKER.length).trim(); + return after.length > 0 ? after.slice(0, 4000) : "Agent requires input"; +}; + +export const isWorkComplete = (text: string): boolean => + text.includes(WORK_COMPLETE_MARKER); + +// --------------------------------------------------------------------------- +// OrbEvent → ProjectRunEvent translation +// --------------------------------------------------------------------------- + +/** + * Translate one normalized OrbEvent into zero or one durable ProjectRunEvent. + * Returns undefined for OrbEvents that have no product-meaningful projection + * (e.g. chunked intermediate output that is only useful in the raw audit log). + */ +export const mapOrbEvent = ( + orbEvent: OrbEvent, + issueId: string, + runId: string +): ProjectRunEvent | undefined => { + const base: { + exitCode: number | undefined; + issueId: string; + runId: string; + sequence: number; + text: string | undefined; + timestamp: string; + toolName: string | undefined; + } = { + exitCode: undefined, + issueId, + runId, + sequence: orbEvent.sequence, + text: undefined, + timestamp: orbEvent.timestamp, + toolName: undefined, + }; + + switch (orbEvent.type) { + case "session_opened": { + return { + ...base, + text: orbEvent.text ?? "Session opened", + type: "run.session_opened", + }; + } + case "vm_booted": { + return undefined; + } + case "agent_message_completed": { + const text = orbEvent.text ?? ""; + if (extractNeedsInputQuestion(text) !== undefined) { + return { + ...base, + text: extractNeedsInputQuestion(text), + type: "run.needs_input", + }; + } + return { + ...base, + text, + type: "run.agent_message", + }; + } + case "agent_message_chunk": + case "agent_thought_chunk": { + return undefined; + } + case "tool_call_started": + case "tool_call_completed": { + return { + ...base, + text: undefined, + toolName: orbEvent.toolName, + type: "run.agent_progress", + }; + } + case "command_executed": { + return { + ...base, + exitCode: orbEvent.exitCode, + text: orbEvent.command, + type: "run.command_executed", + }; + } + case "permission_requested": + case "permission_denied": { + return { + ...base, + text: orbEvent.text ?? "Permission requested", + type: "run.permission_requested", + }; + } + case "session_failed": { + return { + ...base, + text: orbEvent.text ?? "Session failed", + type: "run.failed", + }; + } + case "session_closed": { + return { + ...base, + text: orbEvent.text ?? "Session closed", + type: "run.session_closed", + }; + } + default: { + return undefined; + } + } +};