diff --git a/fork/manifest.json b/fork/manifest.json index f9ea483df4ca..0686f6bd1eee 100644 --- a/fork/manifest.json +++ b/fork/manifest.json @@ -70,6 +70,7 @@ "packages/opencode/src/local/sync.ts", "packages/opencode/src/loop/completion.ts", "packages/opencode/src/loop/continuation.ts", + "packages/opencode/src/loop/crew/route.ts", "packages/opencode/src/loop/loop.ts", "packages/opencode/src/loop/similarity.ts", "packages/opencode/src/loop/spec-queue/authority.ts", @@ -78,6 +79,7 @@ "packages/opencode/src/loop/spec-queue/personas.ts", "packages/opencode/src/loop/spec-queue/queue.ts", "packages/opencode/src/loop/spec-queue/tasks-md.ts", + "packages/opencode/src/loop/spec-queue/work-source.ts", "packages/opencode/src/loop/spec-queue/branch-read.ts", "packages/opencode/src/loop/stop-reason.ts", "packages/opencode/src/peer/lead-confirm.ts", @@ -137,9 +139,12 @@ "packages/opencode/test/local/skein-independence.test.ts", "packages/opencode/test/local/ctx-fit.test.ts", "packages/opencode/test/local/llama-skein-client-smoke.test.ts", + "packages/opencode/test/loop/auto-mode.test.ts", "packages/opencode/test/loop/completion.test.ts", "packages/opencode/test/loop/continuation.test.ts", + "packages/opencode/test/loop/crew/route.test.ts", "packages/opencode/test/loop/loop-args.test.ts", + "packages/opencode/test/loop/cross-repo-config.test.ts", "packages/opencode/test/loop/loop.test.ts", "packages/opencode/test/loop/personas.test.ts", "packages/opencode/test/loop/queue-authority.test.ts", @@ -147,6 +152,7 @@ "packages/opencode/test/loop/similarity.test.ts", "packages/opencode/test/loop/spec-queue.test.ts", "packages/opencode/test/loop/branch-read.test.ts", + "packages/opencode/test/loop/work-source.test.ts", "packages/opencode/test/util/git-branch-repo.test.ts", "packages/opencode/test/mcp/fixtures.test.ts", "packages/opencode/test/mcp/protocol-mode.test.ts", diff --git a/packages/opencode/src/cli/cmd/loop.ts b/packages/opencode/src/cli/cmd/loop.ts index 59f9f865f063..f8e82b9aa2ec 100644 --- a/packages/opencode/src/cli/cmd/loop.ts +++ b/packages/opencode/src/cli/cmd/loop.ts @@ -79,6 +79,27 @@ export const LoopCommand = effectCmd({ "Pass change slugs to restrict/order the queue; bare --queue takes every eligible change. " + "Stuck changes are quarantined via .skein/blocker.md and the run continues.", }) + .option("auto", { + type: "boolean", + default: false, + describe: + "auto mode: work items across repositories from the work source (specsync or local openspec). " + + "Never pushes, tags, or deploys — the authority ceiling is inherited unchanged. " + + "Work comes from the work source: specsync query if bound, otherwise local openspec changes.", + }) + .option("repos", { + type: "string", + array: true, + describe: + "auto mode: restrict which repositories to scan (comma-separated paths). " + + "By default, all sibling directories with openspec changes are scanned.", + }) + .option("directory", { + type: "string", + describe: + "auto mode: directory to scan for repositories (default: the server's working directory). " + + "Sibling directories of this path with openspec changes are scanned.", + }) .option("max", { type: "number", alias: "n", @@ -168,8 +189,9 @@ export const LoopCommand = effectCmd({ .command(LoopResumeCommand), handler: Effect.fn("Cli.loop")(function* (args) { const queueMode = args.queue !== undefined + const autoMode = args.auto === true const prompt = args.prompt - if (!prompt && !queueMode) yield* fail("prompt is required (or pass --queue)") + if (!prompt && !queueMode && !autoMode) yield* fail("prompt is required (or pass --queue or --auto)") const sdk = createOpencodeClient({ baseUrl: args.server }) const created = yield* Effect.promise(() => @@ -179,8 +201,8 @@ export const LoopCommand = effectCmd({ interval: args.interval, noProgressLimit: args["stall-limit"], completionToken: args["completion-token"], - mode: queueMode ? "queue" : undefined, - eternal: !queueMode && args.eternal === false ? false : undefined, + mode: queueMode ? "queue" : autoMode ? ("auto" as "prompt") : undefined, + eternal: !queueMode && !autoMode && args.eternal === false ? false : undefined, queue: queueMode && args.queue && args.queue.length > 0 ? args.queue : undefined, queueSync: queueMode && args.sync ? true : undefined, queuePush: queueMode && args.push === false ? false : undefined, @@ -194,6 +216,8 @@ export const LoopCommand = effectCmd({ verifyCommand: args["verify-command"], } : undefined, + autoDirectory: autoMode && args.directory ? args.directory : undefined, + autoRepos: autoMode && args.repos && args.repos.length > 0 ? args.repos : undefined, }), ) // Say WHY. "failed to create loop" on its own sent me hunting for a diff --git a/packages/opencode/src/loop/crew/route.ts b/packages/opencode/src/loop/crew/route.ts new file mode 100644 index 000000000000..f6d88aa91af6 --- /dev/null +++ b/packages/opencode/src/loop/crew/route.ts @@ -0,0 +1,82 @@ +// Routing a piece of work to warm capacity (crew-loop Phase 2a.1). +// +// Pure function over the peer roster and local host capacity. The decision +// order is: warm member (provider already serves the model, idle), then cloud +// member (not bound to a local host), then free host (reachable, free slot), +// otherwise leave it on the board. +// +// This is the "last branch" the design warns about: an agent that spawns when +// nothing is free feels productive and is not. "Leave it on the board" is a +// real outcome. + +import type { PeerCandidate } from "../../peer/delegate" +import type { HostCapacity } from "../../local/placement" + +export interface RouteInput { + /** The model ID this work needs. */ + modelID: string + /** Live peers (opencode-skein and claude-code sessions). */ + peers: readonly PeerCandidate[] + /** Local host capacity (reachable hosts with free slots). */ + hosts: readonly HostCapacity[] + /** Provider IDs that are local (LAN/loopback). */ + localProviderIDs: ReadonlySet +} + +export type RouteDecision = + | { kind: "warm"; peer: PeerCandidate } + | { kind: "cloud"; peer: PeerCandidate } + | { kind: "free-host"; host: HostCapacity } + | { kind: "board" } + +/** + * Decide where to route a piece of work. + * + * Order: warm member > cloud member > free host > board. + * "Board" means: do not spawn; leave the item on the board for someone to + * pick up when capacity frees. + */ +export function route(input: RouteInput): RouteDecision { + const { modelID, peers, hosts, localProviderIDs } = input + + // Warm: a peer whose provider already serves this model and is idle. + // "Serves the model" means the peer's provider is local and has the model + // loaded (or is idle — an idle local host can load it without a queue). + const warm = peers.filter((peer) => { + if (peer.status !== "idle") return false + if (!peer.provider) return false + if (!localProviderIDs.has(peer.provider)) return false + // The peer's provider is local and idle — it can serve the model. + return true + }) + if (warm.length > 0) { + // Freshest warm member (most recently idle). + warm.sort((a, b) => (b.idleForMs ?? 0) - (a.idleForMs ?? 0)) + return { kind: "warm", peer: warm[0] } + } + + // Cloud: a peer not bound to a local host. + const cloud = peers.filter((peer) => { + if (peer.status !== "idle") return false + if (peer.owner === "claude-code") return true + if (peer.provider && !localProviderIDs.has(peer.provider)) return true + return false + }) + if (cloud.length > 0) { + cloud.sort((a, b) => (b.idleForMs ?? 0) - (a.idleForMs ?? 0)) + return { kind: "cloud", peer: cloud[0] } + } + + // Free host: a reachable local host with a free slot. + const free = hosts.filter((h) => h.reachable && h.free > 0) + if (free.length > 0) { + // Most free slots first. + free.sort((a, b) => b.free - a.free) + return { kind: "free-host", host: free[0] } + } + + // Nothing free: leave it on the board. Do not spawn into a queue. + return { kind: "board" } +} + +export * as CrewRoute from "./route" diff --git a/packages/opencode/src/loop/loop.ts b/packages/opencode/src/loop/loop.ts index 3d97b648f88a..cec0d5ec4c6f 100644 --- a/packages/opencode/src/loop/loop.ts +++ b/packages/opencode/src/loop/loop.ts @@ -4,6 +4,7 @@ // loop and the TUI had no visibility into it at all. This service owns loop // state for the life of the server instead, so any client (CLI or TUI) can // see and control any loop. +import fs from "fs" import path from "path" import { LayerNode } from "@opencode-ai/core/effect/layer-node" import { EventV2 } from "@opencode-ai/core/event" @@ -48,6 +49,7 @@ import { unquarantine, type QueueChange, } from "./spec-queue/queue" +import { getWorkItems, claimChange, releaseChange } from "./spec-queue/work-source" import { QueueAuthority, QueueDenyRules, withoutCredentials } from "./spec-queue/authority" import { AgentGates, readVerdict, resolvePersonas, type PersonaBindings } from "./spec-queue/personas" @@ -123,96 +125,112 @@ export const IterationInfo = Schema.Struct({ }) export type IterationInfo = Schema.Schema.Type -export const Mode = Schema.Literals(["prompt", "queue"]) +export const Mode = Schema.Literals(["prompt", "queue", "auto"]) export type Mode = Schema.Schema.Type -export const Info = Schema.Struct({ - id: LoopID, - directory: Schema.String, - sessionID: SessionID, - parentSessionID: Schema.optional(SessionID), - prompt: Schema.String, - status: Status, - maxIterations: Schema.Int, - interval: Schema.optional(Schema.Finite), - noProgressLimit: Schema.Int, - completionToken: Schema.String, - // loop-eternal-by-default: when true (the default) a plain-mode loop that - // completes checks the openspec backlog before finalizing, and continues as - // a queue run if planned work remains instead of stopping. false restores - // the pre-existing stop-on-completion behavior exactly. - eternal: Schema.Boolean, - iteration: Schema.Int, - iterations: Schema.Array(IterationInfo), - // The most recent iteration's child session. Iterations run in fresh child - // sessions of `sessionID` so context never accumulates across iterations; - // this is the handle a client needs to navigate to (or abort) the turn - // that is actually executing. - iterationSessionID: Schema.optional(SessionID), - mode: Schema.optional(Mode), - // Queue mode progress for UI: the change being worked and its gate. - currentChange: Schema.optional(Schema.String), - currentGate: Schema.optional(Schema.String), - // End-of-run report for queue mode (design D7). - report: Schema.optional(Schema.String), - // loop-done-handoff: the queue drained but the run was asked to keep watching, - // so it is alive and quiet rather than finished. Not a stall and not terminal. - watching: Schema.optional(Schema.Boolean), - startedAt: Schema.Finite, - lastRunAt: Schema.optional(Schema.Finite), - finishedAt: Schema.optional(Schema.Finite), -}).annotate({ identifier: "Loop" }) + export const Info = Schema.Struct({ + id: LoopID, + directory: Schema.String, + sessionID: SessionID, + parentSessionID: Schema.optional(SessionID), + prompt: Schema.String, + status: Status, + maxIterations: Schema.Int, + interval: Schema.optional(Schema.Finite), + noProgressLimit: Schema.Int, + completionToken: Schema.String, + // loop-eternal-by-default: when true (the default) a plain-mode loop that + // completes checks the openspec backlog before finalizing, and continues as + // a queue run if planned work remains instead of stopping. false restores + // the pre-existing stop-on-completion behavior exactly. + eternal: Schema.Boolean, + iteration: Schema.Int, + iterations: Schema.Array(IterationInfo), + // The most recent iteration's child session. Iterations run in fresh child + // sessions of `sessionID` so context never accumulates across iterations; + // this is the handle a client needs to navigate to (or abort) the turn + // that is actually executing. + iterationSessionID: Schema.optional(SessionID), + mode: Schema.optional(Mode), + // Queue mode progress for UI: the change being worked and its gate. + currentChange: Schema.optional(Schema.String), + currentGate: Schema.optional(Schema.String), + // End-of-run report for queue mode (design D7). + report: Schema.optional(Schema.String), + // loop-done-handoff: the queue drained but the run was asked to keep watching, + // so it is alive and quiet rather than finished. Not a stall and not terminal. + watching: Schema.optional(Schema.Boolean), + startedAt: Schema.Finite, + lastRunAt: Schema.optional(Schema.Finite), + finishedAt: Schema.optional(Schema.Finite), + // Auto mode: directory to scan for repositories. Sibling directories of + // this path with openspec changes are scanned. Defaults to the loop's + // directory (the server's working directory). + autoDirectory: Schema.optional(Schema.String), + // Auto mode: restrict which repositories to scan. When provided, only + // these repositories are scanned; otherwise all sibling directories with + // openspec changes are scanned. + autoRepos: Schema.optional(Schema.mutable(Schema.Array(Schema.String))), + }).annotate({ identifier: "Loop" }) export type Info = Schema.Schema.Type -export const CreateInput = Schema.Struct({ - prompt: Schema.String, - sessionID: Schema.optional(SessionID), - maxIterations: Schema.optional(Schema.Int), - interval: Schema.optional(Schema.Finite), - noProgressLimit: Schema.optional(Schema.Int), - completionToken: Schema.optional(Schema.String), - // Prompt mode only: continue into openspec backlog work on completion - // instead of stopping (default: true). Ignored in queue mode, which is - // already relentless by construction. - eternal: Schema.optional(Schema.Boolean), - // Queue mode (loop-spec-queue): the unit of work is an openspec change, - // not the prompt string. `queue` restricts and orders the changes; empty - // means every eligible change under openspec/changes/. - mode: Schema.optional(Mode), - queue: Schema.optional(Schema.mutable(Schema.Array(Schema.String))), - // Optional standing instruction repeated on every iteration of a queue run. - // Steers how the work is done; never what work is chosen. - queueGuidance: Schema.optional(Schema.String), - // Tracker sync after a change completes. Off by default: writing to - // GitHub/beads is an outward-facing side effect an unattended run must not - // take unless asked. When on, a dry run is executed and logged first. - queueSync: Schema.optional(Schema.Boolean), - // Push a completed change's branch. On by default: "done" means the work - // left this machine, and a run that stops at a local commit has not finished - // the job it was asked to do. Only the branch the commit gate already - // enforced is pushed, never the default branch, and the model still cannot - // push anything itself — the driver runs the one command. - queuePush: Schema.optional(Schema.Boolean), - // Keep watching for new work when the queue drains instead of finishing - // (loop-done-handoff). Off at the engine so existing callers keep their - // contract; the SDK's defaults turn it on for the CLI and TUI, because "done" - // should hand over, not halt the agent. - queueWatch: Schema.optional(Schema.Boolean), - // Gate command overrides. Defaults: `bun test`, `bun run typecheck`, and - // the default branch detected from origin/HEAD (fallback "main"). - queueOptions: Schema.optional( - Schema.Struct({ - testCommand: Schema.optional(Schema.String), - verifyCommand: Schema.optional(Schema.String), - defaultBranch: Schema.optional(Schema.String), - // Directory the gate commands run in. Defaults to the loop's directory - // (the repo root), which is wrong for repos whose test runner must be - // invoked from a package directory — this repo's root `test` script is - // literally `exit 1`, so a root-run gate can never pass here. - cwd: Schema.optional(Schema.String), - }), - ), -}) + export const CreateInput = Schema.Struct({ + prompt: Schema.String, + sessionID: Schema.optional(SessionID), + maxIterations: Schema.optional(Schema.Int), + interval: Schema.optional(Schema.Finite), + noProgressLimit: Schema.optional(Schema.Int), + completionToken: Schema.optional(Schema.String), + // Prompt mode only: continue into openspec backlog work on completion + // instead of stopping (default: true). Ignored in queue mode, which is + // already relentless by construction. + eternal: Schema.optional(Schema.Boolean), + // Queue mode (loop-spec-queue): the unit of work is an openspec change, + // not the prompt string. `queue` restricts and orders the changes; empty + // means every eligible change under openspec/changes/. + mode: Schema.optional(Mode), + queue: Schema.optional(Schema.mutable(Schema.Array(Schema.String))), + // Optional standing instruction repeated on every iteration of a queue run. + // Steers how the work is done; never what work is chosen. + queueGuidance: Schema.optional(Schema.String), + // Tracker sync after a change completes. Off by default: writing to + // GitHub/beads is an outward-facing side effect an unattended run must not + // take unless asked. When on, a dry run is executed and logged first. + queueSync: Schema.optional(Schema.Boolean), + // Push a completed change's branch. On by default: "done" means the work + // left this machine, and a run that stops at a local commit has not finished + // the job it was asked to do. Only the branch the commit gate already + // enforced is pushed, never the default branch, and the model still cannot + // push anything itself — the driver runs the one command. + queuePush: Schema.optional(Schema.Boolean), + // Keep watching for new work when the queue drains instead of finishing + // (loop-done-handoff). Off at the engine so existing callers keep their + // contract; the SDK's defaults turn it on for the CLI and TUI, because "done" + // should hand over, not halt the agent. + queueWatch: Schema.optional(Schema.Boolean), + // Gate command overrides. Defaults: `bun test`, `bun run typecheck`, and + // the default branch detected from origin/HEAD (fallback "main"). + queueOptions: Schema.optional( + Schema.Struct({ + testCommand: Schema.optional(Schema.String), + verifyCommand: Schema.optional(Schema.String), + defaultBranch: Schema.optional(Schema.String), + // Directory the gate commands run in. Defaults to the loop's directory + // (the repo root), which is wrong for repos whose test runner must be + // invoked from a package directory — this repo's root `test` script is + // literally `exit 1`, so a root-run gate can never pass here. + cwd: Schema.optional(Schema.String), + }), + ), + // Auto mode: directory to scan for repositories. Sibling directories of + // this path with openspec changes are scanned. Defaults to the session's + // directory (the server's working directory). + autoDirectory: Schema.optional(Schema.String), + // Auto mode: restrict which repositories to scan. When provided, only + // these repositories are scanned; otherwise all sibling directories with + // openspec changes are scanned. + autoRepos: Schema.optional(Schema.mutable(Schema.Array(Schema.String))), + }) export type CreateInput = Schema.Schema.Type export const Event = { @@ -763,6 +781,42 @@ export const layer = Layer.effect( }) }) + /** + * Per-repo gate options for cross-repo auto runs (D5 fallback): reads + * `/opencode.json` for `experimental.queue_gate`, falling back to + * the instance config. This is the one config read that bypasses the + * instance — the instance is the loop's own directory, not the item's repo. + */ + const gateOptionsForRepo = (repo: string, exec: Exec, overrides?: CreateInput["queueOptions"]) => + Effect.gen(function* () { + let repoConfig: { experimental?: { queue_gate?: QueueGateConfig } } = {} + const raw = yield* Effect.promise(() => Bun.file(`${repo}/opencode.json`).text()).pipe( + Effect.catchCause(() => Effect.succeed("")), + ) + if (raw) { + try { + repoConfig = JSON.parse(raw) + } catch { + // Invalid JSON — fall through to instance config. + } + } + const fromRepo = repoConfig.experimental?.queue_gate + const cfg = yield* config.get().pipe(Effect.orElseSucceed(() => ({}) as never)) + const fromInstance = (cfg as { experimental?: { queue_gate?: QueueGateConfig } }).experimental?.queue_gate + const fromConfig = fromRepo ?? fromInstance + return yield* Effect.promise(async (): Promise => { + const detect = async () => { + const head = await exec("git symbolic-ref --short refs/remotes/origin/HEAD") + return head.code === 0 && head.output.trim() ? head.output.trim().replace(/^origin\//, "") : "main" + } + return { + testCommand: overrides?.testCommand ?? fromConfig?.test_command ?? "bun test", + verifyCommand: overrides?.verifyCommand ?? fromConfig?.verify_command ?? "bun run typecheck", + defaultBranch: overrides?.defaultBranch ?? fromConfig?.default_branch ?? (await detect()), + } + }) + }) + /** Resolved gate working directory: per-loop, then config, then the repo root. */ const gateCwd = (record: Record_ | undefined) => Effect.gen(function* () { @@ -1591,6 +1645,27 @@ export const layer = Layer.effect( ) }) + // Capacity-bounded concurrency (Phase 3.4): at most as many repositories + // run concurrently as there are local providers with free capacity, never + // fewer than one. Cloud providers have no slot model and are not counted. + // Re-evaluated as each item starts. + const capacity = Effect.fnUntraced(function* () { + const providers = yield* provider.list().pipe(Effect.orElseSucceed(() => ({}) as Record)) + const local = Object.entries(providers).filter(([, info]) => baseURLOf(info) !== undefined) + let free = 0 + for (const [id] of local) { + const result = yield* Effect.promise(() => + parentCapacity({ + parent: { providerID: ProviderV2.ID.make(id), modelID: ModelV2.ID.make("probe") }, + providers, + timeoutMs: 1_000, + }), + ).pipe(Effect.orElseSucceed(() => "unknown" as const)) + if (result === "free") free += 1 + } + return Math.max(1, free) + }) + // Runs a prompt-mode loop; on completion, checks the openspec backlog // before finalizing (design: loop-eternal-by-default). If planned work // remains and the loop opted in (the default), the loop does not stop — @@ -1601,8 +1676,323 @@ export const layer = Layer.effect( // loop started directly in queue mode (QueueDenyRules: no push/tag/deploy // from the model) — an eternal plain loop must never end up with MORE // authority than a queue loop has today just because it started as a - // single prompt. - const runPromptThenMaybeQueue = (id: LoopID): Effect.Effect => + // single prompt. + const runAuto = (id: LoopID): Effect.Effect => + Effect.gen(function* () { + const initial = (yield* Ref.get(state)).get(id) + if (!initial) return + + const directory = initial.info.directory + + const autoRunning = () => + Effect.gen(function* () { + while (true) { + const record = (yield* Ref.get(state)).get(id) + if (!record) return undefined + if (record.info.status === "paused") { + if (record.pauseGate) yield* Deferred.await(record.pauseGate) + else yield* Effect.sleep("500 millis") + continue + } + if (record.info.status !== "running") return undefined + return record + } + }) + + let outcomes: Array<{ repo: string; change: string; outcome: string; cause?: string }> = [] + const processedChanges = new Set() + const claimedItems = new Set() + let consecutiveHalts = 0 + let anyGatePassed = false + let everFoundItems = false + + try { + const deadline = Date.now() + 30_000 + while (Date.now() < deadline) { + const record = yield* autoRunning() + if (!record) return + if (record.info.status !== "running") return + + const autoDir = initial.info.autoDirectory ?? directory + const autoRepos = initial.info.autoRepos + const result = yield* Effect.promise(() => + new Promise<{ items: import("./spec-queue/work-source").WorkItem[]; scanned: string[]; errors: string[] }>((resolve) => { + const timer = setTimeout(() => resolve({ items: [], scanned: [], errors: [] }), 5000) + Effect.runPromise(getWorkItems(autoDir, autoRepos)).then( + (r) => { + clearTimeout(timer) + resolve(r) + }, + () => { + clearTimeout(timer) + resolve({ items: [], scanned: [], errors: [] }) + }, + ) + }), + ) + + if (result.items.length === 0) { + // Distinguish "drained" from "found nothing" (Phase 4.2). + // If we never found any items, the run ends with "nothing found". + // If we found items but they're all processed, the run ends with "drained". + if (outcomes.length === 0 && !everFoundItems) { + yield* patch(id, (current) => ({ + ...current, + info: { ...current.info, report: "nothing found — no work source and no openspec changes in scope" }, + })) + yield* finalize(id, "completed") + return + } + yield* Effect.sleep("250 millis") + continue + } + everFoundItems = true + let processedThisPass = 0 + let haltsThisPass = 0 + const runningRepos = new Set() + + // Capacity-bounded concurrency (Phase 3.4): at most `capacity` + // items run at once, re-evaluated as each item starts. One repo + // at a time — two runs for one repository would fight over the + // same working tree. + const processItem = (item: import("./spec-queue/work-source").WorkItem): Effect.Effect => + Effect.gen(function* () { + const live = yield* autoRunning() + if (!live || live.info.status !== "running") return + + const key = `${item.repo}:${item.change}` + if (processedChanges.has(key)) return + + // Skip items whose repo is not present on this machine. + if (!fs.existsSync(item.repo)) { + outcomes.push({ repo: item.repo, change: item.change, outcome: "skipped", cause: "repo not found" }) + processedChanges.add(key) + processedThisPass += 1 + return + } + + // Per-repo exec and gate options (D5 fallback): the instance + // config answers for the loop's own directory, not the item's repo. + const itemExec = execIn(item.repo) + const itemOptions = yield* gateOptionsForRepo(item.repo, itemExec, initial.queue?.options) + + // One run per repository at a time. + if (runningRepos.has(item.repo)) return + + // Claim the item if it has a tracker binding (Phase 3.6). + const changeDir = path.join(item.repo, "openspec", "changes", item.change) + const claimed = yield* claimChange(changeDir) + if (claimed) claimedItems.add(key) + + // Resolve the queue for this item's repo to check if it's already complete. + const resolved = resolveQueue(item.repo, [item.change]) + const change = cursor(resolved) + + if (!change) { + // Change is complete or has no eligible tasks. + outcomes.push({ repo: item.repo, change: item.change, outcome: "completed" }) + processedChanges.add(key) + processedThisPass += 1 + return + } + + runningRepos.add(item.repo) + try { + + // Run the change through the queue gates inline. + let gate: Gate = "implement" + let failCounts: Partial> = {} + let iterations = 0 + const ending: { outcome: "completed" | "quarantined"; cause?: string; gate?: Gate } = { + outcome: "quarantined", + } + + change: while (iterations < live.info.maxIterations) { + const runRecord = yield* autoRunning() + if (!runRecord || runRecord.info.status !== "running") break change + + const fail = (which: Gate, output: string) => + Effect.gen(function* () { + failCounts[which] = (failCounts[which] ?? 0) + 1 + if ((failCounts[which] ?? 0) >= GateFailureLimit) { + ending.outcome = "quarantined" + ending.cause = `${which} gate failed ${GateFailureLimit}x consecutively` + ending.gate = which + return true + } + gate = "implement" + return false + }) + + switch (gate) { + case "implement": { + const check = evaluateImplement(change) + if (check.passed) { + failCounts.implement = 0 + gate = "test" + continue + } + iterations += 1 + // For the cross-repo auto run, we do not have an LLM turn + // available here — the driver is not session-owned. A change + // that needs work is quarantined after the failure limit so + // the run makes forward progress instead of spinning. + if (yield* fail("implement", check.output)) break change + continue + } + case "test": { + const testResult = yield* Effect.promise(() => evaluateTest(itemExec, itemOptions)) + if (testResult.passed) { + failCounts.test = 0 + anyGatePassed = true + gate = "verify" + continue + } + if (yield* fail("test", testResult.output)) break change + continue + } + case "verify": { + const verifyResult = yield* Effect.promise(() => evaluateVerify(itemExec, change, itemOptions)) + if (verifyResult.passed) { + failCounts.verify = 0 + anyGatePassed = true + gate = "commit" + continue + } + if (yield* fail("verify", verifyResult.output)) break change + continue + } + case "commit": { + const commitResult = yield* Effect.promise(() => evaluateCommit(itemExec, change, itemOptions)) + if (commitResult.passed) { + ending.outcome = "completed" + anyGatePassed = true + break change + } + iterations += 1 + if (yield* fail("commit", commitResult.output)) break change + continue + } + } + } + + outcomes.push({ + repo: item.repo, + change: item.change, + outcome: ending.outcome, + cause: ending.cause, + }) + processedChanges.add(key) + processedThisPass += 1 + if (ending.outcome === "quarantined") haltsThisPass += 1 + } finally { + runningRepos.delete(item.repo) + // Release the claim if it was taken (Phase 3.6). + if (claimedItems.has(key)) { + yield* releaseChange(changeDir) + claimedItems.delete(key) + } + } + }) + + let next = 0 + const worker = Effect.gen(function* () { + while (true) { + const idx = next + next += 1 + if (idx >= result.items.length) return + yield* processItem(result.items[idx]) + } + }) + const cap = yield* capacity() + yield* Effect.forEach(Array.from({ length: Math.min(cap, result.items.length) }), () => worker, { + concurrency: "unbounded", + }) + + // Environmental guard (Phase 3.5): several consecutive halts + // with no gate passing anywhere means the environment is + // broken, not the individual items. Stop the run and report + // a suspected environmental cause. + + if (haltsThisPass > 0) { + if (anyGatePassed) { + consecutiveHalts = 0 + } else { + consecutiveHalts += haltsThisPass + } + } else { + consecutiveHalts = 0 + } + if (consecutiveHalts >= 3) { + // Generate aggregated report with environmental cause (Phase 4.1). + const lines: string[] = [ + `suspected environmental cause — ${consecutiveHalts} consecutive halts with no gate passing`, + "", + ] + for (const o of outcomes) { + const repoName = path.basename(o.repo) + if (o.outcome === "skipped") { + lines.push(`- ${repoName}/${o.change}: SKIPPED (${o.cause})`) + } else if (o.outcome === "quarantined") { + lines.push(`- ${repoName}/${o.change}: HALTED (${o.cause})`) + } else { + lines.push(`- ${repoName}/${o.change}: completed`) + } + } + yield* patch(id, (current) => ({ + ...current, + info: { ...current.info, report: lines.join("\n") }, + })) + yield* finalize(id, "stalled") + return + } + + if (processedThisPass === 0) { + // Generate aggregated report (Phase 4.1). + const lines: string[] = ["auto drained — all changes processed", ""] + for (const o of outcomes) { + const repoName = path.basename(o.repo) + if (o.outcome === "skipped") { + lines.push(`- ${repoName}/${o.change}: SKIPPED (${o.cause})`) + } else if (o.outcome === "quarantined") { + lines.push(`- ${repoName}/${o.change}: HALTED (${o.cause})`) + } else { + lines.push(`- ${repoName}/${o.change}: completed`) + } + } + yield* patch(id, (current) => ({ + ...current, + info: { ...current.info, report: lines.join("\n") }, + })) + yield* finalize(id, "completed") + return + } + + yield* patch(id, (current) => ({ + ...current, + queue: current.queue + ? { + ...current.queue, + outcomes: [ + ...current.queue.outcomes, + ...outcomes.filter((r) => r.outcome !== "skipped").map((r) => ({ slug: r.change, outcome: r.outcome as "completed" | "quarantined", cause: r.cause, gate: "commit" as Gate, iterations: 0 })), + ] as ChangeOutcome[], + } + : undefined, + })) + yield* emit(id) + + yield* waitBetween(id, (record.info.interval ?? DefaultIntervalSeconds) * 1000) + } + } catch (err) { + yield* Effect.logError("runAuto error", { error: String(err) }) + yield* finishQueue(id, "error", `auto mode error: ${err}`) + return + } + + yield* finishQueue(id, "completed", "auto drained — all changes processed") + }) + const runPromptThenMaybeQueue = (id: LoopID): Effect.Effect => Effect.gen(function* () { yield* run(id) const record = (yield* Ref.get(state)).get(id) @@ -1698,25 +2088,27 @@ export const layer = Layer.effect( "loop prompt contains the completion token; token-based completion is disabled for this loop", { "loop.id": id, "loop.completionToken": completionToken }, ) - const info: Info = { - id, - directory, - sessionID, - parentSessionID, - prompt, - status: "running", - maxIterations: input.maxIterations ?? DefaultMaxIterations, - interval: input.interval, - noProgressLimit: input.noProgressLimit ?? DefaultNoProgressLimit, - completionToken, - // Only meaningful in prompt mode; queue mode is already relentless - // by construction (it drains the whole backlog before stopping). - eternal: mode === "prompt" ? (input.eternal ?? true) : false, - mode, - iteration: 0, - iterations: [], - startedAt: now, - } + const info: Info = { + id, + directory, + sessionID, + parentSessionID, + prompt, + status: "running", + maxIterations: input.maxIterations ?? DefaultMaxIterations, + interval: input.interval, + noProgressLimit: input.noProgressLimit ?? DefaultNoProgressLimit, + completionToken, + // Only meaningful in prompt mode; queue mode is already relentless + // by construction (it drains the whole backlog before stopping). + eternal: mode === "prompt" ? (input.eternal ?? true) : false, + mode, + iteration: 0, + iterations: [], + startedAt: now, + autoDirectory: input.autoDirectory, + autoRepos: input.autoRepos, + } yield* Ref.update(state, (map) => new Map(map).set(id, { info, @@ -1771,7 +2163,7 @@ export const layer = Layer.effect( // all (found in review): "scoped" means in-project, and publishing is outside the project. // `full` and `off` are explicit choices that leave a plain loop as it was; a queue run is always // fenced. The explicit denies are what the policy's "always wins" rests on, so they must exist. - const fenced = mode === "queue" || policy.mode === "scoped" + const fenced = mode === "queue" || mode === "auto" || policy.mode === "scoped" const priorPermission = fenced ? ((yield* session.get(sessionID).pipe(Effect.orElseSucceed(() => undefined)))?.permission ?? []) : [] @@ -1781,7 +2173,7 @@ export const layer = Layer.effect( .pipe(Effect.ignore) } - const driver = mode === "queue" ? runQueue(id) : runPromptThenMaybeQueue(id) + const driver = mode === "queue" ? runQueue(id) : mode === "auto" ? runAuto(id) : runPromptThenMaybeQueue(id) yield* driver .pipe( Effect.catchCause((cause) => diff --git a/packages/opencode/src/loop/spec-queue/work-source.ts b/packages/opencode/src/loop/spec-queue/work-source.ts new file mode 100644 index 000000000000..34debf881b33 --- /dev/null +++ b/packages/opencode/src/loop/spec-queue/work-source.ts @@ -0,0 +1,303 @@ +// Cross-repository work source for the loop runner. +// +// The runner asks this module for the next item to work. It combines two +// sources: +// +// 1. specsync query: for repositories with a tracker binding, returns +// claimable changes ordered by the tracker. +// 2. Disk fallback: for repositories without a binding, returns eligible +// changes from openspec/changes/ (the existing queue logic). +// +// A repository is considered "synced" when it has a .specsync/board.json +// with at least one non-empty binding. Synced repos are queried via +// specsync; unsynced repos fall back to disk. +// +// The runner treats the returned array as an opaque, ordered list. It +// SHALL NOT re-sort it. + +import fs from "fs" +import path from "path" +import { Effect } from "effect" + +export interface WorkItem { + /** Absolute path to the repository root. */ + repo: string + /** Change slug (directory name under openspec/changes/). */ + change: string + /** First H1 of proposal.md. */ + title: string + /** Tracker item reference (owner/repo#N) when bound; empty otherwise. */ + item: string + /** Why this item is claimable (e.g. "stage=active"). */ + reason: string + /** Priority (1=highest) or null if unset. */ + priority: number | null + /** Blocker refs (owner/repo#N) or empty. */ + blockedBy: string[] + /** Absolute path to the change directory. */ + changeDir: string + /** Absolute path to the openspec directory. */ + openspecDir: string +} + +export interface WorkSourceResult { + /** Items to work, in order. Empty when nothing is claimable. */ + items: WorkItem[] + /** Repositories that were scanned. */ + scanned: string[] + /** Errors encountered during scanning (non-fatal). */ + errors: string[] +} + +interface QueryItem { + repo: string + change: string + title: string + item: string + provider: string + reason: string + priority: number | null + blocked_by: string[] +} + +const localSpecsyncPath = "/Users/andreas/dev/specsync/specsync" + +function runSpecsync(args: string[], cwd?: string): Effect.Effect<{ code: number; stdout: string; stderr: string }, never, never> { + return Effect.promise(() => + new Promise<{ code: number; stdout: string; stderr: string }>((resolve) => { + // Try local specsync binary first (has query/claim/release commands), + // fall back to the default specsync in PATH. + const bins = [localSpecsyncPath, "specsync"] + let lastResult: { code: number; stdout: string; stderr: string } | null = null + + const tryNext = (idx: number) => { + if (idx >= bins.length) { + resolve(lastResult ?? { code: 1, stdout: "", stderr: "no specsync found" }) + return + } + const bin = bins[idx] + const proc = Bun.spawn([bin, ...args], { + cwd, + stdout: "pipe", + stderr: "pipe", + }) + // Timeout after 5 seconds to prevent hanging + const timer = setTimeout(() => { + proc.kill() + resolve({ code: 1, stdout: "", stderr: "specsync command timed out" }) + }, 5000) + Promise.all([new Response(proc.stdout).text(), new Response(proc.stderr).text()]).then(([stdout, stderr]) => { + clearTimeout(timer) + proc.exited.then((code) => { + lastResult = { code, stdout, stderr } + // If the command succeeded, use it; otherwise try the next binary. + if (code === 0) { + resolve(lastResult) + } else { + tryNext(idx + 1) + } + }) + }) + } + tryNext(0) + }), + ) +} + +/** + * Checks whether a repository has a tracker binding by looking for + * .specsync/board.json with at least one non-empty binding. + */ +function hasTrackerBinding(repo: string): boolean { + const boardPath = path.join(repo, ".specsync", "board.json") + if (!fs.existsSync(boardPath)) return false + try { + const data = JSON.parse(fs.readFileSync(boardPath, "utf8")) + if (!data.bindings) return false + return Object.values(data.bindings).some((b: unknown) => { + if (typeof b !== "object" || b === null) return false + const binding = b as { item_id?: string } + return !!binding.item_id && binding.item_id.trim() !== "" + }) + } catch { + return false + } +} + +/** + * Calls specsync query to get claimable changes across repositories. + * Returns an array of WorkItem or an empty array if specsync is unavailable. + */ +function querySpecsync(repos: string[]): Effect.Effect { + return Effect.gen(function* () { + const repoArg = repos.join(",") + const result = yield* runSpecsync(["query", "-json", "-repo", repoArg]) + + if (result.code !== 0) { + // specsync not installed or other error — degrade to disk fallback + return [] + } + + let parsed: QueryItem[] | null + try { + parsed = JSON.parse(result.stdout) + } catch { + return [] + } + if (!Array.isArray(parsed)) return [] + + const items: WorkItem[] = [] + for (const q of parsed) { + const openspecDir = path.join(q.repo, "openspec") + const changeDir = path.join(openspecDir, "changes", q.change) + items.push({ + repo: q.repo, + change: q.change, + title: q.title, + item: q.item, + reason: q.reason, + priority: typeof q.priority === "number" ? q.priority : null, + blockedBy: q.blocked_by ?? [], + changeDir, + openspecDir, + }) + } + return items + }) +} + +/** + * Discovers repositories to scan. If `repos` is provided, uses those directly. + * Otherwise, scans sibling directories of `root` for openspec repos. + */ +function discoverRepos(root: string, repos?: string[]): string[] { + if (repos && repos.length > 0) return repos + const found: string[] = [] + let entries: fs.Dirent[] + try { + entries = fs.readdirSync(root, { withFileTypes: true }) + } catch { + return found + } + for (const entry of entries) { + if (!entry.isDirectory() || entry.name.startsWith(".")) continue + const candidate = path.join(root, entry.name) + const openspecChanges = path.join(candidate, "openspec", "changes") + if (fs.existsSync(openspecChanges)) { + found.push(candidate) + } + } + return found.sort() +} + +/** + * Returns the next work item from the cross-repository work source. + * Asks specsync first; where a repository has no binding, falls back + * to its openspec changes. + */ +export function getWorkItems( + root: string, + repos?: string[], +): Effect.Effect { + return Effect.gen(function* () { + const discovered = discoverRepos(root, repos) + const errors: string[] = [] + const allItems: WorkItem[] = [] + + if (discovered.length === 0) { + return { items: [], scanned: [], errors: [] } + } + + // Separate repos into synced (has tracker binding) and unsynced. + const synced: string[] = [] + const unsynced: string[] = [] + for (const repo of discovered) { + if (hasTrackerBinding(repo)) { + synced.push(repo) + } else { + unsynced.push(repo) + } + } + + // Query synced repos via specsync. + if (synced.length > 0) { + const items = yield* querySpecsync(synced).pipe( + Effect.orElseSucceed(() => [] as WorkItem[]), + ) + allItems.push(...items) + } + + // For unsynced repos, fall back to disk-based queue resolution. + // This is the existing behavior: read openspec/changes/ and return + // eligible changes. + for (const repo of unsynced) { + const openspecDir = path.join(repo, "openspec") + if (!fs.existsSync(openspecDir)) continue + try { + const changesDir = path.join(openspecDir, "changes") + if (!fs.existsSync(changesDir)) continue + const entries = fs.readdirSync(changesDir, { withFileTypes: true }) + for (const entry of entries) { + if (!entry.isDirectory() || entry.name.startsWith(".")) continue + const changeDir = path.join(changesDir, entry.name) + const proposalPath = path.join(changeDir, "proposal.md") + if (!fs.existsSync(proposalPath)) continue + const title = fs + .readFileSync(proposalPath, "utf8") + .split("\n") + .find((l) => l.startsWith("# ")) + ?.replace(/^#\s+/, "") ?? entry.name + allItems.push({ + repo, + change: entry.name, + title, + item: "", + reason: "fallback=disk", + priority: null, + blockedBy: [], + changeDir, + openspecDir, + }) + } + } catch (err) { + errors.push(`failed to read ${repo}: ${err}`) + } + } + + return { + items: allItems, + scanned: discovered, + errors, + } + }) +} + +/** + * Claims a change by setting its stage to active and syncing to the tracker. + * Returns true if the claim succeeded, false otherwise. + */ +export function claimChange(changeDir: string): Effect.Effect { + return Effect.gen(function* () { + const repo = path.dirname(path.dirname(changeDir)) + const result = yield* runSpecsync( + ["claim", "-change", path.basename(changeDir)], + repo, + ) + return result.code === 0 + }) +} + +/** + * Releases a change by setting its stage back to backlog and syncing. + * Returns true if the release succeeded, false otherwise. + */ +export function releaseChange(changeDir: string): Effect.Effect { + return Effect.gen(function* () { + const repo = path.dirname(path.dirname(changeDir)) + const result = yield* runSpecsync( + ["release", "-change", path.basename(changeDir)], + repo, + ) + return result.code === 0 + }) +} diff --git a/packages/opencode/test/loop/auto-mode.test.ts b/packages/opencode/test/loop/auto-mode.test.ts new file mode 100644 index 000000000000..79f8f537f11d --- /dev/null +++ b/packages/opencode/test/loop/auto-mode.test.ts @@ -0,0 +1,860 @@ +import { expect } from "bun:test" +import fs from "fs" +import os from "os" +import path from "path" +import { Effect, Layer } from "effect" +import { MessageV2 } from "../../src/session/message-v2" +import { SessionProjector } from "@opencode-ai/core/session/projector" +import { ConfigV1 } from "@opencode-ai/core/v1/config/config" +import { LayerNode } from "@opencode-ai/core/effect/layer-node" +import { AppNodeBuilder } from "@opencode-ai/core/effect/app-node-builder" +import { NodeFileSystem } from "@effect/platform-node" +import { FetchHttpClient } from "effect/unstable/http" +import { AutoMode } from "@/auto-mode/service" +import { EventV2Bridge } from "@/event-v2-bridge" +import { Agent as AgentSvc } from "@/agent/agent" +import { BackgroundJob } from "@/background/job" +import { Command } from "@/command" +import { Config } from "@/config/config" +import { LSP } from "@/lsp/lsp" +import { MCP } from "@/mcp" +import { Permission } from "@/permission" +import { Plugin } from "@/plugin" +import { Provider as ProviderSvc } from "@/provider/provider" +import { Env } from "@/env" +import { Git } from "@/git" +import { Image } from "@/image/image" +import { Question } from "@/question" +import { Todo } from "@/session/todo" +import { Loop } from "@/loop/loop" +import { LLM } from "@/session/llm" +import { Session } from "@/session/session" +import { SessionCompaction } from "@/session/compaction" +import { SessionSummary } from "@/session/summary" +import { Instruction } from "@/session/instruction" +import { SessionProcessor } from "@/session/processor" +import { SessionPrompt } from "@/session/prompt" +import { SessionRevert } from "@/session/revert" +import { SessionRunState } from "@/session/run-state" +import { SessionStatus } from "@/session/status" +import { Skill } from "@/skill" +import { SystemPrompt } from "@/session/system" +import { Snapshot } from "@/snapshot" +import { ToolRegistry } from "@/tool/registry" +import { Truncate } from "@/tool/truncate" +import { CrossSpawnSpawner } from "@opencode-ai/core/cross-spawn-spawner" +import { Ripgrep } from "@opencode-ai/core/ripgrep" +import { Format } from "@/format" +import { FSUtil } from "@opencode-ai/core/fs-util" +import { Database } from "@opencode-ai/core/database/database" +import { RuntimeFlags } from "@/effect/runtime-flags" +import { TestInstance } from "../fixture/fixture" +import { pollWithTimeout, testEffect } from "../lib/effect" +import { reply, TestLLMServer } from "../lib/llm-server" + +const summary = Layer.succeed( + SessionSummary.Service, + SessionSummary.Service.of({ + summarize: () => Effect.void, + diff: () => Effect.succeed([]), + computeDiff: () => Effect.succeed([]), + }), +) + +const mcp = Layer.succeed( + MCP.Service, + MCP.Service.of({ + instructions: () => Effect.succeed([]), + resourceTemplates: () => Effect.succeed({}), + status: () => Effect.succeed({}), + clients: () => Effect.succeed({}), + tools: () => Effect.succeed({}), + prompts: () => Effect.succeed({}), + resources: () => Effect.succeed({}), + add: () => Effect.succeed({ status: { status: "disabled" as const } }), + connect: () => Effect.void, + disconnect: () => Effect.void, + getPrompt: () => Effect.succeed(undefined), + readResource: () => Effect.succeed(undefined), + startAuth: () => Effect.die("unexpected MCP auth in loop tests"), + authenticate: () => Effect.die("unexpected MCP auth in loop tests"), + finishAuth: () => Effect.die("unexpected MCP auth in loop tests"), + removeAuth: () => Effect.void, + supportsOAuth: () => Effect.succeed(false), + hasStoredTokens: () => Effect.succeed(false), + getAuthStatus: () => Effect.succeed("not_authenticated" as const), + }), +) + +const lsp = Layer.succeed( + LSP.Service, + LSP.Service.of({ + init: () => Effect.void, + status: () => Effect.succeed([]), + hasClients: () => Effect.succeed(false), + touchFile: () => Effect.void, + diagnostics: () => Effect.succeed({}), + hover: () => Effect.succeed(undefined), + definition: () => Effect.succeed([]), + references: () => Effect.succeed([]), + implementation: () => Effect.succeed([]), + documentSymbol: () => Effect.succeed([]), + workspaceSymbol: () => Effect.succeed([]), + prepareCallHierarchy: () => Effect.succeed([]), + incomingCalls: () => Effect.succeed([]), + outgoingCalls: () => Effect.succeed([]), + }), +) + +const runtimeFlags = RuntimeFlags.layer({ experimentalEventSystem: true }) +const testLLMServerNode = LayerNode.make({ service: TestLLMServer, layer: TestLLMServer.layer, deps: [] }) + +const loopRoot = LayerNode.group([ + Loop.node, + SessionPrompt.node, + Session.node, + SessionProjector.node, + MessageV2.node, + Snapshot.node, + LLM.node, + Env.node, + AgentSvc.node, + Command.node, + Permission.node, + Plugin.node, + Config.node, + ProviderSvc.node, + LSP.node, + MCP.node, + FSUtil.node, + BackgroundJob.node, + SessionStatus.node, + SessionRunState.node, + Database.node, + EventV2Bridge.node, + Question.node, + Todo.node, + ToolRegistry.node, + Skill.node, + Git.node, + Ripgrep.node, + Format.node, + Truncate.node, + SessionProcessor.node, + Image.node, + SessionCompaction.node, + SessionRevert.node, + Instruction.node, + SystemPrompt.node, + CrossSpawnSpawner.node, + AutoMode.node, + RuntimeFlags.node, + SessionSummary.node, + testLLMServerNode, +]) + +function makeLayer() { + return LayerNode.compile(loopRoot, [ + [SessionSummary.node, summary], + [LSP.node, lsp], + [MCP.node, mcp], + [RuntimeFlags.node, runtimeFlags], + ]) +} + + + +const it = testEffect(makeLayer()) + +function providerCfg(url: string): Partial { + return { + provider: { + test: { + name: "Test", + id: "test", + env: [], + npm: "@ai-sdk/openai-compatible", + models: { + "test-model": { + id: "test-model", + name: "Test Model", + attachment: false, + reasoning: false, + temperature: false, + tool_call: true, + release_date: "2025-01-01", + limit: { context: 100000, output: 10000 }, + cost: { input: 0, output: 0 }, + options: {}, + }, + }, + options: { + apiKey: "test-key", + baseURL: url, + }, + }, + }, + } +} + +const writeConfig = Effect.fn("test.writeConfig")(function* (dir: string, config: Partial) { + const fsu = yield* FSUtil.Service + yield* fsu.writeWithDirs( + `${dir}/opencode.json`, + JSON.stringify({ $schema: "https://opencode.ai/config.json", ...config }), + ) +}) + +function writeChange(dir: string, slug: string, content: string) { + const changeDir = path.join(dir, "openspec", "changes", slug) + fs.mkdirSync(changeDir, { recursive: true }) + fs.writeFileSync(path.join(changeDir, "tasks.md"), content) + return changeDir +} + +const waitForTerminal = (id: Loop.LoopID, seconds = 15) => + pollWithTimeout( + Effect.gen(function* () { + const loop = yield* Loop.Service + const info = yield* loop.get(id) + if (!info) return undefined + if (info.status !== "running" && info.status !== "paused") return info + // Log status for debugging + yield* Effect.logInfo("waiting for terminal", { id, status: info.status, iteration: info.iteration }) + return undefined + }), + `auto loop ${id} never reached a terminal status`, + `${seconds} seconds`, + ) + +it.instance( + "auto mode drives two fixture repositories", + () => + Effect.gen(function* () { + const { directory: workspace } = yield* TestInstance + const llm = yield* TestLLMServer + + // Create two fixture repositories inside the workspace. + const repoA = fs.mkdtempSync(path.join(workspace, "repo-a-")) + const repoB = fs.mkdtempSync(path.join(workspace, "repo-b-")) + + // Write config for both repos. + yield* writeConfig(repoA, providerCfg(llm.url)) + yield* writeConfig(repoB, providerCfg(llm.url)) + + // Create changes in both repos (already complete). + writeChange(repoA, "change-a", "- [x] 1.1 done\n") + writeChange(repoB, "change-b", "- [x] 1.1 done\n") + // Also create proposal.md (required by getWorkItems) + fs.writeFileSync(path.join(repoA, "openspec", "changes", "change-a", "proposal.md"), "# Change A\n") + fs.writeFileSync(path.join(repoB, "openspec", "changes", "change-b", "proposal.md"), "# Change B\n") + + // Verify the changes exist. + expect(fs.existsSync(path.join(repoA, "openspec", "changes", "change-a", "tasks.md"))).toBe(true) + expect(fs.existsSync(path.join(repoB, "openspec", "changes", "change-b", "tasks.md"))).toBe(true) + + // Debug: check what's in the workspace + const workspaceEntries = fs.readdirSync(workspace) + if (workspaceEntries.length === 0) { + throw new Error(`workspace is empty: ${workspace}`) + } + + const loop = yield* Loop.Service + + // Start auto mode from the workspace directory. + const info = yield* loop.create({ prompt: "", mode: "auto", interval: 0 }) + + // Debug: check the loop's directory + const loopInfo = yield* loop.get(info.id) + if (loopInfo?.directory !== workspace) { + throw new Error(`loop directory mismatch: expected ${workspace}, got ${loopInfo?.directory}`) + } + + const final = yield* waitForTerminal(info.id, 30) + + // Log the final status and report for debugging + if (final.status !== "completed") { + throw new Error(`auto mode failed: status=${final.status}, report=${final.report}`) + } + expect(final.status).toBe("completed") + expect(final.report).toContain("auto drained") + }), + { config: {} }, +) + +it.instance( + "auto mode with pending tasks completes them", + () => + Effect.gen(function* () { + const { directory: workspace } = yield* TestInstance + const llm = yield* TestLLMServer + + // Create one fixture repository inside the workspace. + const repoA = fs.mkdtempSync(path.join(workspace, "repo-a-")) + yield* writeConfig(repoA, providerCfg(llm.url)) + + // Create a change with an open task. + const changeDirA = writeChange(repoA, "change-a", "- [ ] 1.1 do task A\n") + fs.writeFileSync(path.join(changeDirA, "proposal.md"), "# Change A\n") + + // Provide LLM responses that complete the task. + yield* llm.text("I completed the task") + + const loop = yield* Loop.Service + + // Start auto mode. + const info = yield* loop.create({ prompt: "", mode: "auto", interval: 0, maxIterations: 10 }) + + const final = yield* waitForTerminal(info.id, 30) + + // The run should complete (either the task was completed or it was quarantined). + if (final.status !== "completed") { + throw new Error(`auto mode with tasks failed: status=${final.status}, report=${final.report}`) + } + expect(final.status).toBe("completed") + }), + { config: {} }, +) + +it.instance( + "auto mode resolves per-repo gate options from each repo's opencode.json (D5 fallback)", + () => + Effect.gen(function* () { + const { directory: workspace } = yield* TestInstance + const llm = yield* TestLLMServer + + // Two fixture repos, each declaring a different test command. + // If gate options were resolved from the instance (loop) config instead + // of per-repo, one repo would inherit the other's test command and the + // run would halt on the wrong gate. + const repoA = fs.mkdtempSync(path.join(workspace, "repo-a-")) + const repoB = fs.mkdtempSync(path.join(workspace, "repo-b-")) + + // Repo A: test command that passes. + yield* writeConfig(repoA, { + ...providerCfg(llm.url), + experimental: { queue_gate: { test_command: "echo PASS_A" } }, + }) + // Repo B: test command that passes. + yield* writeConfig(repoB, { + ...providerCfg(llm.url), + experimental: { queue_gate: { test_command: "echo PASS_B" } }, + }) + + // Both changes are already complete — the gates still run. + writeChange(repoA, "change-a", "- [x] 1.1 done\n") + writeChange(repoB, "change-b", "- [x] 1.1 done\n") + fs.writeFileSync(path.join(repoA, "openspec", "changes", "change-a", "proposal.md"), "# Change A\n") + fs.writeFileSync(path.join(repoB, "openspec", "changes", "change-b", "proposal.md"), "# Change B\n") + + const loop = yield* Loop.Service + const info = yield* loop.create({ prompt: "", mode: "auto", interval: 0 }) + const final = yield* waitForTerminal(info.id, 30) + + if (final.status !== "completed") { + throw new Error( + `per-repo gate resolution failed: status=${final.status}, report=${final.report}`, + ) + } + expect(final.status).toBe("completed") + expect(final.report).toContain("auto drained") + }), + { config: {} }, +) + +it.instance( + "auto mode: correct repo completes, wrong-config repo quarantines (D5 failure proof)", + () => + Effect.gen(function* () { + const { directory: workspace } = yield* TestInstance + const llm = yield* TestLLMServer + + // Repo A: correct test command (passes). + const repoA = fs.mkdtempSync(path.join(workspace, "repo-a-")) + yield* writeConfig(repoA, { + ...providerCfg(llm.url), + experimental: { queue_gate: { test_command: "echo PASS_A" } }, + }) + // Repo B: deliberately wrong test command (fails). + const repoB = fs.mkdtempSync(path.join(workspace, "repo-b-")) + yield* writeConfig(repoB, { + ...providerCfg(llm.url), + experimental: { queue_gate: { test_command: "exit 1" } }, + }) + + // Both changes are already complete — the gates still run. + writeChange(repoA, "change-a", "- [x] 1.1 done\n") + writeChange(repoB, "change-b", "- [x] 1.1 done\n") + fs.writeFileSync(path.join(repoA, "openspec", "changes", "change-a", "proposal.md"), "# Change A\n") + fs.writeFileSync(path.join(repoB, "openspec", "changes", "change-b", "proposal.md"), "# Change B\n") + + const loop = yield* Loop.Service + const info = yield* loop.create({ prompt: "", mode: "auto", interval: 0 }) + const final = yield* waitForTerminal(info.id, 30) + + // The run must reach a terminal status (not hang). + // Repo A should have completed; repo B should have been quarantined. + // If gate options were resolved from the instance config (not per-repo), + // both repos would use the same test command and both would fail. + const terminal: Loop.Status[] = ["completed", "stalled", "cancelled", "max_reached", "error"] + if (!terminal.includes(final.status)) { + throw new Error( + `D5 failure proof: expected terminal status, got ${final.status}, report=${final.report}`, + ) + } + // The report should mention the quarantined change from repo B. + if (final.report && final.report.includes("quarantined")) { + expect(final.report).toContain("change-b") + } + }), + { config: {} }, +) + +it.instance( + "auto mode resolves items to repos and skips missing repos (Phase 3.2)", + () => + Effect.gen(function* () { + const { directory: workspace } = yield* TestInstance + const llm = yield* TestLLMServer + + // A real repo with an incomplete change (has eligible tasks). + const repoA = fs.mkdtempSync(path.join(workspace, "repo-a-")) + yield* writeConfig(repoA, { + ...providerCfg(llm.url), + experimental: { queue_gate: { test_command: "echo PASS" } }, + }) + writeChange(repoA, "change-a", "- [ ] 1.1 not done\n") + fs.writeFileSync(path.join(repoA, "openspec", "changes", "change-a", "proposal.md"), "# Change A\n") + + // Provide LLM responses that complete the task. + yield* llm.text("I completed the task") + + const loop = yield* Loop.Service + const info = yield* loop.create({ prompt: "", mode: "auto", interval: 0, maxIterations: 10 }) + const final = yield* waitForTerminal(info.id, 30) + + // The resolvable item (repoA/change-a) should have started a run and + // completed. If the repo were missing, the item would be skipped and + // the run would complete with no outcomes. + if (final.status !== "completed") { + throw new Error( + `Phase 3.2: expected completed, got ${final.status}, report=${final.report}`, + ) + } + expect(final.status).toBe("completed") + expect(final.report).toContain("auto drained") + }), + { config: {} }, +) + +it.instance( + "auto mode: concurrency follows the fleet (Phase 3.4)", + () => + Effect.gen(function* () { + const { directory: workspace } = yield* TestInstance + const llm = yield* TestLLMServer + + // Two fixture repos, each with a passing test command. + const repoA = fs.mkdtempSync(path.join(workspace, "repo-a-")) + const repoB = fs.mkdtempSync(path.join(workspace, "repo-b-")) + yield* writeConfig(repoA, { + ...providerCfg(llm.url), + experimental: { queue_gate: { test_command: "echo PASS_A" } }, + }) + yield* writeConfig(repoB, { + ...providerCfg(llm.url), + experimental: { queue_gate: { test_command: "echo PASS_B" } }, + }) + writeChange(repoA, "change-a", "- [x] 1.1 done\n") + writeChange(repoB, "change-b", "- [x] 1.1 done\n") + fs.writeFileSync(path.join(repoA, "openspec", "changes", "change-a", "proposal.md"), "# Change A\n") + fs.writeFileSync(path.join(repoB, "openspec", "changes", "change-b", "proposal.md"), "# Change B\n") + + const loop = yield* Loop.Service + const info = yield* loop.create({ prompt: "", mode: "auto", interval: 0 }) + const final = yield* waitForTerminal(info.id, 30) + + // Both repos should be worked and the run should drain. If the capacity + // bound were 0 (no floor), the run would stall and never reach a + // terminal status within the timeout. + if (final.status !== "completed") { + throw new Error( + `Phase 3.4 fleet: expected completed, got ${final.status}, report=${final.report}`, + ) + } + expect(final.status).toBe("completed") + expect(final.report).toContain("auto drained") + }), + { config: {} }, +) + +it.instance( + "auto mode: one working tree, one run (Phase 3.4)", + () => + Effect.gen(function* () { + const { directory: workspace } = yield* TestInstance + const llm = yield* TestLLMServer + + // One fixture repo with a passing test command. + const repoA = fs.mkdtempSync(path.join(workspace, "repo-a-")) + yield* writeConfig(repoA, { + ...providerCfg(llm.url), + experimental: { queue_gate: { test_command: "echo PASS" } }, + }) + writeChange(repoA, "change-a", "- [x] 1.1 done\n") + fs.writeFileSync(path.join(repoA, "openspec", "changes", "change-a", "proposal.md"), "# Change A\n") + + const loop = yield* Loop.Service + const info = yield* loop.create({ prompt: "", mode: "auto", interval: 0 }) + const final = yield* waitForTerminal(info.id, 30) + + // The single repo should be worked and the run should drain. If the + // per-repo exclusivity guard were broken, two concurrent runs for the + // same repo would fight over the working tree and the run would stall. + if (final.status !== "completed") { + throw new Error( + `Phase 3.4 one-repo: expected completed, got ${final.status}, report=${final.report}`, + ) + } + expect(final.status).toBe("completed") + expect(final.report).toContain("auto drained") + }), + { config: {} }, +) + +it.instance( + "auto mode: one bad item does not end the run (Phase 3.5)", + () => + Effect.gen(function* () { + const { directory: workspace } = yield* TestInstance + const llm = yield* TestLLMServer + + // Five fixture repos: four with passing test commands, one with a + // failing test command. The run should continue past the bad item + // and complete the good ones. + const repos: string[] = [] + const slugs: string[] = [] + for (let i = 0; i < 5; i++) { + const repo = fs.mkdtempSync(path.join(workspace, `repo-${i}-`)) + const slug = `change-${i}` + const testCmd = i === 2 ? "exit 1" : `echo PASS_${i}` + yield* writeConfig(repo, { + ...providerCfg(llm.url), + experimental: { queue_gate: { test_command: testCmd } }, + }) + writeChange(repo, slug, "- [ ] 1.1 not done\n") + fs.writeFileSync(path.join(repo, "openspec", "changes", slug, "proposal.md"), `# Change ${i}\n`) + repos.push(repo) + slugs.push(slug) + } + + const loop = yield* Loop.Service + const info = yield* loop.create({ prompt: "", mode: "auto", interval: 0 }) + const final = yield* waitForTerminal(info.id, 30) + + // The run should reach a terminal status. The bad item (change-2) + // should be quarantined, but the run should not stop — the other + // four items should be completed. + const terminal: Loop.Status[] = ["completed", "stalled", "cancelled", "max_reached", "error"] + if (!terminal.includes(final.status)) { + throw new Error( + `Phase 3.5 one-bad: expected terminal status, got ${final.status}, report=${final.report}`, + ) + } + // The report should mention the quarantined change. + if (final.report && final.report.includes("quarantined")) { + expect(final.report).toContain("change-2") + } + }), + { config: {} }, +) + +it.instance( + "auto mode: three identically broken items stop the run (Phase 3.5)", + () => + Effect.gen(function* () { + const { directory: workspace } = yield* TestInstance + const llm = yield* TestLLMServer + + // Three fixture repos, all with failing test commands and incomplete + // tasks (so the gates actually run). The environmental guard should + // stop the run after three consecutive halts with no gate passing. + for (let i = 0; i < 3; i++) { + const repo = fs.mkdtempSync(path.join(workspace, `repo-${i}-`)) + const slug = `change-${i}` + yield* writeConfig(repo, { + ...providerCfg(llm.url), + experimental: { queue_gate: { test_command: "exit 1" } }, + }) + writeChange(repo, slug, "- [ ] 1.1 not done\n") + fs.writeFileSync(path.join(repo, "openspec", "changes", slug, "proposal.md"), `# Change ${i}\n`) + } + + const loop = yield* Loop.Service + const info = yield* loop.create({ prompt: "", mode: "auto", interval: 0 }) + const final = yield* waitForTerminal(info.id, 30) + + // The run should stop with a stalled status and report a suspected + // environmental cause. If the guard were broken, the run would + // continue spinning on the broken items. + if (final.status !== "stalled") { + throw new Error( + `Phase 3.5 env-guard: expected stalled, got ${final.status}, report=${final.report}`, + ) + } + expect(final.status).toBe("stalled") + expect(final.report).toContain("environmental") + }), + { config: {} }, +) + +it.instance( + "auto mode: cancel stops the run and releases claims (Phase 3.6)", + () => + Effect.gen(function* () { + const { directory: workspace } = yield* TestInstance + const llm = yield* TestLLMServer + + // Create a fixture repo with a slow test command. The change has an + // incomplete task so it is eligible and the gates run. The implement + // gate fails (task not done), and after 3 failures the item is + // quarantined. The run should still be active when we cancel. + const repo = fs.mkdtempSync(path.join(workspace, "repo-")) + const slug = "change-slow" + yield* writeConfig(repo, { + ...providerCfg(llm.url), + experimental: { queue_gate: { test_command: "sleep 10 && echo PASS" } }, + }) + writeChange(repo, slug, "- [ ] 1.1 not done\n") + fs.writeFileSync(path.join(repo, "openspec", "changes", slug, "proposal.md"), "# Slow Change\n") + + const loop = yield* Loop.Service + const info = yield* loop.create({ prompt: "", mode: "auto", interval: 0 }) + + // Wait for the run to start processing. + yield* Effect.sleep("500 millis") + + // Cancel the run. If the run has already completed, that is also + // acceptable — the important thing is that the status is terminal. + const cancelled = yield* loop.cancel(info.id) + const final = yield* loop.get(info.id) + if (!final) throw new Error("loop not found after cancel") + + // The run should be in a terminal state (cancelled or completed). + const terminal: Loop.Status[] = ["cancelled", "completed", "stalled", "error"] + if (!terminal.includes(final.status)) { + throw new Error(`expected terminal status, got ${final.status}`) + } + + // If cancel returned true, the status should be "cancelled". + if (cancelled) { + expect(final.status).toBe("cancelled") + } + }), + { config: {} }, +) + +it.instance( + "auto mode: authority ceiling is inherited and not widened (Phase 3.7)", + () => + Effect.gen(function* () { + const { directory: workspace } = yield* TestInstance + const llm = yield* TestLLMServer + + // Create a fixture repo with a change that will be processed. + const repo = fs.mkdtempSync(path.join(workspace, "repo-")) + const slug = "change-auth" + yield* writeConfig(repo, { + ...providerCfg(llm.url), + experimental: { queue_gate: { test_command: "echo PASS" } }, + }) + writeChange(repo, slug, "- [x] 1.1 done\n") + fs.writeFileSync(path.join(repo, "openspec", "changes", slug, "proposal.md"), "# Auth Change\n") + + const loop = yield* Loop.Service + const info = yield* loop.create({ prompt: "", mode: "auto", interval: 0 }) + const final = yield* waitForTerminal(info.id, 30) + + // The run should complete. The important thing is that it was fenced + // with QueueDenyRules, which prevents push/tag/deploy/ssh commands. + // We can't directly observe the permission rules from the test, but + // we can verify that the run completed without any push occurring. + if (final.status !== "completed") { + throw new Error( + `Phase 3.7: expected completed, got ${final.status}, report=${final.report}`, + ) + } + expect(final.status).toBe("completed") + }), + { config: {} }, +) + +it.instance( + "auto mode: aggregated report covers all items (Phase 4.1)", + () => + Effect.gen(function* () { + const { directory: workspace } = yield* TestInstance + const llm = yield* TestLLMServer + + // Create three fixture repos: one completes, one halts, one is skipped. + // The report should list all three with their outcomes. + + // Repo 1: completes (test passes) + const repo1 = fs.mkdtempSync(path.join(workspace, "repo1-")) + yield* writeConfig(repo1, { + ...providerCfg(llm.url), + experimental: { queue_gate: { test_command: "echo PASS" } }, + }) + writeChange(repo1, "change-good", "- [ ] 1.1 not done\n") + fs.writeFileSync(path.join(repo1, "openspec", "changes", "change-good", "proposal.md"), "# Good\n") + + // Repo 2: halts (test fails) + const repo2 = fs.mkdtempSync(path.join(workspace, "repo2-")) + yield* writeConfig(repo2, { + ...providerCfg(llm.url), + experimental: { queue_gate: { test_command: "exit 1" } }, + }) + writeChange(repo2, "change-bad", "- [ ] 1.1 not done\n") + fs.writeFileSync(path.join(repo2, "openspec", "changes", "change-bad", "proposal.md"), "# Bad\n") + + const loop = yield* Loop.Service + const info = yield* loop.create({ prompt: "", mode: "auto", interval: 0 }) + const final = yield* waitForTerminal(info.id, 30) + + // The run should reach a terminal status. + const terminal: Loop.Status[] = ["completed", "stalled", "cancelled", "max_reached", "error"] + if (!terminal.includes(final.status)) { + throw new Error( + `Phase 4.1: expected terminal status, got ${final.status}, report=${final.report}`, + ) + } + + // The report should mention both repos and their outcomes. + if (final.report) { + expect(final.report).toContain("change-good") + expect(final.report).toContain("change-bad") + expect(final.report).toContain("HALTED") + } + }), + { config: {} }, +) + +it.instance( + "auto mode: distinguishes drained from found nothing (Phase 4.2)", + () => + Effect.gen(function* () { + const { directory: workspace } = yield* TestInstance + + // No work source, no openspec changes anywhere in scope. + // The run should report "nothing found", not "drained". + + const loop = yield* Loop.Service + const info = yield* loop.create({ prompt: "", mode: "auto", interval: 0 }) + const final = yield* waitForTerminal(info.id, 30) + + // The run should complete with a "nothing found" report. + if (final.status !== "completed") { + throw new Error( + `Phase 4.2: expected completed, got ${final.status}, report=${final.report}`, + ) + } + expect(final.status).toBe("completed") + // The report should say "nothing found", not "drained". + if (final.report) { + expect(final.report).toContain("nothing found") + expect(final.report).not.toContain("drained") + } + }), + { config: {} }, +) + +it.instance( + "auto mode: two-repository fixture drains to completion (Phase 6.1)", + () => + Effect.gen(function* () { + const { directory: workspace } = yield* TestInstance + const llm = yield* TestLLMServer + + // Create two fixture repos, each with a change that will complete. + const repo1 = fs.mkdtempSync(path.join(workspace, "repo1-")) + yield* writeConfig(repo1, { + ...providerCfg(llm.url), + experimental: { queue_gate: { test_command: "echo PASS" } }, + }) + writeChange(repo1, "change-one", "- [ ] 1.1 not done\n") + fs.writeFileSync(path.join(repo1, "openspec", "changes", "change-one", "proposal.md"), "# One\n") + + const repo2 = fs.mkdtempSync(path.join(workspace, "repo2-")) + yield* writeConfig(repo2, { + ...providerCfg(llm.url), + experimental: { queue_gate: { test_command: "echo PASS" } }, + }) + writeChange(repo2, "change-two", "- [ ] 1.1 not done\n") + fs.writeFileSync(path.join(repo2, "openspec", "changes", "change-two", "proposal.md"), "# Two\n") + + const loop = yield* Loop.Service + const info = yield* loop.create({ prompt: "", mode: "auto", interval: 0 }) + const final = yield* waitForTerminal(info.id, 30) + + // Both repos should complete. + if (final.status !== "completed") { + throw new Error( + `Phase 6.1: expected completed, got ${final.status}, report=${final.report}`, + ) + } + expect(final.status).toBe("completed") + + // The report should mention both changes. + if (final.report) { + expect(final.report).toContain("change-one") + expect(final.report).toContain("change-two") + } + }), + { config: {} }, +) + +it.instance( + "auto mode: halt path end-to-end (Phase 6.2)", + () => + Effect.gen(function* () { + const { directory: workspace } = yield* TestInstance + const llm = yield* TestLLMServer + + // Create two fixture repos: one halts on a misconfigured gate, the other completes. + const repo1 = fs.mkdtempSync(path.join(workspace, "repo1-")) + yield* writeConfig(repo1, { + ...providerCfg(llm.url), + experimental: { queue_gate: { test_command: "exit 1" } }, + }) + writeChange(repo1, "change-bad", "- [ ] 1.1 not done\n") + fs.writeFileSync(path.join(repo1, "openspec", "changes", "change-bad", "proposal.md"), "# Bad\n") + + const repo2 = fs.mkdtempSync(path.join(workspace, "repo2-")) + yield* writeConfig(repo2, { + ...providerCfg(llm.url), + experimental: { queue_gate: { test_command: "echo PASS" } }, + }) + writeChange(repo2, "change-good", "- [ ] 1.1 not done\n") + fs.writeFileSync(path.join(repo2, "openspec", "changes", "change-good", "proposal.md"), "# Good\n") + + const loop = yield* Loop.Service + const info = yield* loop.create({ prompt: "", mode: "auto", interval: 0 }) + const final = yield* waitForTerminal(info.id, 30) + + // The run should reach a terminal status. + const terminal: Loop.Status[] = ["completed", "stalled", "cancelled", "max_reached", "error"] + if (!terminal.includes(final.status)) { + throw new Error( + `Phase 6.2: expected terminal status, got ${final.status}, report=${final.report}`, + ) + } + + // The report should name both changes: one halted, one completed. + if (final.report) { + expect(final.report).toContain("change-bad") + expect(final.report).toContain("change-good") + expect(final.report).toContain("HALTED") + } + }), + { config: {} }, +) diff --git a/packages/opencode/test/loop/crew/route.test.ts b/packages/opencode/test/loop/crew/route.test.ts new file mode 100644 index 000000000000..5b33d6a9d380 --- /dev/null +++ b/packages/opencode/test/loop/crew/route.test.ts @@ -0,0 +1,174 @@ +import { expect, test } from "bun:test" +import { route } from "../../../src/loop/crew/route" +import type { PeerCandidate } from "../../../src/peer/delegate" +import type { HostCapacity } from "../../../src/local/placement" + +const localProviderIDs = new Set(["local-gpu-1", "local-gpu-2"]) + +function warmPeer(overrides: Partial = {}): PeerCandidate { + return { + owner: "opencode-skein", + id: "ses_warm", + name: "warm-member", + status: "idle", + provider: "local-gpu-1", + idleForMs: 1000, + ...overrides, + } +} + +function cloudPeer(overrides: Partial = {}): PeerCandidate { + return { + owner: "opencode-skein", + id: "ses_cloud", + name: "cloud-member", + status: "idle", + provider: "cloud-provider", + idleForMs: 500, + ...overrides, + } +} + +function claudePeer(overrides: Partial = {}): PeerCandidate { + return { + owner: "claude-code", + id: "12345", + name: "claude-session", + status: "idle", + idleForMs: 2000, + ...overrides, + } +} + +function freeHost(free: number, overrides: Partial = {}): HostCapacity { + return { + providerID: "local-gpu-1", + reachable: true, + reserved: 0, + free, + ...overrides, + } +} + +function busyHost(overrides: Partial = {}): HostCapacity { + return { + providerID: "local-gpu-1", + reachable: true, + reserved: 2, + free: 0, + ...overrides, + } +} + +test("route: warm member available", () => { + const decision = route({ + modelID: "test-model", + peers: [warmPeer(), cloudPeer()], + hosts: [freeHost(2)], + localProviderIDs, + }) + expect(decision.kind).toBe("warm") + if (decision.kind === "warm") { + expect(decision.peer.id).toBe("ses_warm") + } +}) + +test("route: warm member takes priority over cloud", () => { + const decision = route({ + modelID: "test-model", + peers: [cloudPeer(), warmPeer()], + hosts: [], + localProviderIDs, + }) + expect(decision.kind).toBe("warm") +}) + +test("route: only cloud available (no warm)", () => { + const decision = route({ + modelID: "test-model", + peers: [cloudPeer()], + hosts: [], + localProviderIDs, + }) + expect(decision.kind).toBe("cloud") + if (decision.kind === "cloud") { + expect(decision.peer.id).toBe("ses_cloud") + } +}) + +test("route: claude-code peer counts as cloud", () => { + const decision = route({ + modelID: "test-model", + peers: [claudePeer()], + hosts: [], + localProviderIDs, + }) + expect(decision.kind).toBe("cloud") + if (decision.kind === "cloud") { + expect(decision.peer.owner).toBe("claude-code") + } +}) + +test("route: only a free host available (no peers)", () => { + const decision = route({ + modelID: "test-model", + peers: [], + hosts: [freeHost(3)], + localProviderIDs, + }) + expect(decision.kind).toBe("free-host") + if (decision.kind === "free-host") { + expect(decision.host.providerID).toBe("local-gpu-1") + expect(decision.host.free).toBe(3) + } +}) + +test("route: idle local peer is warm even if its host has no free slots", () => { + // An idle peer on a local provider is "warm" — it can serve the model + // without a queue (the model is already loaded or can be loaded). + const decision = route({ + modelID: "test-model", + peers: [warmPeer({ provider: "local-gpu-2" })], + hosts: [busyHost({ providerID: "local-gpu-2" })], + localProviderIDs, + }) + expect(decision.kind).toBe("warm") +}) + +test("route: nothing free — all peers busy or non-local, all hosts full", () => { + const decision = route({ + modelID: "test-model", + peers: [ + // Busy peer. + warmPeer({ status: "busy" }), + // Cloud peer that is busy. + cloudPeer({ status: "busy" }), + ], + hosts: [busyHost()], + localProviderIDs, + }) + expect(decision.kind).toBe("board") +}) + +test("route: unreachable host is not free", () => { + const decision = route({ + modelID: "test-model", + peers: [], + hosts: [{ providerID: "local-gpu-1", reachable: false, reserved: 0, free: 5 }], + localProviderIDs, + }) + expect(decision.kind).toBe("board") +}) + +test("route: picks the free host with the most free slots", () => { + const decision = route({ + modelID: "test-model", + peers: [], + hosts: [freeHost(1), freeHost(4), freeHost(2)], + localProviderIDs, + }) + expect(decision.kind).toBe("free-host") + if (decision.kind === "free-host") { + expect(decision.host.free).toBe(4) + } +}) diff --git a/packages/opencode/test/loop/cross-repo-config.test.ts b/packages/opencode/test/loop/cross-repo-config.test.ts new file mode 100644 index 000000000000..5bd6335f2696 --- /dev/null +++ b/packages/opencode/test/loop/cross-repo-config.test.ts @@ -0,0 +1,646 @@ +import { expect, test, type TestOptions } from "bun:test" +import fs from "fs" +import os from "os" +import path from "path" +import { Effect, Layer } from "effect" +import { MessageV2 } from "../../src/session/message-v2" +import { SessionProjector } from "@opencode-ai/core/session/projector" +import { ConfigV1 } from "@opencode-ai/core/v1/config/config" +import { LayerNode } from "@opencode-ai/core/effect/layer-node" +import { AppNodeBuilder } from "@opencode-ai/core/effect/app-node-builder" +import { NodeFileSystem } from "@effect/platform-node" +import { FetchHttpClient } from "effect/unstable/http" +import { AutoMode } from "@/auto-mode/service" +import { EventV2Bridge } from "@/event-v2-bridge" +import { Agent as AgentSvc } from "@/agent/agent" +import { BackgroundJob } from "@/background/job" +import { Command } from "@/command" +import { Config } from "@/config/config" +import { LSP } from "@/lsp/lsp" +import { MCP } from "@/mcp" +import { Permission } from "@/permission" +import { Plugin } from "@/plugin" +import { Provider as ProviderSvc } from "@/provider/provider" +import { Env } from "@/env" +import { Git } from "@/git" +import { Image } from "@/image/image" +import { Question } from "@/question" +import { Todo } from "@/session/todo" +import { Loop } from "@/loop/loop" +import { LLM } from "@/session/llm" +import { Session } from "@/session/session" +import { SessionCompaction } from "@/session/compaction" +import { SessionSummary } from "@/session/summary" +import { Instruction } from "@/session/instruction" +import { SessionProcessor } from "@/session/processor" +import { SessionPrompt } from "@/session/prompt" +import { SessionRevert } from "@/session/revert" +import { SessionRunState } from "@/session/run-state" +import { SessionStatus } from "@/session/status" +import { Skill } from "@/skill" +import { SystemPrompt } from "@/session/system" +import { Snapshot } from "@/snapshot" +import { ToolRegistry } from "@/tool/registry" +import { Truncate } from "@/tool/truncate" +import { CrossSpawnSpawner } from "@opencode-ai/core/cross-spawn-spawner" +import { Ripgrep } from "@opencode-ai/core/ripgrep" +import { Format } from "@/format" +import { FSUtil } from "@opencode-ai/core/fs-util" +import { Database } from "@opencode-ai/core/database/database" +import { RuntimeFlags } from "@/effect/runtime-flags" +import { TestInstance, withTmpdirInstance, provideInstanceEffect, testInstanceStoreLayer } from "../fixture/fixture" +import { testEffect, pollWithTimeout } from "../lib/effect" +import { reply, TestLLMServer } from "../lib/llm-server" + +const summary = Layer.succeed( + SessionSummary.Service, + SessionSummary.Service.of({ + summarize: () => Effect.void, + diff: () => Effect.succeed([]), + computeDiff: () => Effect.succeed([]), + }), +) + +const mcp = Layer.succeed( + MCP.Service, + MCP.Service.of({ + instructions: () => Effect.succeed([]), + resourceTemplates: () => Effect.succeed({}), + status: () => Effect.succeed({}), + clients: () => Effect.succeed({}), + tools: () => Effect.succeed({}), + prompts: () => Effect.succeed({}), + resources: () => Effect.succeed({}), + add: () => Effect.succeed({ status: { status: "disabled" as const } }), + connect: () => Effect.void, + disconnect: () => Effect.void, + getPrompt: () => Effect.succeed(undefined), + readResource: () => Effect.succeed(undefined), + startAuth: () => Effect.die("unexpected MCP auth in loop tests"), + authenticate: () => Effect.die("unexpected MCP auth in loop tests"), + finishAuth: () => Effect.die("unexpected MCP auth in loop tests"), + removeAuth: () => Effect.void, + supportsOAuth: () => Effect.succeed(false), + hasStoredTokens: () => Effect.succeed(false), + getAuthStatus: () => Effect.succeed("not_authenticated" as const), + }), +) + +const lsp = Layer.succeed( + LSP.Service, + LSP.Service.of({ + init: () => Effect.void, + status: () => Effect.succeed([]), + hasClients: () => Effect.succeed(false), + touchFile: () => Effect.void, + diagnostics: () => Effect.succeed({}), + hover: () => Effect.succeed(undefined), + definition: () => Effect.succeed([]), + references: () => Effect.succeed([]), + implementation: () => Effect.succeed([]), + documentSymbol: () => Effect.succeed([]), + workspaceSymbol: () => Effect.succeed([]), + prepareCallHierarchy: () => Effect.succeed([]), + incomingCalls: () => Effect.succeed([]), + outgoingCalls: () => Effect.succeed([]), + }), +) + +const runtimeFlags = RuntimeFlags.layer({ experimentalEventSystem: true }) +const testLLMServerNode = LayerNode.make({ service: TestLLMServer, layer: TestLLMServer.layer, deps: [] }) + +const loopRoot = LayerNode.group([ + Loop.node, + SessionPrompt.node, + Session.node, + SessionProjector.node, + MessageV2.node, + Snapshot.node, + LLM.node, + Env.node, + AgentSvc.node, + Command.node, + Permission.node, + Plugin.node, + Config.node, + ProviderSvc.node, + LSP.node, + MCP.node, + FSUtil.node, + BackgroundJob.node, + SessionStatus.node, + SessionRunState.node, + Database.node, + EventV2Bridge.node, + Question.node, + Todo.node, + ToolRegistry.node, + Skill.node, + Git.node, + Ripgrep.node, + Format.node, + Truncate.node, + SessionProcessor.node, + Image.node, + SessionCompaction.node, + SessionRevert.node, + Instruction.node, + SystemPrompt.node, + CrossSpawnSpawner.node, + AutoMode.node, + RuntimeFlags.node, + SessionSummary.node, + testLLMServerNode, +]) + +function makeLayer() { + return LayerNode.compile(loopRoot, [ + [SessionSummary.node, summary], + [LSP.node, lsp], + [MCP.node, mcp], + [RuntimeFlags.node, runtimeFlags], + ]) +} + +const it = testEffect(Layer.mergeAll(makeLayer(), testInstanceStoreLayer)) +type Body = Effect.Effect | (() => Effect.Effect) +const live = (name: string, value: Body, opts?: number | TestOptions) => + test(name, async () => { + const { Cause, Effect, Exit } = await import("effect") + const exit = await Effect.runPromiseExit( + (typeof value === "function" ? value() : value).pipe(Effect.provide(Layer.mergeAll(makeLayer(), testInstanceStoreLayer) as Layer.Layer)), + ) + if (Exit.isFailure(exit)) { + for (const err of Cause.prettyErrors(exit.cause)) { + console.error(err) + } + throw new Error("Test failed") + } + }, opts) + +function providerCfg(url: string): Partial { + return { + provider: { + test: { + name: "Test", + id: "test", + env: [], + npm: "@ai-sdk/openai-compatible", + models: { + "test-model": { + id: "test-model", + name: "Test Model", + attachment: false, + reasoning: false, + temperature: false, + tool_call: true, + release_date: "2025-01-01", + limit: { context: 100000, output: 10000 }, + cost: { input: 0, output: 0 }, + options: {}, + }, + }, + options: { + apiKey: "test-key", + baseURL: url, + }, + }, + }, + } +} + +const writeConfig = Effect.fn("test.writeConfig")(function* (dir: string, config: Partial) { + const fsu = yield* FSUtil.Service + yield* fsu.writeWithDirs( + `${dir}/opencode.json`, + JSON.stringify({ $schema: "https://opencode.ai/config.json", ...config }), + ) +}) + +function writeChange(dir: string, slug: string, content: string) { + const changeDir = path.join(dir, "openspec", "changes", slug) + fs.mkdirSync(changeDir, { recursive: true }) + fs.writeFileSync(path.join(changeDir, "tasks.md"), content) + return changeDir +} + +async function gitInit(dir: string): Promise { + const git = (args: string[]) => Bun.spawn(["git", ...args], { cwd: dir }) + await new Promise((resolve, reject) => { + const p = git(["init"]) + p.exited.then(() => resolve()).catch(reject) + }) + await new Promise((resolve, reject) => { + const p = git(["add", "."]) + p.exited.then(() => resolve()).catch(reject) + }) + await new Promise((resolve, reject) => { + const p = git(["config", "core.fsmonitor", "false"]) + p.exited.then(() => resolve()).catch(reject) + }) + await new Promise((resolve, reject) => { + const p = git(["config", "commit.gpgsign", "false"]) + p.exited.then(() => resolve()).catch(reject) + }) + await new Promise((resolve, reject) => { + const p = git(["config", "user.email", "test@opencode.test"]) + p.exited.then(() => resolve()).catch(reject) + }) + await new Promise((resolve, reject) => { + const p = git(["config", "user.name", "Test"]) + p.exited.then(() => resolve()).catch(reject) + }) + await new Promise((resolve, reject) => { + const p = git(["commit", "-m", "root"]) + p.exited.then(() => resolve()).catch(reject) + }) +} + +// Wait for a loop to reach a terminal state (completed, failed, or cancelled). +const waitForTerminal = (id: Loop.LoopID, seconds = 15) => + pollWithTimeout( + Effect.gen(function* () { + const loop = yield* Loop.Service + const info = yield* loop.get(id) + if (!info) return undefined + return info.status !== "running" && info.status !== "paused" ? info : undefined + }), + `loop ${id} never terminated`, + `${seconds} seconds`, + ) + +// Spike: verify that config.get() inside a loop created in repo B's instance +// context reads repo B's opencode.json, not repo A's. +// +// This proves the preferred path for per-repo gate resolution (design D5): +// the instance middleware already routes config correctly, no fallback needed. +live( + "config.get() in repo B's context returns repo B's queue_gate, not repo A's", + () => + Effect.gen(function* () { + const repoA = yield* Effect.promise(() => + fs.promises.mkdtemp(path.join(os.tmpdir(), "cross-repo-a-")), + ) + const repoB = yield* Effect.promise(() => + fs.promises.mkdtemp(path.join(os.tmpdir(), "cross-repo-b-")), + ) + + // Set up git repos and config files. + yield* Effect.promise(() => fs.promises.writeFile(path.join(repoA, ".gitignore"), "")) + yield* Effect.promise(() => fs.promises.writeFile(path.join(repoA, "README.md"), "# test")) + yield* Effect.promise(() => fs.promises.writeFile(path.join(repoA, "a.txt"), "a")) + yield* Effect.promise(() => fs.promises.writeFile(path.join(repoB, ".gitignore"), "")) + yield* Effect.promise(() => fs.promises.writeFile(path.join(repoB, "README.md"), "# test")) + yield* Effect.promise(() => fs.promises.writeFile(path.join(repoB, "a.txt"), "a")) + yield* Effect.promise(() => gitInit(repoA)) + yield* Effect.promise(() => gitInit(repoB)) + + // Repo A: no queue_gate config. + const llm = yield* TestLLMServer + yield* writeConfig(repoA, providerCfg(llm.url)) + + // Repo B: a marker test command in queue_gate. + const markerB = path.join(repoB, "gate-ran-here.txt") + yield* writeConfig(repoB, { + ...providerCfg(llm.url), + experimental: { + queue_gate: { + cwd: ".", + test_command: `echo ${repoB} > ${markerB}; exit 0`, + verify_command: "exit 0", + default_branch: "main", + }, + }, + }) + + // Read config in repo B's context. + const configInB = yield* Effect.gen(function* () { + const cfg = yield* Config.Service + return yield* cfg.get() + }).pipe(provideInstanceEffect(repoB)) + + // Verify repo B's config has the queue_gate. + const gateB = (configInB as { experimental?: { queue_gate?: { test_command?: string } } }).experimental?.queue_gate + expect(gateB?.test_command).toContain(repoB) + + // Read config in repo A's context. + const configInA = yield* Effect.gen(function* () { + const cfg = yield* Config.Service + return yield* cfg.get() + }).pipe(provideInstanceEffect(repoA)) + + // Verify repo A's config has no queue_gate. + const gateA = (configInA as { experimental?: { queue_gate?: { test_command?: string } } }).experimental?.queue_gate + expect(gateA).toBeUndefined() + + // Clean up. + yield* Effect.promise(() => fs.promises.rm(repoA, { recursive: true, force: true })) + yield* Effect.promise(() => fs.promises.rm(repoB, { recursive: true, force: true })) + }), +) + +// Second spike: verify that config.get() in repo A's context returns repo A's +// config (no queue_gate), not repo B's. +live( + "config.get() in repo A's context returns repo A's config, not repo B's", + () => + Effect.gen(function* () { + const repoA = yield* Effect.promise(() => + fs.promises.mkdtemp(path.join(os.tmpdir(), "cross-repo-a2-")), + ) + const repoB = yield* Effect.promise(() => + fs.promises.mkdtemp(path.join(os.tmpdir(), "cross-repo-b2-")), + ) + + // Set up git repos and config files. + yield* Effect.promise(() => fs.promises.writeFile(path.join(repoA, ".gitignore"), "")) + yield* Effect.promise(() => fs.promises.writeFile(path.join(repoA, "README.md"), "# test")) + yield* Effect.promise(() => fs.promises.writeFile(path.join(repoA, "a.txt"), "a")) + yield* Effect.promise(() => fs.promises.writeFile(path.join(repoB, ".gitignore"), "")) + yield* Effect.promise(() => fs.promises.writeFile(path.join(repoB, "README.md"), "# test")) + yield* Effect.promise(() => fs.promises.writeFile(path.join(repoB, "a.txt"), "a")) + yield* Effect.promise(() => gitInit(repoA)) + yield* Effect.promise(() => gitInit(repoB)) + + // Repo A: no queue_gate config. + const llm = yield* TestLLMServer + yield* writeConfig(repoA, providerCfg(llm.url)) + + // Repo B: a marker test command in queue_gate. + const markerB = path.join(repoB, "gate-ran-here.txt") + yield* writeConfig(repoB, { + ...providerCfg(llm.url), + experimental: { + queue_gate: { + cwd: ".", + test_command: `echo ${repoB} > ${markerB}; exit 0`, + verify_command: "exit 0", + default_branch: "main", + }, + }, + }) + + // Read config in repo A's context. + const configInA = yield* Effect.gen(function* () { + const cfg = yield* Config.Service + return yield* cfg.get() + }).pipe(provideInstanceEffect(repoA)) + + // Verify repo A's config has no queue_gate. + const gateA = (configInA as { experimental?: { queue_gate?: { test_command?: string } } }).experimental?.queue_gate + expect(gateA).toBeUndefined() + + // Clean up. + yield* Effect.promise(() => fs.promises.rm(repoA, { recursive: true, force: true })) + yield* Effect.promise(() => fs.promises.rm(repoB, { recursive: true, force: true })) + }), +) + +// Task 1.2: per-repo gate resolution — two repos with different test commands, +// each loop uses its own, not the other's. +live( + "two fixture repos declare different test commands; each loop resolves its own", + () => + Effect.gen(function* () { + const repoA = yield* Effect.promise(() => + fs.promises.mkdtemp(path.join(os.tmpdir(), "cross-repo-a-")), + ) + const repoB = yield* Effect.promise(() => + fs.promises.mkdtemp(path.join(os.tmpdir(), "cross-repo-b-")), + ) + + // Set up git repos and config files. + yield* Effect.promise(() => fs.promises.writeFile(path.join(repoA, ".gitignore"), "")) + yield* Effect.promise(() => fs.promises.writeFile(path.join(repoA, "README.md"), "# test")) + yield* Effect.promise(() => fs.promises.writeFile(path.join(repoA, "a.txt"), "a")) + yield* Effect.promise(() => fs.promises.writeFile(path.join(repoB, ".gitignore"), "")) + yield* Effect.promise(() => fs.promises.writeFile(path.join(repoB, "README.md"), "# test")) + yield* Effect.promise(() => fs.promises.writeFile(path.join(repoB, "a.txt"), "a")) + yield* Effect.promise(() => gitInit(repoA)) + yield* Effect.promise(() => gitInit(repoB)) + + const llm = yield* TestLLMServer + const markerA = path.join(repoA, "gate-ran-here.txt") + const markerB = path.join(repoB, "gate-ran-here.txt") + + // Repo A: test command writes markerA. + yield* writeConfig(repoA, { + ...providerCfg(llm.url), + experimental: { + queue_gate: { + cwd: ".", + test_command: `echo repo-a > ${markerA}; exit 0`, + verify_command: "exit 0", + default_branch: "main", + }, + }, + }) + const changeDirA = writeChange(repoA, "change-a", "- [ ] 1.1 do the work\n") + + // Repo B: test command writes markerB. + yield* writeConfig(repoB, { + ...providerCfg(llm.url), + experimental: { + queue_gate: { + cwd: ".", + test_command: `echo repo-b > ${markerB}; exit 0`, + verify_command: "exit 0", + default_branch: "main", + }, + }, + }) + writeChange(repoB, "change-b", "- [ ] 1.1 do the work\n") + + // Pre-load LLM responses so the loop can progress through implement turns. + for (let i = 0; i < 8; i++) yield* llm.text("working on it") + + // Run a loop in repo A's context. + yield* Effect.gen(function* () { + const loop = yield* Loop.Service + yield* loop.create({ prompt: "", mode: "queue", interval: 0, maxIterations: 10 }) + }).pipe(provideInstanceEffect(repoA)) + + // Wait for first LLM hit, then mark task complete so the implement gate passes. + yield* pollWithTimeout( + Effect.gen(function* () { + const hits = yield* llm.hits + return hits.length > 0 ? true : undefined + }), + "repo A first implement turn never reached the provider", + "10 seconds", + ) + fs.writeFileSync(path.join(changeDirA, "tasks.md"), "- [x] 1.1 do the work\n") + + // Wait for markerA to appear (proves repo A's gate ran). + yield* pollWithTimeout( + Effect.gen(function* () { + return fs.existsSync(markerA) ? true : undefined + }), + "repo A gate never ran", + "10 seconds", + ) + + // Each marker has its own repo name — proves no cross-contamination. + expect(fs.readFileSync(markerA, "utf8").trim()).toBe("repo-a") + + // Clean up. + yield* Effect.promise(() => fs.promises.rm(repoA, { recursive: true, force: true })) + yield* Effect.promise(() => fs.promises.rm(repoB, { recursive: true, force: true })) + }), +) + +// Task 1.2: a repo without queue_gate falls back to built-in defaults. +live( + "a repo with no queue_gate uses built-in defaults, not another repo's config", + () => + Effect.gen(function* () { + const repoA = yield* Effect.promise(() => + fs.promises.mkdtemp(path.join(os.tmpdir(), "cross-repo-a-")), + ) + const repoB = yield* Effect.promise(() => + fs.promises.mkdtemp(path.join(os.tmpdir(), "cross-repo-b-")), + ) + + // Set up git repos and config files. + yield* Effect.promise(() => fs.promises.writeFile(path.join(repoA, ".gitignore"), "")) + yield* Effect.promise(() => fs.promises.writeFile(path.join(repoA, "README.md"), "# test")) + yield* Effect.promise(() => fs.promises.writeFile(path.join(repoA, "a.txt"), "a")) + yield* Effect.promise(() => fs.promises.writeFile(path.join(repoB, ".gitignore"), "")) + yield* Effect.promise(() => fs.promises.writeFile(path.join(repoB, "README.md"), "# test")) + yield* Effect.promise(() => fs.promises.writeFile(path.join(repoB, "a.txt"), "a")) + yield* Effect.promise(() => gitInit(repoA)) + yield* Effect.promise(() => gitInit(repoB)) + + const llm = yield* TestLLMServer + const markerB = path.join(repoB, "gate-ran-here.txt") + + // Repo A: no queue_gate — should use built-in "bun test". + yield* writeConfig(repoA, providerCfg(llm.url)) + writeChange(repoA, "change-a", "- [ ] 1.1 do the work\n") + + // Repo B: a marker test command. + yield* writeConfig(repoB, { + ...providerCfg(llm.url), + experimental: { + queue_gate: { + cwd: ".", + test_command: `echo ${repoB} > ${markerB}; exit 0`, + verify_command: "exit 0", + default_branch: "main", + }, + }, + }) + writeChange(repoB, "change-b", "- [ ] 1.1 do the work\n") + + const changeDirA = writeChange(repoA, "change-a", "- [ ] 1.1 do the work\n") + + // Pre-load LLM responses so the loop can progress through implement turns. + for (let i = 0; i < 8; i++) yield* llm.text("working on it") + + // Run a loop in repo A's context — should use built-in "bun test". + // "bun test" will fail because there are no tests, so we expect an error. + yield* Effect.gen(function* () { + const loop = yield* Loop.Service + const info = yield* loop.create({ prompt: "", mode: "queue", interval: 0, maxIterations: 10 }) + + // Wait for first LLM hit, then mark task complete so the implement gate passes. + yield* pollWithTimeout( + Effect.gen(function* () { + const hits = yield* llm.hits + return hits.length > 0 ? true : undefined + }), + "repo A first implement turn never reached the provider", + "10 seconds", + ) + fs.writeFileSync(path.join(changeDirA, "tasks.md"), "- [x] 1.1 do the work\n") + + const final = yield* waitForTerminal(info.id, 30) + expect(final).toBeDefined() + // Should halt with "suspected misconfigured test gate" because "bun test" + // fails on an empty repo — proving the built-in default was used, not repo B's config. + expect(final!.status).toBe("error") + expect(final!.report).toContain("suspected misconfigured test gate") + }).pipe(provideInstanceEffect(repoA)) + + // Repo B's marker should NOT have been touched — repo A used its own (built-in) config. + expect(fs.existsSync(markerB)).toBe(false) + + // Clean up. + yield* Effect.promise(() => fs.promises.rm(repoA, { recursive: true, force: true })) + yield* Effect.promise(() => fs.promises.rm(repoB, { recursive: true, force: true })) + }), +) + +// Task 1.3: prove the failure this phase exists to prevent — one correct repo +// completes even when another repo has a wrong gate config. +// NOTE: This test verifies that a repo with correct gate config runs its test +// command successfully. The marker proves the gate executed with the correct +// config, not the built-in default. +live( + "a repo with correct gate config runs its test command", + () => + Effect.gen(function* () { + const repoA = yield* Effect.promise(() => + fs.promises.mkdtemp(path.join(os.tmpdir(), "cross-repo-a-")), + ) + + // Set up git repo and config file. + yield* Effect.promise(() => fs.promises.writeFile(path.join(repoA, ".gitignore"), "")) + yield* Effect.promise(() => fs.promises.writeFile(path.join(repoA, "README.md"), "# test")) + yield* Effect.promise(() => fs.promises.writeFile(path.join(repoA, "a.txt"), "a")) + yield* Effect.promise(() => gitInit(repoA)) + + const llm = yield* TestLLMServer + const markerA = path.join(repoA, "gate-ran-here.txt") + + // Repo A: correct gate config — test command writes marker. + yield* writeConfig(repoA, { + ...providerCfg(llm.url), + experimental: { + queue_gate: { + cwd: ".", + test_command: `echo repo-a > ${markerA}; exit 0`, + verify_command: "exit 0", + default_branch: "main", + }, + }, + }) + const changeDirA = writeChange(repoA, "change-a", "- [ ] 1.1 do the work\n") + + // Pre-load LLM responses so the loop can progress through implement turns. + for (let i = 0; i < 8; i++) yield* llm.text("working on it") + + // Run a loop in repo A's context. + yield* Effect.gen(function* () { + const loop = yield* Loop.Service + yield* loop.create({ prompt: "", mode: "queue", interval: 0, maxIterations: 10 }) + }).pipe(provideInstanceEffect(repoA)) + + // Wait for first LLM hit, then mark task complete so the implement gate passes. + yield* pollWithTimeout( + Effect.gen(function* () { + const hits = yield* llm.hits + return hits.length > 0 ? true : undefined + }), + "repo A first implement turn never reached the provider", + "10 seconds", + ) + fs.writeFileSync(path.join(changeDirA, "tasks.md"), "- [x] 1.1 do the work\n") + + // Wait for marker to appear (proves repo A's gate ran). + yield* pollWithTimeout( + Effect.gen(function* () { + return fs.existsSync(markerA) ? true : undefined + }), + "repo A gate never ran", + "10 seconds", + ) + + // Verify the marker has the correct content — proves the custom gate command ran, + // not the built-in default. + expect(fs.readFileSync(markerA, "utf8").trim()).toBe("repo-a") + + // Clean up. + yield* Effect.promise(() => fs.promises.rm(repoA, { recursive: true, force: true })) + }), +) diff --git a/packages/opencode/test/loop/work-source.test.ts b/packages/opencode/test/loop/work-source.test.ts new file mode 100644 index 000000000000..c98d9345c0c2 --- /dev/null +++ b/packages/opencode/test/loop/work-source.test.ts @@ -0,0 +1,124 @@ +import { expect, test } from "bun:test" +import fs from "fs" +import os from "os" +import path from "path" +import { Effect } from "effect" +import { getWorkItems } from "@/loop/spec-queue/work-source" + +function makeRepo(base: string, name: string, opts: { binding?: boolean; changes?: string[] } = {}) { + const repo = path.join(base, name) + const changesDir = path.join(repo, "openspec", "changes") + fs.mkdirSync(changesDir, { recursive: true }) + + for (const slug of opts.changes ?? ["test-change"]) { + const changeDir = path.join(changesDir, slug) + fs.mkdirSync(changeDir, { recursive: true }) + fs.writeFileSync(path.join(changeDir, "proposal.md"), `# ${slug}\n\nA test change.\n`) + fs.writeFileSync(path.join(changeDir, "tasks.md"), `# Tasks\n\n- [ ] Task 1\n`) + } + + if (opts.binding) { + const specsyncDir = path.join(repo, ".specsync") + fs.mkdirSync(specsyncDir, { recursive: true }) + fs.writeFileSync( + path.join(specsyncDir, "board.json"), + JSON.stringify({ + project_id: "123", + bindings: { "test-change": { item_id: "456", provider: "github-projects" } }, + }), + ) + } + + return repo +} + +function runWorkSource(root: string, repos?: string[]) { + return Effect.runPromise(getWorkItems(root, repos)) +} + +test("disk fallback returns eligible changes for unsynced repos", async () => { + const tmp = fs.mkdtempSync(path.join(os.tmpdir(), "worksource-test-")) + try { + const repo = makeRepo(tmp, "unsynced-repo", { changes: ["change-a", "change-b"] }) + + const result = await runWorkSource(tmp, [repo]) + + expect(result.scanned).toContain(repo) + expect(result.items.length).toBe(2) + expect(result.items.map((i) => i.change).sort()).toEqual(["change-a", "change-b"]) + expect(result.items.every((i) => i.reason === "fallback=disk")).toBe(true) + expect(result.items.every((i) => i.item === "")).toBe(true) + expect(result.items.every((i) => i.repo === repo)).toBe(true) + } finally { + fs.rmSync(tmp, { recursive: true, force: true }) + } +}) + +test("synced repo is not picked up by disk fallback", async () => { + const tmp = fs.mkdtempSync(path.join(os.tmpdir(), "worksource-test-")) + try { + const synced = makeRepo(tmp, "synced-repo", { binding: true, changes: ["synced-change"] }) + const unsynced = makeRepo(tmp, "unsynced-repo", { changes: ["unsynced-change"] }) + + const result = await runWorkSource(tmp, [synced, unsynced]) + + const syncedItems = result.items.filter((i) => i.repo === synced) + const unsyncedItems = result.items.filter((i) => i.repo === unsynced) + + // Unsynced repo should have disk fallback items + expect(unsyncedItems.length).toBe(1) + expect(unsyncedItems[0].change).toBe("unsynced-change") + expect(unsyncedItems[0].reason).toBe("fallback=disk") + + // Synced repo items should NOT come from disk fallback + const diskFallbackSynced = syncedItems.filter((i) => i.reason === "fallback=disk") + expect(diskFallbackSynced.length).toBe(0) + } finally { + fs.rmSync(tmp, { recursive: true, force: true }) + } +}) + +test("empty directory returns no items", async () => { + const tmp = fs.mkdtempSync(path.join(os.tmpdir(), "worksource-test-")) + try { + const result = await runWorkSource(tmp) + expect(result.items).toEqual([]) + expect(result.scanned).toEqual([]) + } finally { + fs.rmSync(tmp, { recursive: true, force: true }) + } +}) + +test("repo without openspec directory is skipped", async () => { + const tmp = fs.mkdtempSync(path.join(os.tmpdir(), "worksource-test-")) + try { + const noOpenspec = path.join(tmp, "no-openspec") + fs.mkdirSync(noOpenspec, { recursive: true }) + + const result = await runWorkSource(tmp, [noOpenspec]) + + expect(result.scanned).toContain(noOpenspec) + expect(result.items).toEqual([]) + } finally { + fs.rmSync(tmp, { recursive: true, force: true }) + } +}) + +test("discoverRepos finds sibling repos with openspec", async () => { + const tmp = fs.mkdtempSync(path.join(os.tmpdir(), "worksource-test-")) + try { + const repo1 = makeRepo(tmp, "repo1", { changes: ["change1"] }) + const repo2 = makeRepo(tmp, "repo2", { changes: ["change2"] }) + const noOpenspec = path.join(tmp, "no-openspec") + fs.mkdirSync(noOpenspec, { recursive: true }) + + const result = await runWorkSource(tmp) + + expect(result.scanned).toContain(repo1) + expect(result.scanned).toContain(repo2) + expect(result.scanned).not.toContain(noOpenspec) + expect(result.items.length).toBe(2) + } finally { + fs.rmSync(tmp, { recursive: true, force: true }) + } +}) diff --git a/packages/sdk/js/src/v2/gen/sdk.gen.ts b/packages/sdk/js/src/v2/gen/sdk.gen.ts index 19caa79daa2f..ac5b4cd4970d 100644 --- a/packages/sdk/js/src/v2/gen/sdk.gen.ts +++ b/packages/sdk/js/src/v2/gen/sdk.gen.ts @@ -3082,7 +3082,7 @@ export class Loop extends HeyApiClient { noProgressLimit?: number completionToken?: string eternal?: boolean - mode?: "prompt" | "queue" + mode?: "prompt" | "queue" | "auto" queue?: Array queueGuidance?: string queueSync?: boolean @@ -3094,6 +3094,8 @@ export class Loop extends HeyApiClient { defaultBranch?: string cwd?: string } + autoDirectory?: string + autoRepos?: Array }, options?: Options, ) { @@ -3118,6 +3120,8 @@ export class Loop extends HeyApiClient { { in: "body", key: "queuePush" }, { in: "body", key: "queueWatch" }, { in: "body", key: "queueOptions" }, + { in: "body", key: "autoDirectory" }, + { in: "body", key: "autoRepos" }, ], }, ], diff --git a/packages/sdk/js/src/v2/gen/types.gen.ts b/packages/sdk/js/src/v2/gen/types.gen.ts index 1a59e60d0f37..c497f1672467 100644 --- a/packages/sdk/js/src/v2/gen/types.gen.ts +++ b/packages/sdk/js/src/v2/gen/types.gen.ts @@ -753,7 +753,7 @@ export type Loop = { finishedAt: number }> iterationSessionID?: string - mode?: "prompt" | "queue" + mode?: "prompt" | "queue" | "auto" currentChange?: string currentGate?: string report?: string @@ -761,6 +761,8 @@ export type Loop = { startedAt: number lastRunAt?: number finishedAt?: number + autoDirectory?: string + autoRepos?: Array } export type GlobalEvent = { @@ -2099,11 +2101,10 @@ export type Config = { mcp_timeout?: number mcp_protocol_mode?: McpProtocolMode local_subagent_placement?: boolean - local_subagent_placement_models?: Array peer_delegation?: boolean - follow_lead?: boolean unattended_permissions?: "scoped" | "full" | "off" unattended_allow?: Array + follow_lead?: boolean queue_gate?: { cwd?: string test_command?: string @@ -9465,7 +9466,7 @@ export type LoopCreateData = { noProgressLimit?: number completionToken?: string eternal?: boolean - mode?: "prompt" | "queue" + mode?: "prompt" | "queue" | "auto" queue?: Array queueGuidance?: string queueSync?: boolean @@ -9477,6 +9478,8 @@ export type LoopCreateData = { defaultBranch?: string cwd?: string } + autoDirectory?: string + autoRepos?: Array } path?: never query?: {