mirror of
https://github.com/anomalyco/opencode.git
synced 2026-07-17 12:56:41 +02:00
Compare commits
14 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 82b13ca184 | |||
| dd14413a64 | |||
| b9e7cbf13c | |||
| 3f74abc6cd | |||
| e0d0fe1ff7 | |||
| f7dbb4dac4 | |||
| c5849e56cc | |||
| e46ab34d27 | |||
| 1d4613006a | |||
| 71040c54aa | |||
| fec78154b5 | |||
| 3e2ec192cf | |||
| ec960da42a | |||
| 45de4975de |
@@ -299,7 +299,6 @@ export async function handler(
|
||||
let buffer = ""
|
||||
let responseLength = 0
|
||||
let timestampFirstByte = 0
|
||||
let timestampLastByte = 0
|
||||
|
||||
function pump(): Promise<void> {
|
||||
return (
|
||||
|
||||
@@ -78,7 +78,7 @@ const streamText = LLM.stream(request).pipe(
|
||||
Stream.tap((event) =>
|
||||
Effect.sync(() => {
|
||||
if (event.type === "text-delta") process.stdout.write(`\ntext: ${event.text}`)
|
||||
if (event.type === "request-finish") process.stdout.write(`\nfinish: ${event.reason}\n`)
|
||||
if (event.type === "finish") process.stdout.write(`\nfinish: ${event.reason}\n`)
|
||||
}),
|
||||
),
|
||||
Stream.runDrain,
|
||||
@@ -185,7 +185,7 @@ const FakeProtocol = Protocol.make<FakeBody, string, string, void>({
|
||||
event: Schema.String,
|
||||
initial: () => undefined,
|
||||
step: (_, frame) => Effect.succeed([undefined, [{ type: "text-delta", id: "text-0", text: frame }]] as const),
|
||||
onHalt: () => [{ type: "request-finish", reason: "stop" }],
|
||||
onHalt: () => [{ type: "finish", reason: "stop" }],
|
||||
},
|
||||
})
|
||||
|
||||
|
||||
@@ -17,6 +17,7 @@ export type {
|
||||
ExecutableTools,
|
||||
Tool as ToolShape,
|
||||
ToolExecute,
|
||||
ToolExecuteContext,
|
||||
Tools,
|
||||
ToolSchema,
|
||||
} from "./tool"
|
||||
|
||||
@@ -380,7 +380,7 @@ type StepResult = readonly [ParserState, ReadonlyArray<LLMEvent>]
|
||||
const NO_EVENTS: StepResult["1"] = []
|
||||
|
||||
// `response.completed` / `response.incomplete` are clean finishes that emit a
|
||||
// `request-finish` event; `response.failed` is a hard failure that emits a
|
||||
// `finish` event; `response.failed` is a hard failure that emits a
|
||||
// `provider-error`. All three end the stream — kept in one set so `step` and
|
||||
// the protocol's `terminal` predicate stay in sync.
|
||||
const TERMINAL_TYPES = new Set(["response.completed", "response.incomplete", "response.failed"])
|
||||
|
||||
@@ -80,7 +80,7 @@ export const finish = (
|
||||
usage: input.usage,
|
||||
providerMetadata: input.providerMetadata,
|
||||
}),
|
||||
LLMEvent.requestFinish(input),
|
||||
LLMEvent.finish(input),
|
||||
)
|
||||
return { ...stepped, stepStarted: false }
|
||||
}
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
import { Schema } from "effect"
|
||||
import { ContentBlockID, FinishReason, ProtocolID, ProviderMetadata, ResponseID, RouteID, ToolCallID } from "./ids"
|
||||
import { ContentBlockID, FinishReason, ProtocolID, ProviderMetadata, RouteID, ToolCallID } from "./ids"
|
||||
import { ModelRef } from "./options"
|
||||
import { ToolResultValue } from "./messages"
|
||||
|
||||
@@ -66,14 +66,13 @@ export class Usage extends Schema.Class<Usage>("LLM.Usage")({
|
||||
get visibleOutputTokens() {
|
||||
return Math.max(0, (this.outputTokens ?? 0) - (this.reasoningTokens ?? 0))
|
||||
}
|
||||
|
||||
static from(input: UsageInput) {
|
||||
return input instanceof Usage ? input : new Usage(input)
|
||||
}
|
||||
}
|
||||
|
||||
export const RequestStart = Schema.Struct({
|
||||
type: Schema.tag("request-start"),
|
||||
id: ResponseID,
|
||||
model: ModelRef,
|
||||
}).annotate({ identifier: "LLM.Event.RequestStart" })
|
||||
export type RequestStart = Schema.Schema.Type<typeof RequestStart>
|
||||
export type UsageInput = Usage | ConstructorParameters<typeof Usage>[0]
|
||||
|
||||
export const StepStart = Schema.Struct({
|
||||
type: Schema.tag("step-start"),
|
||||
@@ -185,13 +184,13 @@ export const StepFinish = Schema.Struct({
|
||||
}).annotate({ identifier: "LLM.Event.StepFinish" })
|
||||
export type StepFinish = Schema.Schema.Type<typeof StepFinish>
|
||||
|
||||
export const RequestFinish = Schema.Struct({
|
||||
type: Schema.tag("request-finish"),
|
||||
export const Finish = Schema.Struct({
|
||||
type: Schema.tag("finish"),
|
||||
reason: FinishReason,
|
||||
usage: Schema.optional(Usage),
|
||||
providerMetadata: Schema.optional(ProviderMetadata),
|
||||
}).annotate({ identifier: "LLM.Event.RequestFinish" })
|
||||
export type RequestFinish = Schema.Schema.Type<typeof RequestFinish>
|
||||
}).annotate({ identifier: "LLM.Event.Finish" })
|
||||
export type Finish = Schema.Schema.Type<typeof Finish>
|
||||
|
||||
export const ProviderErrorEvent = Schema.Struct({
|
||||
type: Schema.tag("provider-error"),
|
||||
@@ -202,7 +201,6 @@ export const ProviderErrorEvent = Schema.Struct({
|
||||
export type ProviderErrorEvent = Schema.Schema.Type<typeof ProviderErrorEvent>
|
||||
|
||||
const llmEventTagged = Schema.Union([
|
||||
RequestStart,
|
||||
StepStart,
|
||||
TextStart,
|
||||
TextDelta,
|
||||
@@ -217,13 +215,15 @@ const llmEventTagged = Schema.Union([
|
||||
ToolResult,
|
||||
ToolError,
|
||||
StepFinish,
|
||||
RequestFinish,
|
||||
Finish,
|
||||
ProviderErrorEvent,
|
||||
]).pipe(Schema.toTaggedUnion("type"))
|
||||
|
||||
type WithID<Event extends { readonly id: unknown }, ID> = Omit<Event, "type" | "id"> & { readonly id: ID | string }
|
||||
type WithUsage<Event extends { readonly usage?: Usage }> = Omit<Event, "type" | "usage"> & {
|
||||
readonly usage?: UsageInput
|
||||
}
|
||||
|
||||
const responseID = (value: ResponseID | string) => ResponseID.make(value)
|
||||
const contentBlockID = (value: ContentBlockID | string) => ContentBlockID.make(value)
|
||||
const toolCallID = (value: ToolCallID | string) => ToolCallID.make(value)
|
||||
|
||||
@@ -233,7 +233,6 @@ const toolCallID = (value: ToolCallID | string) => ToolCallID.make(value)
|
||||
* `events.filter(LLMEvent.guards["tool-call"])`.
|
||||
*/
|
||||
export const LLMEvent = Object.assign(llmEventTagged, {
|
||||
requestStart: (input: WithID<RequestStart, ResponseID>) => RequestStart.make({ ...input, id: responseID(input.id) }),
|
||||
stepStart: StepStart.make,
|
||||
textStart: (input: WithID<TextStart, ContentBlockID>) => TextStart.make({ ...input, id: contentBlockID(input.id) }),
|
||||
textDelta: (input: WithID<TextDelta, ContentBlockID>) => TextDelta.make({ ...input, id: contentBlockID(input.id) }),
|
||||
@@ -252,11 +251,18 @@ export const LLMEvent = Object.assign(llmEventTagged, {
|
||||
toolCall: (input: WithID<ToolCall, ToolCallID>) => ToolCall.make({ ...input, id: toolCallID(input.id) }),
|
||||
toolResult: (input: WithID<ToolResult, ToolCallID>) => ToolResult.make({ ...input, id: toolCallID(input.id) }),
|
||||
toolError: (input: WithID<ToolError, ToolCallID>) => ToolError.make({ ...input, id: toolCallID(input.id) }),
|
||||
stepFinish: StepFinish.make,
|
||||
requestFinish: RequestFinish.make,
|
||||
stepFinish: (input: WithUsage<StepFinish>) =>
|
||||
StepFinish.make({
|
||||
...input,
|
||||
usage: input.usage === undefined ? undefined : Usage.from(input.usage),
|
||||
}),
|
||||
finish: (input: WithUsage<Finish>) =>
|
||||
Finish.make({
|
||||
...input,
|
||||
usage: input.usage === undefined ? undefined : Usage.from(input.usage),
|
||||
}),
|
||||
providerError: ProviderErrorEvent.make,
|
||||
is: {
|
||||
requestStart: llmEventTagged.guards["request-start"],
|
||||
stepStart: llmEventTagged.guards["step-start"],
|
||||
textStart: llmEventTagged.guards["text-start"],
|
||||
textDelta: llmEventTagged.guards["text-delta"],
|
||||
@@ -271,7 +277,7 @@ export const LLMEvent = Object.assign(llmEventTagged, {
|
||||
toolResult: llmEventTagged.guards["tool-result"],
|
||||
toolError: llmEventTagged.guards["tool-error"],
|
||||
stepFinish: llmEventTagged.guards["step-finish"],
|
||||
requestFinish: llmEventTagged.guards["request-finish"],
|
||||
finish: llmEventTagged.guards.finish,
|
||||
providerError: llmEventTagged.guards["provider-error"],
|
||||
},
|
||||
})
|
||||
|
||||
@@ -12,6 +12,7 @@ import {
|
||||
ToolFailure,
|
||||
ToolResultPart,
|
||||
type ToolResultValue,
|
||||
Usage,
|
||||
} from "./schema"
|
||||
import { type AnyTool, type ExecutableTools, type Tools, toDefinitions } from "./tool"
|
||||
|
||||
@@ -72,19 +73,42 @@ export const stream = <T extends Tools>(options: StreamOptions<T>): Stream.Strea
|
||||
tools: [...options.request.tools.filter((tool) => !runtimeToolNames.has(tool.name)), ...runtimeTools],
|
||||
})
|
||||
|
||||
const loop = (request: LLMRequest, step: number): Stream.Stream<LLMEvent, LLMError> =>
|
||||
const loop = (
|
||||
request: LLMRequest,
|
||||
step: number,
|
||||
usage: Usage | undefined,
|
||||
providerMetadata: ProviderMetadata | undefined,
|
||||
): Stream.Stream<LLMEvent, LLMError> =>
|
||||
Stream.unwrap(
|
||||
Effect.gen(function* () {
|
||||
const state: StepState = { assistantContent: [], toolCalls: [], finishReason: undefined }
|
||||
const state: StepState = {
|
||||
assistantContent: [],
|
||||
toolCalls: [],
|
||||
finishReason: undefined,
|
||||
usage: undefined,
|
||||
providerMetadata: undefined,
|
||||
}
|
||||
|
||||
const modelStream = options
|
||||
.stream(request)
|
||||
.pipe(Stream.map((event) => indexStep(event, step)))
|
||||
.pipe(Stream.tap((event) => Effect.sync(() => accumulate(state, event))))
|
||||
.pipe(Stream.filter((event) => event.type !== "finish"))
|
||||
|
||||
const continuation = Stream.unwrap(
|
||||
Effect.gen(function* () {
|
||||
if (state.finishReason !== "tool-calls" || state.toolCalls.length === 0) return Stream.empty
|
||||
if (options.toolExecution === "none") return Stream.empty
|
||||
const totalUsage = addUsage(usage, state.usage)
|
||||
const totalProviderMetadata = mergeProviderMetadata(providerMetadata, state.providerMetadata)
|
||||
const finishStream = Stream.fromIterable([
|
||||
LLMEvent.finish({
|
||||
reason: state.finishReason ?? "unknown",
|
||||
usage: totalUsage,
|
||||
providerMetadata: totalProviderMetadata,
|
||||
}),
|
||||
])
|
||||
|
||||
if (state.finishReason !== "tool-calls" || state.toolCalls.length === 0) return finishStream
|
||||
if (options.toolExecution === "none") return finishStream
|
||||
|
||||
const dispatched = yield* Effect.forEach(
|
||||
state.toolCalls,
|
||||
@@ -93,10 +117,14 @@ export const stream = <T extends Tools>(options: StreamOptions<T>): Stream.Strea
|
||||
)
|
||||
const resultStream = Stream.fromIterable(dispatched.flatMap(([call, result]) => emitEvents(call, result)))
|
||||
|
||||
if (!options.stopWhen) return resultStream
|
||||
if (options.stopWhen({ step, request })) return resultStream
|
||||
if (!options.stopWhen) return resultStream.pipe(Stream.concat(finishStream))
|
||||
if (options.stopWhen({ step, request })) return resultStream.pipe(Stream.concat(finishStream))
|
||||
|
||||
return resultStream.pipe(Stream.concat(loop(followUpRequest(request, state, dispatched), step + 1)))
|
||||
return resultStream.pipe(
|
||||
Stream.concat(
|
||||
loop(followUpRequest(request, state, dispatched), step + 1, totalUsage, totalProviderMetadata),
|
||||
),
|
||||
)
|
||||
}),
|
||||
)
|
||||
|
||||
@@ -104,13 +132,21 @@ export const stream = <T extends Tools>(options: StreamOptions<T>): Stream.Strea
|
||||
}),
|
||||
)
|
||||
|
||||
return loop(initialRequest, 0)
|
||||
return loop(initialRequest, 0, undefined, undefined)
|
||||
}
|
||||
|
||||
const indexStep = (event: LLMEvent, index: number): LLMEvent => {
|
||||
if (event.type === "step-start") return LLMEvent.stepStart({ index })
|
||||
if (event.type === "step-finish") return LLMEvent.stepFinish({ ...event, index })
|
||||
return event
|
||||
}
|
||||
|
||||
interface StepState {
|
||||
assistantContent: ContentPart[]
|
||||
toolCalls: ToolCallPart[]
|
||||
finishReason: FinishReason | undefined
|
||||
usage: Usage | undefined
|
||||
providerMetadata: ProviderMetadata | undefined
|
||||
}
|
||||
|
||||
const accumulate = (state: StepState, event: LLMEvent) => {
|
||||
@@ -154,9 +190,43 @@ const accumulate = (state: StepState, event: LLMEvent) => {
|
||||
)
|
||||
return
|
||||
}
|
||||
if (event.type === "step-finish" || event.type === "request-finish") {
|
||||
if (event.type === "step-finish") {
|
||||
state.finishReason = event.reason === "stop" && state.toolCalls.length > 0 ? "tool-calls" : event.reason
|
||||
state.usage = addUsage(state.usage, event.usage)
|
||||
state.providerMetadata = mergeProviderMetadata(state.providerMetadata, event.providerMetadata)
|
||||
return
|
||||
}
|
||||
if (event.type === "finish") {
|
||||
state.finishReason ??= event.reason
|
||||
state.usage ??= event.usage
|
||||
state.providerMetadata = mergeProviderMetadata(state.providerMetadata, event.providerMetadata)
|
||||
}
|
||||
}
|
||||
|
||||
const addUsage = (left: Usage | undefined, right: Usage | undefined) => {
|
||||
if (!left) return right
|
||||
if (!right) return left
|
||||
type UsageKey =
|
||||
| "inputTokens"
|
||||
| "outputTokens"
|
||||
| "nonCachedInputTokens"
|
||||
| "cacheReadInputTokens"
|
||||
| "cacheWriteInputTokens"
|
||||
| "reasoningTokens"
|
||||
| "totalTokens"
|
||||
const sum = (key: UsageKey) =>
|
||||
left[key] === undefined && right[key] === undefined ? undefined : Number(left[key] ?? 0) + Number(right[key] ?? 0)
|
||||
|
||||
return new Usage({
|
||||
inputTokens: sum("inputTokens"),
|
||||
outputTokens: sum("outputTokens"),
|
||||
nonCachedInputTokens: sum("nonCachedInputTokens"),
|
||||
cacheReadInputTokens: sum("cacheReadInputTokens"),
|
||||
cacheWriteInputTokens: sum("cacheWriteInputTokens"),
|
||||
reasoningTokens: sum("reasoningTokens"),
|
||||
totalTokens: sum("totalTokens"),
|
||||
providerMetadata: mergeProviderMetadata(left.providerMetadata, right.providerMetadata),
|
||||
})
|
||||
}
|
||||
|
||||
const sameProviderMetadata = (left: ProviderMetadata | undefined, right: ProviderMetadata | undefined) =>
|
||||
@@ -200,17 +270,17 @@ const dispatch = (tools: Tools, call: ToolCallPart): Effect.Effect<ToolResultVal
|
||||
if (!tool.execute)
|
||||
return Effect.succeed({ type: "error" as const, value: `Tool has no execute handler: ${call.name}` })
|
||||
|
||||
return decodeAndExecute(tool, call.input).pipe(
|
||||
return decodeAndExecute(tool, call).pipe(
|
||||
Effect.catchTag("LLM.ToolFailure", (failure) =>
|
||||
Effect.succeed({ type: "error" as const, value: failure.message } satisfies ToolResultValue),
|
||||
),
|
||||
)
|
||||
}
|
||||
|
||||
const decodeAndExecute = (tool: AnyTool, input: unknown): Effect.Effect<ToolResultValue, ToolFailure> =>
|
||||
tool._decode(input).pipe(
|
||||
const decodeAndExecute = (tool: AnyTool, call: ToolCallPart): Effect.Effect<ToolResultValue, ToolFailure> =>
|
||||
tool._decode(call.input).pipe(
|
||||
Effect.mapError((error) => new ToolFailure({ message: `Invalid tool input: ${error.message}` })),
|
||||
Effect.flatMap((decoded) => tool.execute!(decoded)),
|
||||
Effect.flatMap((decoded) => tool.execute!(decoded, { id: call.id, name: call.name })),
|
||||
Effect.flatMap((value) =>
|
||||
tool._encode(value).pipe(
|
||||
Effect.mapError(
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
import { Effect, JsonSchema, Schema } from "effect"
|
||||
import type { ToolDefinition as ToolDefinitionClass } from "./schema"
|
||||
import type { ToolCallPart, ToolDefinition as ToolDefinitionClass } from "./schema"
|
||||
import { ToolDefinition, ToolFailure } from "./schema"
|
||||
|
||||
/**
|
||||
@@ -8,9 +8,14 @@ import { ToolDefinition, ToolFailure } from "./schema"
|
||||
* beyond pure data conversion belongs in the handler closure.
|
||||
*/
|
||||
export type ToolSchema<T> = Schema.Codec<T, any, never, never>
|
||||
export interface ToolExecuteContext {
|
||||
readonly id: ToolCallPart["id"]
|
||||
readonly name: ToolCallPart["name"]
|
||||
}
|
||||
|
||||
export type ToolExecute<Parameters extends ToolSchema<any>, Success extends ToolSchema<any>> = (
|
||||
params: Schema.Schema.Type<Parameters>,
|
||||
context?: ToolExecuteContext,
|
||||
) => Effect.Effect<Schema.Schema.Type<Success>, ToolFailure>
|
||||
|
||||
/**
|
||||
@@ -61,7 +66,7 @@ type TypedToolConfig = {
|
||||
type DynamicToolConfig = {
|
||||
readonly description: string
|
||||
readonly jsonSchema: JsonSchema.JsonSchema
|
||||
readonly execute?: (params: unknown) => Effect.Effect<unknown, ToolFailure>
|
||||
readonly execute?: (params: unknown, context?: ToolExecuteContext) => Effect.Effect<unknown, ToolFailure>
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -110,7 +115,7 @@ export function make<Parameters extends ToolSchema<any>, Success extends ToolSch
|
||||
export function make(config: {
|
||||
readonly description: string
|
||||
readonly jsonSchema: JsonSchema.JsonSchema
|
||||
readonly execute: (params: unknown) => Effect.Effect<unknown, ToolFailure>
|
||||
readonly execute: (params: unknown, context?: ToolExecuteContext) => Effect.Effect<unknown, ToolFailure>
|
||||
}): AnyExecutableTool
|
||||
export function make(config: {
|
||||
readonly description: string
|
||||
|
||||
@@ -51,7 +51,7 @@ const request = LLM.request({
|
||||
|
||||
const raiseEvent = (event: FakeEvent): import("../src/schema").LLMEvent =>
|
||||
event.type === "finish"
|
||||
? { type: "request-finish", reason: event.reason }
|
||||
? { type: "finish", reason: event.reason }
|
||||
: { type: "text-delta", id: "text-0", text: event.text }
|
||||
|
||||
const fakeProtocol = Protocol.make<FakeBody, FakeEvent, FakeEvent, void>({
|
||||
@@ -112,8 +112,8 @@ describe("llm route", () => {
|
||||
const events = Array.from(yield* llm.stream(request).pipe(Stream.runCollect))
|
||||
const response = yield* llm.generate(request)
|
||||
|
||||
expect(events.map((event) => event.type)).toEqual(["text-delta", "request-finish"])
|
||||
expect(response.events.map((event) => event.type)).toEqual(["text-delta", "request-finish"])
|
||||
expect(events.map((event) => event.type)).toEqual(["text-delta", "finish"])
|
||||
expect(response.events.map((event) => event.type)).toEqual(["text-delta", "finish"])
|
||||
}),
|
||||
)
|
||||
|
||||
|
||||
@@ -127,7 +127,7 @@ describe("llm constructors", () => {
|
||||
LLMResponse.text({
|
||||
events: [
|
||||
{ type: "text-delta", id: "text-0", text: "hi" },
|
||||
{ type: "request-finish", reason: "stop" },
|
||||
{ type: "finish", reason: "stop" },
|
||||
],
|
||||
}),
|
||||
).toBe("hi")
|
||||
|
||||
@@ -124,7 +124,7 @@ describe("Anthropic Messages route", () => {
|
||||
providerMetadata: { anthropic: { signature: "sig_1" } },
|
||||
})
|
||||
expect(response.events.at(-1)).toMatchObject({
|
||||
type: "request-finish",
|
||||
type: "finish",
|
||||
reason: "stop",
|
||||
providerMetadata: { anthropic: { stopSequence: "\n\nHuman:" } },
|
||||
})
|
||||
@@ -182,7 +182,7 @@ describe("Anthropic Messages route", () => {
|
||||
},
|
||||
{ type: "step-finish", index: 0, reason: "tool-calls", usage, providerMetadata: undefined },
|
||||
{
|
||||
type: "request-finish",
|
||||
type: "finish",
|
||||
reason: "tool-calls",
|
||||
providerMetadata: undefined,
|
||||
usage,
|
||||
@@ -275,7 +275,7 @@ describe("Anthropic Messages route", () => {
|
||||
providerMetadata: { anthropic: { blockType: "web_search_tool_result" } },
|
||||
})
|
||||
expect(response.text).toBe("Found it.")
|
||||
expect(response.events.at(-1)).toMatchObject({ type: "request-finish", reason: "stop" })
|
||||
expect(response.events.at(-1)).toMatchObject({ type: "finish", reason: "stop" })
|
||||
}),
|
||||
)
|
||||
|
||||
|
||||
@@ -169,12 +169,12 @@ describe("Bedrock Converse route", () => {
|
||||
const response = yield* LLMClient.generate(baseRequest).pipe(Effect.provide(fixedBytes(body)))
|
||||
|
||||
expect(response.text).toBe("Hello!")
|
||||
const finishes = response.events.filter((event) => event.type === "request-finish")
|
||||
const finishes = response.events.filter((event) => event.type === "finish")
|
||||
// Bedrock splits the finish across `messageStop` (carries reason) and
|
||||
// `metadata` (carries usage). We consolidate them into a single
|
||||
// terminal `request-finish` event with both.
|
||||
// terminal `finish` event with both.
|
||||
expect(finishes).toHaveLength(1)
|
||||
expect(finishes[0]).toMatchObject({ type: "request-finish", reason: "stop" })
|
||||
expect(finishes[0]).toMatchObject({ type: "finish", reason: "stop" })
|
||||
expect(response.usage).toMatchObject({
|
||||
inputTokens: 5,
|
||||
outputTokens: 2,
|
||||
@@ -213,7 +213,7 @@ describe("Bedrock Converse route", () => {
|
||||
{ type: "tool-input-delta", id: "tool_1", name: "lookup", text: '{"query"' },
|
||||
{ type: "tool-input-delta", id: "tool_1", name: "lookup", text: ':"weather"}' },
|
||||
])
|
||||
expect(response.events.at(-1)).toMatchObject({ type: "request-finish", reason: "tool-calls" })
|
||||
expect(response.events.at(-1)).toMatchObject({ type: "finish", reason: "tool-calls" })
|
||||
}),
|
||||
)
|
||||
|
||||
|
||||
@@ -232,7 +232,7 @@ describe("Gemini route", () => {
|
||||
{ type: "text-end", id: "text-0" },
|
||||
{ type: "step-finish", index: 0, reason: "stop", usage, providerMetadata: undefined },
|
||||
{
|
||||
type: "request-finish",
|
||||
type: "finish",
|
||||
reason: "stop",
|
||||
usage,
|
||||
},
|
||||
@@ -291,7 +291,7 @@ describe("Gemini route", () => {
|
||||
},
|
||||
{ type: "step-finish", index: 0, reason: "tool-calls", usage, providerMetadata: undefined },
|
||||
{
|
||||
type: "request-finish",
|
||||
type: "finish",
|
||||
reason: "tool-calls",
|
||||
usage,
|
||||
},
|
||||
@@ -325,7 +325,7 @@ describe("Gemini route", () => {
|
||||
{ type: "tool-call", id: "tool_0", name: "lookup", input: { query: "weather" } },
|
||||
{ type: "tool-call", id: "tool_1", name: "lookup", input: { query: "news" } },
|
||||
])
|
||||
expect(response.events.at(-1)).toMatchObject({ type: "request-finish", reason: "tool-calls" })
|
||||
expect(response.events.at(-1)).toMatchObject({ type: "finish", reason: "tool-calls" })
|
||||
}),
|
||||
)
|
||||
|
||||
@@ -344,10 +344,10 @@ describe("Gemini route", () => {
|
||||
),
|
||||
)
|
||||
|
||||
expect(length.events.map((event) => event.type)).toEqual(["step-start", "step-finish", "request-finish"])
|
||||
expect(length.events.at(-1)).toMatchObject({ type: "request-finish", reason: "length" })
|
||||
expect(filtered.events.map((event) => event.type)).toEqual(["step-start", "step-finish", "request-finish"])
|
||||
expect(filtered.events.at(-1)).toMatchObject({ type: "request-finish", reason: "content-filter" })
|
||||
expect(length.events.map((event) => event.type)).toEqual(["step-start", "step-finish", "finish"])
|
||||
expect(length.events.at(-1)).toMatchObject({ type: "finish", reason: "length" })
|
||||
expect(filtered.events.map((event) => event.type)).toEqual(["step-start", "step-finish", "finish"])
|
||||
expect(filtered.events.at(-1)).toMatchObject({ type: "finish", reason: "content-filter" })
|
||||
}),
|
||||
)
|
||||
|
||||
|
||||
@@ -249,7 +249,7 @@ describe("OpenAI Chat route", () => {
|
||||
{ type: "text-end", id: "text-0" },
|
||||
{ type: "step-finish", index: 0, reason: "stop", usage, providerMetadata: undefined },
|
||||
{
|
||||
type: "request-finish",
|
||||
type: "finish",
|
||||
reason: "stop",
|
||||
usage,
|
||||
},
|
||||
@@ -288,7 +288,7 @@ describe("OpenAI Chat route", () => {
|
||||
providerMetadata: undefined,
|
||||
},
|
||||
{ type: "step-finish", index: 0, reason: "tool-calls", usage: undefined, providerMetadata: undefined },
|
||||
{ type: "request-finish", reason: "tool-calls", usage: undefined },
|
||||
{ type: "finish", reason: "tool-calls", usage: undefined },
|
||||
])
|
||||
}),
|
||||
)
|
||||
|
||||
@@ -231,7 +231,7 @@ describe("OpenAI-compatible Chat route", () => {
|
||||
|
||||
expect(response.text).toBe("Hello!")
|
||||
expect(response.usage).toMatchObject({ inputTokens: 5, outputTokens: 2, totalTokens: 7 })
|
||||
expect(response.events.at(-1)).toMatchObject({ type: "request-finish", reason: "stop" })
|
||||
expect(response.events.at(-1)).toMatchObject({ type: "finish", reason: "stop" })
|
||||
}),
|
||||
)
|
||||
})
|
||||
|
||||
@@ -366,7 +366,7 @@ describe("OpenAI Responses route", () => {
|
||||
usage,
|
||||
},
|
||||
{
|
||||
type: "request-finish",
|
||||
type: "finish",
|
||||
reason: "stop",
|
||||
providerMetadata: { openai: { responseId: "resp_1", serviceTier: "default" } },
|
||||
usage,
|
||||
@@ -447,7 +447,7 @@ describe("OpenAI Responses route", () => {
|
||||
},
|
||||
{ type: "step-finish", index: 0, reason: "tool-calls", usage, providerMetadata: undefined },
|
||||
{
|
||||
type: "request-finish",
|
||||
type: "finish",
|
||||
reason: "tool-calls",
|
||||
providerMetadata: undefined,
|
||||
usage,
|
||||
|
||||
@@ -120,8 +120,8 @@ export const runWeatherToolLoop = (request: LLMRequest) =>
|
||||
|
||||
export const expectFinish = (
|
||||
events: ReadonlyArray<LLMEvent>,
|
||||
reason: Extract<LLMEvent, { readonly type: "request-finish" }>["reason"],
|
||||
) => expect(events.at(-1)).toMatchObject({ type: "request-finish", reason })
|
||||
reason: Extract<LLMEvent, { readonly type: "finish" }>["reason"],
|
||||
) => expect(events.at(-1)).toMatchObject({ type: "finish", reason })
|
||||
|
||||
export const expectWeatherToolCall = (response: LLMResponse) =>
|
||||
expect(response.toolCalls).toMatchObject([
|
||||
@@ -129,10 +129,12 @@ export const expectWeatherToolCall = (response: LLMResponse) =>
|
||||
])
|
||||
|
||||
export const expectWeatherToolLoop = (events: ReadonlyArray<LLMEvent>) => {
|
||||
const finishes = events.filter(LLMEvent.is.requestFinish)
|
||||
expect(finishes).toHaveLength(2)
|
||||
expect(finishes[0]?.reason).toBe("tool-calls")
|
||||
expect(finishes.at(-1)?.reason).toBe("stop")
|
||||
const finishes = events.filter(LLMEvent.is.finish)
|
||||
expect(finishes).toHaveLength(1)
|
||||
expect(finishes[0]?.reason).toBe("stop")
|
||||
|
||||
const stepFinishes = events.filter(LLMEvent.is.stepFinish)
|
||||
expect(stepFinishes.map((event) => event.reason)).toEqual(["tool-calls", "stop"])
|
||||
|
||||
const toolCalls = events.filter(LLMEvent.is.toolCall)
|
||||
expect(toolCalls).toHaveLength(1)
|
||||
@@ -272,7 +274,7 @@ export const eventSummary = (events: ReadonlyArray<LLMEvent>) => {
|
||||
summary.push({ type: "tool-error", name: event.name, message: event.message })
|
||||
continue
|
||||
}
|
||||
if (event.type === "request-finish") {
|
||||
if (event.type === "finish") {
|
||||
summary.push({ type: "finish", reason: event.reason, usage: usageSummary(event.usage) })
|
||||
}
|
||||
}
|
||||
|
||||
@@ -44,6 +44,11 @@ describe("llm schema", () => {
|
||||
expect(() => Schema.decodeUnknownSync(LLMEvent)({ type: "bogus" })).toThrow()
|
||||
})
|
||||
|
||||
test("finish constructors accept usage input", () => {
|
||||
expect(LLMEvent.stepFinish({ index: 0, reason: "stop", usage: { inputTokens: 1 } }).usage).toBeInstanceOf(Usage)
|
||||
expect(LLMEvent.finish({ reason: "stop", usage: { outputTokens: 2 } }).usage).toBeInstanceOf(Usage)
|
||||
})
|
||||
|
||||
test("content part tagged union exposes guards", () => {
|
||||
expect(ContentPart.guards.text({ type: "text", text: "hi" })).toBe(true)
|
||||
expect(ContentPart.guards.media({ type: "text", text: "hi" })).toBe(false)
|
||||
|
||||
@@ -4,7 +4,8 @@ import { GenerationOptions, LLM, LLMEvent, LLMRequest, LLMResponse, ToolChoice }
|
||||
import { LLMClient } from "../src/route"
|
||||
import * as AnthropicMessages from "../src/protocols/anthropic-messages"
|
||||
import * as OpenAIChat from "../src/protocols/openai-chat"
|
||||
import { tool, ToolFailure } from "../src/tool"
|
||||
import { tool, ToolFailure, type ToolExecuteContext } from "../src/tool"
|
||||
import { ToolRuntime } from "../src/tool-runtime"
|
||||
import { it } from "./lib/effect"
|
||||
import * as TestToolRuntime from "./lib/tool-runtime"
|
||||
import { dynamicResponse, scriptedResponses } from "./lib/http"
|
||||
@@ -129,7 +130,7 @@ describe("LLMClient tools", () => {
|
||||
name: "get_weather",
|
||||
result: { type: "json", value: { temperature: 22, condition: "sunny" } },
|
||||
})
|
||||
expect(events.at(-1)?.type).toBe("request-finish")
|
||||
expect(events.at(-1)?.type).toBe("finish")
|
||||
expect(LLMResponse.text({ events })).toBe("It's sunny in Paris.")
|
||||
}),
|
||||
)
|
||||
@@ -148,11 +149,40 @@ describe("LLMClient tools", () => {
|
||||
),
|
||||
)
|
||||
|
||||
expect(events.filter(LLMEvent.is.requestFinish)).toHaveLength(1)
|
||||
expect(events.filter(LLMEvent.is.finish)).toHaveLength(1)
|
||||
expect(events.find(LLMEvent.is.toolResult)).toMatchObject({ type: "tool-result", id: "call_1" })
|
||||
}),
|
||||
)
|
||||
|
||||
it.effect("passes tool call context to execute", () =>
|
||||
Effect.gen(function* () {
|
||||
let context: ToolExecuteContext | undefined
|
||||
const contextual = tool({
|
||||
description: "Capture tool context.",
|
||||
parameters: Schema.Struct({ value: Schema.String }),
|
||||
success: Schema.Struct({ ok: Schema.Boolean }),
|
||||
execute: (_params, ctx) =>
|
||||
Effect.sync(() => {
|
||||
context = ctx
|
||||
return { ok: true }
|
||||
}),
|
||||
})
|
||||
const events = Array.from(
|
||||
yield* TestToolRuntime.runTools({ request: baseRequest, tools: { contextual } }).pipe(
|
||||
Stream.runCollect,
|
||||
Effect.provide(
|
||||
scriptedResponses([
|
||||
sseEvents(toolCallChunk("call_ctx", "contextual", '{"value":"x"}'), finishChunk("tool_calls")),
|
||||
]),
|
||||
),
|
||||
),
|
||||
)
|
||||
|
||||
expect(events.some(LLMEvent.is.toolResult)).toBe(true)
|
||||
expect(context).toEqual({ id: "call_ctx", name: "contextual" })
|
||||
}),
|
||||
)
|
||||
|
||||
it.effect("can expose tool schemas without executing tool calls", () =>
|
||||
Effect.gen(function* () {
|
||||
const layer = scriptedResponses([
|
||||
@@ -319,7 +349,7 @@ describe("LLMClient tools", () => {
|
||||
"text-delta",
|
||||
"text-end",
|
||||
"step-finish",
|
||||
"request-finish",
|
||||
"finish",
|
||||
])
|
||||
expect(LLMResponse.text({ events })).toBe("Done.")
|
||||
}),
|
||||
@@ -343,7 +373,57 @@ describe("LLMClient tools", () => {
|
||||
),
|
||||
)
|
||||
|
||||
expect(events.filter(LLMEvent.is.requestFinish)).toHaveLength(2)
|
||||
expect(events.filter(LLMEvent.is.finish)).toHaveLength(1)
|
||||
expect(events.filter(LLMEvent.is.stepStart).map((event) => event.index)).toEqual([0, 1])
|
||||
expect(events.filter(LLMEvent.is.stepFinish).map((event) => event.index)).toEqual([0, 1])
|
||||
}),
|
||||
)
|
||||
|
||||
it.effect("emits one final finish with aggregate usage", () =>
|
||||
Effect.gen(function* () {
|
||||
let calls = 0
|
||||
const events = Array.from(
|
||||
yield* ToolRuntime.stream({
|
||||
request: baseRequest,
|
||||
tools: { get_weather },
|
||||
stopWhen: ToolRuntime.stepCountIs(2),
|
||||
stream: () =>
|
||||
Stream.fromIterable<LLMEvent>(
|
||||
calls++ === 0
|
||||
? [
|
||||
LLMEvent.stepStart({ index: 0 }),
|
||||
LLMEvent.toolCall({ id: "call_1", name: "get_weather", input: { city: "Paris" } }),
|
||||
LLMEvent.stepFinish({
|
||||
index: 0,
|
||||
reason: "tool-calls",
|
||||
usage: { inputTokens: 1, outputTokens: 2, totalTokens: 3 },
|
||||
}),
|
||||
LLMEvent.finish({
|
||||
reason: "tool-calls",
|
||||
usage: { inputTokens: 1, outputTokens: 2, totalTokens: 3 },
|
||||
}),
|
||||
]
|
||||
: [
|
||||
LLMEvent.stepStart({ index: 0 }),
|
||||
LLMEvent.textDelta({ id: "text_1", text: "Done." }),
|
||||
LLMEvent.stepFinish({
|
||||
index: 0,
|
||||
reason: "stop",
|
||||
usage: { inputTokens: 4, outputTokens: 5, totalTokens: 9 },
|
||||
}),
|
||||
LLMEvent.finish({ reason: "stop", usage: { inputTokens: 4, outputTokens: 5, totalTokens: 9 } }),
|
||||
],
|
||||
),
|
||||
}).pipe(Stream.runCollect),
|
||||
)
|
||||
|
||||
expect(events.filter(LLMEvent.is.stepFinish).map((event) => event.index)).toEqual([0, 1])
|
||||
expect(events.filter(LLMEvent.is.finish)).toHaveLength(1)
|
||||
expect(events.find(LLMEvent.is.finish)?.usage).toMatchObject({
|
||||
inputTokens: 5,
|
||||
outputTokens: 7,
|
||||
totalTokens: 12,
|
||||
})
|
||||
}),
|
||||
)
|
||||
|
||||
@@ -362,7 +442,7 @@ describe("LLMClient tools", () => {
|
||||
}).pipe(Stream.runCollect, Effect.provide(layer)),
|
||||
)
|
||||
|
||||
expect(events.filter(LLMEvent.is.requestFinish)).toHaveLength(1)
|
||||
expect(events.filter(LLMEvent.is.finish)).toHaveLength(1)
|
||||
expect(events.find(LLMEvent.is.toolResult)).toMatchObject({ type: "tool-result", id: "call_1" })
|
||||
}),
|
||||
)
|
||||
|
||||
@@ -1094,8 +1094,8 @@ export class Agent implements ACPAgent {
|
||||
|
||||
const currentModeId = await (async () => {
|
||||
if (!availableModes.length) return undefined
|
||||
const defaultAgentName = await AppRuntime.runPromise(AgentModule.Service.use((svc) => svc.defaultAgent()))
|
||||
const resolvedModeId = availableModes.find((mode) => mode.name === defaultAgentName)?.id ?? availableModes[0].id
|
||||
const defaultAgent = await AppRuntime.runPromise(AgentModule.Service.use((svc) => svc.defaultInfo()))
|
||||
const resolvedModeId = availableModes.find((mode) => mode.name === defaultAgent.name)?.id ?? availableModes[0].id
|
||||
this.sessionManager.setMode(sessionId, resolvedModeId)
|
||||
return resolvedModeId
|
||||
})()
|
||||
@@ -1328,7 +1328,8 @@ export class Agent implements ACPAgent {
|
||||
if (!current) {
|
||||
this.sessionManager.setModel(session.id, model)
|
||||
}
|
||||
const agent = session.modeId ?? (await AppRuntime.runPromise(AgentModule.Service.use((svc) => svc.defaultAgent())))
|
||||
const agent =
|
||||
session.modeId ?? (await AppRuntime.runPromise(AgentModule.Service.use((svc) => svc.defaultInfo()))).name
|
||||
|
||||
const parts: Array<
|
||||
| { type: "text"; text: string; synthetic?: boolean; ignored?: boolean }
|
||||
|
||||
@@ -57,6 +57,7 @@ const GeneratedAgent = Schema.Struct({
|
||||
export interface Interface {
|
||||
readonly get: (agent: string) => Effect.Effect<Info>
|
||||
readonly list: () => Effect.Effect<Info[]>
|
||||
readonly defaultInfo: () => Effect.Effect<Info>
|
||||
readonly defaultAgent: () => Effect.Effect<string>
|
||||
readonly generate: (input: {
|
||||
description: string
|
||||
@@ -333,23 +334,28 @@ export const layer = Layer.effect(
|
||||
)
|
||||
})
|
||||
|
||||
const defaultAgent = Effect.fnUntraced(function* () {
|
||||
const defaultInfo = Effect.fnUntraced(function* () {
|
||||
const c = yield* config.get()
|
||||
if (c.default_agent) {
|
||||
const agent = agents[c.default_agent]
|
||||
if (!agent) throw new Error(`default agent "${c.default_agent}" not found`)
|
||||
if (agent.mode === "subagent") throw new Error(`default agent "${c.default_agent}" is a subagent`)
|
||||
if (agent.hidden === true) throw new Error(`default agent "${c.default_agent}" is hidden`)
|
||||
return agent.name
|
||||
return agent
|
||||
}
|
||||
const visible = Object.values(agents).find((a) => a.mode !== "subagent" && a.hidden !== true)
|
||||
if (!visible) throw new Error("no primary visible agent found")
|
||||
return visible.name
|
||||
return visible
|
||||
})
|
||||
|
||||
const defaultAgent = Effect.fnUntraced(function* () {
|
||||
return (yield* defaultInfo()).name
|
||||
})
|
||||
|
||||
return {
|
||||
get,
|
||||
list,
|
||||
defaultInfo,
|
||||
defaultAgent,
|
||||
} satisfies State
|
||||
}),
|
||||
@@ -362,6 +368,9 @@ export const layer = Layer.effect(
|
||||
list: Effect.fn("Agent.list")(function* () {
|
||||
return yield* InstanceState.useEffect(state, (s) => s.list())
|
||||
}),
|
||||
defaultInfo: Effect.fn("Agent.defaultInfo")(function* () {
|
||||
return yield* InstanceState.useEffect(state, (s) => s.defaultInfo())
|
||||
}),
|
||||
defaultAgent: Effect.fn("Agent.defaultAgent")(function* () {
|
||||
return yield* InstanceState.useEffect(state, (s) => s.defaultAgent())
|
||||
}),
|
||||
|
||||
@@ -29,14 +29,14 @@ export const PrCommand = effectCmd({
|
||||
UI.println(`Fetching and checking out PR #${prNumber}...`)
|
||||
|
||||
const checkout = yield* Effect.promise(() =>
|
||||
Process.run(["gh", "pr", "checkout", `${prNumber}`, "--branch", localBranchName, "--force"], { nothrow: true }),
|
||||
Process.runPromise(["gh", "pr", "checkout", `${prNumber}`, "--branch", localBranchName, "--force"], { nothrow: true }),
|
||||
)
|
||||
if (checkout.code !== 0) {
|
||||
return yield* fail(`Failed to checkout PR #${prNumber}. Make sure you have gh CLI installed and authenticated.`)
|
||||
}
|
||||
|
||||
const prInfoResult = yield* Effect.promise(() =>
|
||||
Process.text(
|
||||
Process.textPromise(
|
||||
[
|
||||
"gh",
|
||||
"pr",
|
||||
@@ -80,7 +80,7 @@ export const PrCommand = effectCmd({
|
||||
UI.println(`Importing session...`)
|
||||
|
||||
const importResult = yield* Effect.promise(() =>
|
||||
Process.text(["opencode", "import", sessionUrl], { nothrow: true }),
|
||||
Process.textPromise(["opencode", "import", sessionUrl], { nothrow: true }),
|
||||
)
|
||||
if (importResult.code === 0) {
|
||||
const sessionIdMatch = importResult.text.trim().match(/Imported session: ([a-zA-Z0-9_-]+)/)
|
||||
|
||||
@@ -50,7 +50,7 @@ export async function read(): Promise<Content | undefined> {
|
||||
if (os === "darwin") {
|
||||
const tmpfile = path.join(tmpdir(), "opencode-clipboard.png")
|
||||
try {
|
||||
await Process.run(
|
||||
await Process.runPromise(
|
||||
[
|
||||
"osascript",
|
||||
"-e",
|
||||
@@ -79,7 +79,7 @@ export async function read(): Promise<Content | undefined> {
|
||||
if (os === "win32" || release().includes("WSL")) {
|
||||
const script =
|
||||
"Add-Type -AssemblyName System.Windows.Forms; $img = [System.Windows.Forms.Clipboard]::GetImage(); if ($img) { $ms = New-Object System.IO.MemoryStream; $img.Save($ms, [System.Drawing.Imaging.ImageFormat]::Png); [System.Convert]::ToBase64String($ms.ToArray()) }"
|
||||
const base64 = await Process.text(["powershell.exe", "-NonInteractive", "-NoProfile", "-command", script], {
|
||||
const base64 = await Process.textPromise(["powershell.exe", "-NonInteractive", "-NoProfile", "-command", script], {
|
||||
nothrow: true,
|
||||
})
|
||||
if (base64.text) {
|
||||
@@ -91,11 +91,11 @@ export async function read(): Promise<Content | undefined> {
|
||||
}
|
||||
|
||||
if (os === "linux") {
|
||||
const wayland = await Process.run(["wl-paste", "-t", "image/png"], { nothrow: true })
|
||||
const wayland = await Process.runPromise(["wl-paste", "-t", "image/png"], { nothrow: true })
|
||||
if (wayland.stdout.byteLength > 0) {
|
||||
return { data: Buffer.from(wayland.stdout).toString("base64"), mime: "image/png" }
|
||||
}
|
||||
const x11 = await Process.run(["xclip", "-selection", "clipboard", "-t", "image/png", "-o"], {
|
||||
const x11 = await Process.runPromise(["xclip", "-selection", "clipboard", "-t", "image/png", "-o"], {
|
||||
nothrow: true,
|
||||
})
|
||||
if (x11.stdout.byteLength > 0) {
|
||||
@@ -118,7 +118,7 @@ const getCopyMethod = lazy(async () => {
|
||||
console.log("clipboard: using osascript")
|
||||
return async (text: string) => {
|
||||
const escaped = text.replace(/\\/g, "\\\\").replace(/"/g, '\\"')
|
||||
await Process.run(["osascript", "-e", `set the clipboard to "${escaped}"`], { nothrow: true })
|
||||
await Process.runPromise(["osascript", "-e", `set the clipboard to "${escaped}"`], { nothrow: true })
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -192,7 +192,7 @@ async function executeUninstall(method: Installation.Method, targets: RemovalTar
|
||||
const cmd = cmds[method]
|
||||
if (cmd) {
|
||||
spinner.start(`Running ${cmd.join(" ")}...`)
|
||||
const result = await Process.run(method === "choco" ? ["choco", "uninstall", "opencode", "-y", "-r"] : cmd, {
|
||||
const result = await Process.runPromise(method === "choco" ? ["choco", "uninstall", "opencode", "-y", "-r"] : cmd, {
|
||||
nothrow: true,
|
||||
})
|
||||
if (result.code !== 0) {
|
||||
|
||||
@@ -56,7 +56,7 @@ export async function readManagedPreferences() {
|
||||
for (const plist of paths) {
|
||||
if (!existsSync(plist)) continue
|
||||
log.info("reading macOS managed preferences", { path: plist })
|
||||
const result = await Process.run(["plutil", "-convert", "json", "-o", "-", plist], { nothrow: true })
|
||||
const result = await Process.runPromise(["plutil", "-convert", "json", "-o", "-", plist], { nothrow: true })
|
||||
if (result.code !== 0) {
|
||||
log.warn("failed to convert managed preferences plist", { path: plist })
|
||||
continue
|
||||
|
||||
@@ -221,7 +221,7 @@ export const rlang: Info = {
|
||||
const air = which("air")
|
||||
if (air == null) return false
|
||||
|
||||
const output = await Process.text([air, "--help"], { nothrow: true })
|
||||
const output = await Process.textPromise([air, "--help"], { nothrow: true })
|
||||
|
||||
// Check for "Air: An R language server and formatter"
|
||||
const firstLine = output.text.split("\n")[0]
|
||||
@@ -239,7 +239,7 @@ export const uvformat: Info = {
|
||||
if (await ruff.enabled(context)) return false
|
||||
const uv = which("uv")
|
||||
if (uv == null) return false
|
||||
const output = await Process.run([uv, "format", "--help"], { nothrow: true })
|
||||
const output = await Process.runPromise([uv, "format", "--help"], { nothrow: true })
|
||||
if (output.code === 0) return [uv, "format", "--", "$FILE"]
|
||||
return false
|
||||
},
|
||||
|
||||
@@ -47,7 +47,7 @@ export async function install(ide: (typeof SUPPORTED_IDES)[number]["name"]) {
|
||||
const cmd = SUPPORTED_IDES.find((i) => i.name === ide)?.cmd
|
||||
if (!cmd) throw new Error(`Unknown IDE: ${ide}`)
|
||||
|
||||
const p = await Process.run([cmd, "--install-extension", "sst-dev.opencode"], {
|
||||
const p = await Process.runPromise([cmd, "--install-extension", "sst-dev.opencode"], {
|
||||
nothrow: true,
|
||||
})
|
||||
const stdout = p.stdout.toString()
|
||||
|
||||
@@ -5,6 +5,9 @@ import { Global } from "@opencode-ai/core/global"
|
||||
import * as Log from "@opencode-ai/core/util/log"
|
||||
import { text } from "node:stream/consumers"
|
||||
import fs from "fs/promises"
|
||||
import { Effect, ManagedRuntime } from "effect"
|
||||
import { CrossSpawnSpawner } from "@opencode-ai/core/cross-spawn-spawner"
|
||||
import { memoMap } from "@opencode-ai/core/effect/memo-map"
|
||||
import { Filesystem } from "@/util/filesystem"
|
||||
import type { InstanceContext } from "../project/instance"
|
||||
import { Flag } from "@opencode-ai/core/flag/flag"
|
||||
@@ -21,8 +24,17 @@ const pathExists = async (p: string) =>
|
||||
.stat(p)
|
||||
.then(() => true)
|
||||
.catch(() => false)
|
||||
const run = (cmd: string[], opts: Process.RunOptions = {}) => Process.run(cmd, { ...opts, nothrow: true })
|
||||
const output = (cmd: string[], opts: Process.RunOptions = {}) => Process.text(cmd, { ...opts, nothrow: true })
|
||||
|
||||
// Private runtime so the Effect-returning `Process.run` / `Process.text` can be
|
||||
// invoked from this file's promise-based spawn callbacks. The `LSP` service
|
||||
// layer in `lsp.ts` calls these spawn functions inside `Effect.promise(async
|
||||
// () => ...)`, so a re-entry point is needed. Sharing `memoMap` keeps a single
|
||||
// `ChildProcessSpawner` instance across the process.
|
||||
const processRuntime = ManagedRuntime.make(CrossSpawnSpawner.defaultLayer, { memoMap })
|
||||
const run = (cmd: string[], opts: Process.RunOptions = {}) =>
|
||||
processRuntime.runPromise(Process.run(cmd, { ...opts, nothrow: true }))
|
||||
const output = (cmd: string[], opts: Process.RunOptions = {}) =>
|
||||
processRuntime.runPromise(Process.text(cmd, { ...opts, nothrow: true }))
|
||||
|
||||
export interface Handle {
|
||||
process: ChildProcessWithoutNullStreams
|
||||
@@ -188,8 +200,12 @@ export const ESLint: Info = {
|
||||
await fs.rename(extractedPath, finalPath)
|
||||
|
||||
const npmCmd = process.platform === "win32" ? "npm.cmd" : "npm"
|
||||
await Process.run([npmCmd, "install"], { cwd: finalPath })
|
||||
await Process.run([npmCmd, "run", "compile"], { cwd: finalPath })
|
||||
await processRuntime.runPromise(
|
||||
Effect.gen(function* () {
|
||||
yield* Process.run([npmCmd, "install"], { cwd: finalPath })
|
||||
yield* Process.run([npmCmd, "run", "compile"], { cwd: finalPath })
|
||||
}),
|
||||
)
|
||||
|
||||
log.info("installed VS Code ESLint server", { serverPath })
|
||||
}
|
||||
@@ -570,9 +586,13 @@ export const ElixirLS: Info = {
|
||||
|
||||
const cwd = path.join(Global.Path.bin, "elixir-ls-master")
|
||||
const env = { MIX_ENV: "prod", ...process.env }
|
||||
await Process.run(["mix", "deps.get"], { cwd, env })
|
||||
await Process.run(["mix", "compile"], { cwd, env })
|
||||
await Process.run(["mix", "elixir_ls.release2", "-o", "release"], { cwd, env })
|
||||
await processRuntime.runPromise(
|
||||
Effect.gen(function* () {
|
||||
yield* Process.run(["mix", "deps.get"], { cwd, env })
|
||||
yield* Process.run(["mix", "compile"], { cwd, env })
|
||||
yield* Process.run(["mix", "elixir_ls.release2", "-o", "release"], { cwd, env })
|
||||
}),
|
||||
)
|
||||
|
||||
log.info(`installed elixir-ls`, {
|
||||
path: elixirLsPath,
|
||||
|
||||
@@ -79,7 +79,7 @@ export const experimentalHandlers = HttpApiBuilder.group(InstanceHttpApi, "exper
|
||||
const list = yield* registry.tools({
|
||||
providerID: ctx.query.provider,
|
||||
modelID: ctx.query.model,
|
||||
agent: yield* agents.get(yield* agents.defaultAgent()),
|
||||
agent: yield* agents.defaultInfo(),
|
||||
})
|
||||
return list.map((item) => ({
|
||||
id: item.id,
|
||||
|
||||
@@ -1083,8 +1083,8 @@ NOTE: At any point in time through this workflow you should feel free to ask the
|
||||
})
|
||||
|
||||
const createUserMessage = Effect.fn("SessionPrompt.createUserMessage")(function* (input: PromptInput) {
|
||||
const agentName = input.agent || (yield* agents.defaultAgent())
|
||||
const ag = yield* agents.get(agentName)
|
||||
const agentName = input.agent
|
||||
const ag = agentName ? yield* agents.get(agentName) : yield* agents.defaultInfo()
|
||||
if (!ag) {
|
||||
const available = (yield* agents.list()).filter((a) => !a.hidden).map((a) => a.name)
|
||||
const hint = available.length ? ` Available agents: ${available.join(", ")}` : ""
|
||||
@@ -1875,7 +1875,7 @@ NOTE: At any point in time through this workflow you should feel free to ask the
|
||||
yield* bus.publish(Session.Event.Error, { sessionID: input.sessionID, error: error.toObject() })
|
||||
throw error
|
||||
}
|
||||
const agentName = cmd.agent ?? input.agent ?? (yield* agents.defaultAgent())
|
||||
const agentName = cmd.agent ?? input.agent
|
||||
|
||||
const raw = input.arguments.match(argsRegex) ?? []
|
||||
const args = raw.map((arg) => arg.replace(quoteTrimRegex, ""))
|
||||
@@ -1908,7 +1908,7 @@ NOTE: At any point in time through this workflow you should feel free to ask the
|
||||
const sh = Shell.preferred(cfg.shell)
|
||||
const results = yield* Effect.promise(() =>
|
||||
Promise.all(
|
||||
shellMatches.map(async ([, cmd]) => (await Process.text([cmd], { shell: sh, nothrow: true })).text),
|
||||
shellMatches.map(async ([, cmd]) => (await Process.textPromise([cmd], { shell: sh, nothrow: true })).text),
|
||||
),
|
||||
)
|
||||
let index = 0
|
||||
@@ -1928,7 +1928,7 @@ NOTE: At any point in time through this workflow you should feel free to ask the
|
||||
|
||||
yield* getModel(taskModel.providerID, taskModel.modelID, input.sessionID)
|
||||
|
||||
const agent = yield* agents.get(agentName)
|
||||
const agent = agentName ? yield* agents.get(agentName) : yield* agents.defaultInfo()
|
||||
if (!agent) {
|
||||
const available = (yield* agents.list()).filter((a) => !a.hidden).map((a) => a.name)
|
||||
const hint = available.length ? ` Available agents: ${available.join(", ")}` : ""
|
||||
@@ -1952,7 +1952,7 @@ NOTE: At any point in time through this workflow you should feel free to ask the
|
||||
]
|
||||
: [...templateParts, ...(input.parts ?? [])]
|
||||
|
||||
const userAgent = isSubtask ? (input.agent ?? (yield* agents.defaultAgent())) : agentName
|
||||
const userAgent = isSubtask ? (input.agent ?? (yield* agents.defaultInfo()).name) : agent.name
|
||||
const userModel = isSubtask
|
||||
? input.model
|
||||
? Provider.parseModel(input.model)
|
||||
|
||||
@@ -7,11 +7,11 @@ export async function extractZip(zipPath: string, destDir: string) {
|
||||
const winDestDir = path.resolve(destDir)
|
||||
// $global:ProgressPreference suppresses PowerShell's blue progress bar popup
|
||||
const cmd = `$global:ProgressPreference = 'SilentlyContinue'; Expand-Archive -Path '${winZipPath}' -DestinationPath '${winDestDir}' -Force`
|
||||
await Process.run(["powershell", "-NoProfile", "-NonInteractive", "-Command", cmd])
|
||||
await Process.runPromise(["powershell", "-NoProfile", "-NonInteractive", "-Command", cmd])
|
||||
return
|
||||
}
|
||||
|
||||
await Process.run(["unzip", "-o", "-q", zipPath, "-d", destDir])
|
||||
await Process.runPromise(["unzip", "-o", "-q", zipPath, "-d", destDir])
|
||||
}
|
||||
|
||||
export * as Archive from "./archive"
|
||||
|
||||
@@ -1,6 +1,8 @@
|
||||
import { type ChildProcess } from "child_process"
|
||||
import { type ChildProcess as NodeChildProcess } from "child_process"
|
||||
import launch from "cross-spawn"
|
||||
import { buffer } from "node:stream/consumers"
|
||||
import { Effect, Stream } from "effect"
|
||||
import { ChildProcess, ChildProcessSpawner } from "effect/unstable/process"
|
||||
import { errorMessage } from "./error"
|
||||
|
||||
export type Stdio = "inherit" | "pipe" | "ignore"
|
||||
@@ -53,7 +55,7 @@ export class RunFailedError extends Error {
|
||||
}
|
||||
}
|
||||
|
||||
export type Child = ChildProcess & { exited: Promise<number> }
|
||||
export type Child = NodeChildProcess & { exited: Promise<number> }
|
||||
|
||||
export function spawn(cmd: string[], opts: Options = {}): Child {
|
||||
if (cmd.length === 0) throw new Error("Command is required")
|
||||
@@ -110,8 +112,104 @@ export function spawn(cmd: string[], opts: Options = {}): Child {
|
||||
return child
|
||||
}
|
||||
|
||||
export async function run(cmd: string[], opts: RunOptions = {}): Promise<Result> {
|
||||
const proc = spawn(cmd, {
|
||||
// Duplicated in `packages/sdk/js/src/process.ts` because the SDK cannot import
|
||||
// `opencode` without creating a cycle. Keep both copies in sync.
|
||||
export async function stop(proc: NodeChildProcess) {
|
||||
if (proc.exitCode !== null || proc.signalCode !== null) return
|
||||
|
||||
if (process.platform !== "win32" || !proc.pid) {
|
||||
proc.kill()
|
||||
return
|
||||
}
|
||||
|
||||
const out = await runPromise(["taskkill", "/pid", String(proc.pid), "/T", "/F"], {
|
||||
nothrow: true,
|
||||
})
|
||||
|
||||
if (out.code === 0) return
|
||||
proc.kill()
|
||||
}
|
||||
|
||||
const mergeEnv = (env: NodeJS.ProcessEnv | null | undefined): { env: Record<string, string>; extendEnv: boolean } => {
|
||||
if (env === null) return { env: {}, extendEnv: false }
|
||||
if (env === undefined) return { env: {}, extendEnv: true }
|
||||
const out: Record<string, string> = {}
|
||||
for (const [k, v] of Object.entries(env)) {
|
||||
if (v !== undefined) out[k] = v
|
||||
}
|
||||
return { env: out, extendEnv: true }
|
||||
}
|
||||
|
||||
export const run = Effect.fn("Process.run")(function* (cmd: string[], opts: RunOptions = {}) {
|
||||
if (cmd.length === 0) return yield* Effect.die(new Error("Command is required"))
|
||||
const spawner = yield* ChildProcessSpawner.ChildProcessSpawner
|
||||
const { env, extendEnv } = mergeEnv(opts.env)
|
||||
|
||||
const result = yield* Effect.scoped(
|
||||
Effect.gen(function* () {
|
||||
const proc = ChildProcess.make(cmd[0], cmd.slice(1), {
|
||||
cwd: opts.cwd,
|
||||
env,
|
||||
extendEnv,
|
||||
shell: opts.shell,
|
||||
stdin: opts.stdin ?? "ignore",
|
||||
stdout: "pipe",
|
||||
stderr: "pipe",
|
||||
})
|
||||
const handle = yield* spawner.spawn(proc)
|
||||
const [stdoutBytes, stderrBytes, exitCode] = yield* Effect.all(
|
||||
[Stream.mkUint8Array(handle.stdout), Stream.mkUint8Array(handle.stderr), handle.exitCode],
|
||||
{ concurrency: 3 },
|
||||
)
|
||||
return {
|
||||
code: exitCode as number,
|
||||
stdout: Buffer.from(stdoutBytes),
|
||||
stderr: Buffer.from(stderrBytes),
|
||||
} satisfies Result
|
||||
}),
|
||||
).pipe(
|
||||
Effect.catch((err) =>
|
||||
opts.nothrow
|
||||
? Effect.succeed({
|
||||
code: 1,
|
||||
stdout: Buffer.alloc(0),
|
||||
stderr: Buffer.from(errorMessage(err)),
|
||||
} satisfies Result)
|
||||
: Effect.die(err),
|
||||
),
|
||||
)
|
||||
|
||||
if (result.code === 0 || opts.nothrow) return result
|
||||
return yield* Effect.die(new RunFailedError(cmd, result.code, result.stdout, result.stderr))
|
||||
})
|
||||
|
||||
export const text = Effect.fn("Process.text")(function* (cmd: string[], opts: RunOptions = {}) {
|
||||
const out = yield* run(cmd, opts)
|
||||
return {
|
||||
...out,
|
||||
text: out.stdout.toString(),
|
||||
} satisfies TextResult
|
||||
})
|
||||
|
||||
export const lines = Effect.fn("Process.lines")(function* (cmd: string[], opts: RunOptions = {}) {
|
||||
const out = yield* text(cmd, opts)
|
||||
return out.text.split(/\r?\n/).filter(Boolean)
|
||||
})
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Promise-returning facades for legacy non-Effect callers.
|
||||
//
|
||||
// The new `run` / `text` / `lines` exports above return Effects. These
|
||||
// wrappers preserve the original Promise-based shape (failing with
|
||||
// `RunFailedError` on non-zero exit, etc.) and the legacy AbortSignal /
|
||||
// timeout semantics by using `spawn(...)` directly.
|
||||
//
|
||||
// New code should yield the Effect versions. These wrappers exist only to
|
||||
// avoid touching the remaining non-Effect call sites in this PR.
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
export function runPromise(cmd: string[], opts: RunOptions = {}): Promise<Result> {
|
||||
const spawnOpts = {
|
||||
cwd: opts.cwd,
|
||||
env: opts.env,
|
||||
stdin: opts.stdin,
|
||||
@@ -119,13 +217,16 @@ export async function run(cmd: string[], opts: RunOptions = {}): Promise<Result>
|
||||
abort: opts.abort,
|
||||
kill: opts.kill,
|
||||
timeout: opts.timeout,
|
||||
stdout: "pipe",
|
||||
stderr: "pipe",
|
||||
})
|
||||
stdout: "pipe" as const,
|
||||
stderr: "pipe" as const,
|
||||
}
|
||||
|
||||
if (!proc.stdout || !proc.stderr) throw new Error("Process output not available")
|
||||
// Preserve the legacy abort/timeout semantics by using `spawn(...)` directly
|
||||
// rather than the Effect path (which lacks AbortSignal hooks today).
|
||||
const proc = spawn(cmd, spawnOpts)
|
||||
if (!proc.stdout || !proc.stderr) return Promise.reject(new Error("Process output not available"))
|
||||
|
||||
const out = await Promise.all([proc.exited, buffer(proc.stdout), buffer(proc.stderr)])
|
||||
return Promise.all([proc.exited, buffer(proc.stdout), buffer(proc.stderr)])
|
||||
.then(([code, stdout, stderr]) => ({
|
||||
code,
|
||||
stdout,
|
||||
@@ -137,40 +238,24 @@ export async function run(cmd: string[], opts: RunOptions = {}): Promise<Result>
|
||||
code: 1,
|
||||
stdout: Buffer.alloc(0),
|
||||
stderr: Buffer.from(errorMessage(err)),
|
||||
}
|
||||
} satisfies Result
|
||||
})
|
||||
.then((out) => {
|
||||
if (out.code === 0 || opts.nothrow) return out
|
||||
throw new RunFailedError(cmd, out.code, out.stdout, out.stderr)
|
||||
})
|
||||
if (out.code === 0 || opts.nothrow) return out
|
||||
throw new RunFailedError(cmd, out.code, out.stdout, out.stderr)
|
||||
}
|
||||
|
||||
// Duplicated in `packages/sdk/js/src/process.ts` because the SDK cannot import
|
||||
// `opencode` without creating a cycle. Keep both copies in sync.
|
||||
export async function stop(proc: ChildProcess) {
|
||||
if (proc.exitCode !== null || proc.signalCode !== null) return
|
||||
|
||||
if (process.platform !== "win32" || !proc.pid) {
|
||||
proc.kill()
|
||||
return
|
||||
}
|
||||
|
||||
const out = await run(["taskkill", "/pid", String(proc.pid), "/T", "/F"], {
|
||||
nothrow: true,
|
||||
})
|
||||
|
||||
if (out.code === 0) return
|
||||
proc.kill()
|
||||
}
|
||||
|
||||
export async function text(cmd: string[], opts: RunOptions = {}): Promise<TextResult> {
|
||||
const out = await run(cmd, opts)
|
||||
export async function textPromise(cmd: string[], opts: RunOptions = {}): Promise<TextResult> {
|
||||
const out = await runPromise(cmd, opts)
|
||||
return {
|
||||
...out,
|
||||
text: out.stdout.toString(),
|
||||
}
|
||||
}
|
||||
|
||||
export async function lines(cmd: string[], opts: RunOptions = {}): Promise<string[]> {
|
||||
return (await text(cmd, opts)).text.split(/\r?\n/).filter(Boolean)
|
||||
export async function linesPromise(cmd: string[], opts: RunOptions = {}): Promise<string[]> {
|
||||
return (await textPromise(cmd, opts)).text.split(/\r?\n/).filter(Boolean)
|
||||
}
|
||||
|
||||
export * as Process from "./process"
|
||||
|
||||
@@ -638,6 +638,14 @@ it.instance("defaultAgent returns build when no default_agent config", () =>
|
||||
}),
|
||||
)
|
||||
|
||||
it.instance("defaultInfo returns resolved build agent when no default_agent config", () =>
|
||||
Effect.gen(function* () {
|
||||
const agent = yield* load((svc) => svc.defaultInfo())
|
||||
expect(agent.name).toBe("build")
|
||||
expect(agent.mode).toBe("primary")
|
||||
}),
|
||||
)
|
||||
|
||||
it.instance(
|
||||
"defaultAgent respects default_agent config set to plan",
|
||||
() =>
|
||||
|
||||
@@ -1,88 +1,88 @@
|
||||
import { afterEach, describe, expect, test } from "bun:test"
|
||||
import { Schema } from "effect"
|
||||
import { CrossSpawnSpawner } from "@opencode-ai/core/cross-spawn-spawner"
|
||||
import { afterEach, describe, expect } from "bun:test"
|
||||
import { Deferred, Effect, Layer, Schema } from "effect"
|
||||
import { Bus } from "../../src/bus"
|
||||
import { BusEvent } from "../../src/bus/bus-event"
|
||||
import { Instance } from "../../src/project/instance"
|
||||
import { WithInstance } from "../../src/project/with-instance"
|
||||
import { disposeAllInstances, tmpdir } from "../fixture/fixture"
|
||||
import { disposeAllInstances, provideInstance, tmpdirScoped } from "../fixture/fixture"
|
||||
import { testEffect } from "../lib/effect"
|
||||
|
||||
const TestEvent = BusEvent.define("test.integration", Schema.Struct({ value: Schema.Number }))
|
||||
|
||||
function withInstance(directory: string, fn: () => Promise<void>) {
|
||||
return WithInstance.provide({ directory, fn })
|
||||
}
|
||||
const it = testEffect(Layer.mergeAll(Bus.layer, CrossSpawnSpawner.defaultLayer))
|
||||
|
||||
describe("Bus integration: acquireRelease subscriber pattern", () => {
|
||||
afterEach(() => disposeAllInstances())
|
||||
|
||||
test("subscriber via callback facade receives events and cleans up on unsub", async () => {
|
||||
await using tmp = await tmpdir()
|
||||
const received: number[] = []
|
||||
it.instance("subscriber via callback facade receives events and cleans up on unsub", () =>
|
||||
Effect.gen(function* () {
|
||||
const bus = yield* Bus.Service
|
||||
const received: number[] = []
|
||||
const receivedTwo = yield* Deferred.make<void>()
|
||||
|
||||
await withInstance(tmp.path, async () => {
|
||||
const unsub = Bus.subscribe(TestEvent, (evt) => {
|
||||
const unsub = yield* bus.subscribeCallback(TestEvent, (evt) => {
|
||||
received.push(evt.properties.value)
|
||||
if (received.length === 2) Deferred.doneUnsafe(receivedTwo, Effect.void)
|
||||
})
|
||||
await Bun.sleep(10)
|
||||
await Bus.publish(TestEvent, { value: 1 })
|
||||
await Bus.publish(TestEvent, { value: 2 })
|
||||
await Bun.sleep(10)
|
||||
yield* bus.publish(TestEvent, { value: 1 })
|
||||
yield* bus.publish(TestEvent, { value: 2 })
|
||||
yield* Deferred.await(receivedTwo).pipe(Effect.timeout("2 seconds"))
|
||||
|
||||
expect(received).toEqual([1, 2])
|
||||
|
||||
unsub()
|
||||
await Bun.sleep(10)
|
||||
await Bus.publish(TestEvent, { value: 3 })
|
||||
await Bun.sleep(10)
|
||||
yield* Effect.sync(unsub)
|
||||
yield* bus.publish(TestEvent, { value: 3 })
|
||||
yield* Effect.sleep("10 millis")
|
||||
|
||||
expect(received).toEqual([1, 2])
|
||||
})
|
||||
})
|
||||
}),
|
||||
)
|
||||
|
||||
test("subscribeAll receives events from multiple types", async () => {
|
||||
await using tmp = await tmpdir()
|
||||
const received: Array<{ type: string; value?: number }> = []
|
||||
it.instance("subscribeAll receives events from multiple types", () =>
|
||||
Effect.gen(function* () {
|
||||
const bus = yield* Bus.Service
|
||||
const received: Array<{ type: string; value?: number }> = []
|
||||
const OtherEvent = BusEvent.define("test.other", Schema.Struct({ value: Schema.Number }))
|
||||
const receivedTwo = yield* Deferred.make<void>()
|
||||
|
||||
const OtherEvent = BusEvent.define("test.other", Schema.Struct({ value: Schema.Number }))
|
||||
|
||||
await withInstance(tmp.path, async () => {
|
||||
Bus.subscribeAll((evt) => {
|
||||
yield* bus.subscribeAllCallback((evt) => {
|
||||
received.push({ type: evt.type, value: evt.properties.value })
|
||||
if (received.length === 2) Deferred.doneUnsafe(receivedTwo, Effect.void)
|
||||
})
|
||||
await Bun.sleep(10)
|
||||
await Bus.publish(TestEvent, { value: 10 })
|
||||
await Bus.publish(OtherEvent, { value: 20 })
|
||||
await Bun.sleep(10)
|
||||
})
|
||||
yield* bus.publish(TestEvent, { value: 10 })
|
||||
yield* bus.publish(OtherEvent, { value: 20 })
|
||||
yield* Deferred.await(receivedTwo).pipe(Effect.timeout("2 seconds"))
|
||||
|
||||
expect(received).toEqual([
|
||||
{ type: "test.integration", value: 10 },
|
||||
{ type: "test.other", value: 20 },
|
||||
])
|
||||
})
|
||||
expect(received).toEqual([
|
||||
{ type: "test.integration", value: 10 },
|
||||
{ type: "test.other", value: 20 },
|
||||
])
|
||||
}),
|
||||
)
|
||||
|
||||
test("subscriber cleanup on instance disposal interrupts the stream", async () => {
|
||||
await using tmp = await tmpdir()
|
||||
const received: number[] = []
|
||||
let disposed = false
|
||||
it.live("subscriber cleanup on instance disposal interrupts the stream", () =>
|
||||
Effect.gen(function* () {
|
||||
const dir = yield* tmpdirScoped()
|
||||
const received: number[] = []
|
||||
const seen = yield* Deferred.make<void>()
|
||||
const disposed = yield* Deferred.make<void>()
|
||||
|
||||
await withInstance(tmp.path, async () => {
|
||||
Bus.subscribeAll((evt) => {
|
||||
if (evt.type === Bus.InstanceDisposed.type) {
|
||||
disposed = true
|
||||
return
|
||||
}
|
||||
received.push(evt.properties.value)
|
||||
})
|
||||
await Bun.sleep(10)
|
||||
await Bus.publish(TestEvent, { value: 1 })
|
||||
await Bun.sleep(10)
|
||||
})
|
||||
yield* Effect.gen(function* () {
|
||||
const bus = yield* Bus.Service
|
||||
yield* bus.subscribeAllCallback((evt) => {
|
||||
if (evt.type === Bus.InstanceDisposed.type) {
|
||||
Deferred.doneUnsafe(disposed, Effect.void)
|
||||
return
|
||||
}
|
||||
received.push(evt.properties.value)
|
||||
Deferred.doneUnsafe(seen, Effect.void)
|
||||
})
|
||||
yield* bus.publish(TestEvent, { value: 1 })
|
||||
yield* Deferred.await(seen).pipe(Effect.timeout("2 seconds"))
|
||||
}).pipe(provideInstance(dir))
|
||||
|
||||
await disposeAllInstances()
|
||||
await Bun.sleep(50)
|
||||
yield* Effect.promise(() => disposeAllInstances())
|
||||
yield* Deferred.await(disposed).pipe(Effect.timeout("2 seconds"))
|
||||
|
||||
expect(received).toEqual([1])
|
||||
expect(disposed).toBe(true)
|
||||
})
|
||||
expect(received).toEqual([1])
|
||||
}),
|
||||
)
|
||||
})
|
||||
|
||||
@@ -1,220 +1,240 @@
|
||||
import { afterEach, describe, expect, test } from "bun:test"
|
||||
import { Schema } from "effect"
|
||||
import { afterEach, describe, expect } from "bun:test"
|
||||
import { CrossSpawnSpawner } from "@opencode-ai/core/cross-spawn-spawner"
|
||||
import { Deferred, Effect, Layer, Schema } from "effect"
|
||||
import { Bus } from "../../src/bus"
|
||||
import { BusEvent } from "../../src/bus/bus-event"
|
||||
import { Instance } from "../../src/project/instance"
|
||||
import { WithInstance } from "../../src/project/with-instance"
|
||||
import { disposeAllInstances, tmpdir } from "../fixture/fixture"
|
||||
import { disposeAllInstances, provideInstance, tmpdirScoped } from "../fixture/fixture"
|
||||
import { testEffect } from "../lib/effect"
|
||||
|
||||
const TestEvent = {
|
||||
Ping: BusEvent.define("test.ping", Schema.Struct({ value: Schema.Number })),
|
||||
Pong: BusEvent.define("test.pong", Schema.Struct({ message: Schema.String })),
|
||||
}
|
||||
|
||||
function withInstance(directory: string, fn: () => Promise<void>) {
|
||||
return WithInstance.provide({ directory, fn })
|
||||
}
|
||||
const it = testEffect(Layer.mergeAll(Bus.layer, CrossSpawnSpawner.defaultLayer))
|
||||
|
||||
describe("Bus", () => {
|
||||
afterEach(() => disposeAllInstances())
|
||||
|
||||
describe("publish + subscribe", () => {
|
||||
test("subscriber is live immediately after subscribe returns", async () => {
|
||||
await using tmp = await tmpdir()
|
||||
const received: number[] = []
|
||||
it.instance("subscriber is live immediately after subscribe returns", () =>
|
||||
Effect.gen(function* () {
|
||||
const bus = yield* Bus.Service
|
||||
const received: number[] = []
|
||||
const done = yield* Deferred.make<void>()
|
||||
|
||||
await withInstance(tmp.path, async () => {
|
||||
Bus.subscribe(TestEvent.Ping, (evt) => {
|
||||
yield* bus.subscribeCallback(TestEvent.Ping, (evt) => {
|
||||
received.push(evt.properties.value)
|
||||
Deferred.doneUnsafe(done, Effect.void)
|
||||
})
|
||||
await Bus.publish(TestEvent.Ping, { value: 42 })
|
||||
await Bun.sleep(10)
|
||||
})
|
||||
yield* bus.publish(TestEvent.Ping, { value: 42 })
|
||||
yield* Deferred.await(done).pipe(Effect.timeout("2 seconds"))
|
||||
|
||||
expect(received).toEqual([42])
|
||||
})
|
||||
expect(received).toEqual([42])
|
||||
}),
|
||||
)
|
||||
|
||||
test("subscriber receives matching events", async () => {
|
||||
await using tmp = await tmpdir()
|
||||
const received: number[] = []
|
||||
it.instance("subscriber receives matching events", () =>
|
||||
Effect.gen(function* () {
|
||||
const bus = yield* Bus.Service
|
||||
const received: number[] = []
|
||||
const done = yield* Deferred.make<void>()
|
||||
|
||||
await withInstance(tmp.path, async () => {
|
||||
Bus.subscribe(TestEvent.Ping, (evt) => {
|
||||
yield* bus.subscribeCallback(TestEvent.Ping, (evt) => {
|
||||
received.push(evt.properties.value)
|
||||
if (received.length === 2) Deferred.doneUnsafe(done, Effect.void)
|
||||
})
|
||||
// Give the subscriber fiber time to start consuming
|
||||
await Bun.sleep(10)
|
||||
await Bus.publish(TestEvent.Ping, { value: 42 })
|
||||
await Bus.publish(TestEvent.Ping, { value: 99 })
|
||||
// Give subscriber time to process
|
||||
await Bun.sleep(10)
|
||||
})
|
||||
yield* bus.publish(TestEvent.Ping, { value: 42 })
|
||||
yield* bus.publish(TestEvent.Ping, { value: 99 })
|
||||
yield* Deferred.await(done).pipe(Effect.timeout("2 seconds"))
|
||||
|
||||
expect(received).toEqual([42, 99])
|
||||
})
|
||||
expect(received).toEqual([42, 99])
|
||||
}),
|
||||
)
|
||||
|
||||
test("subscriber does not receive events of other types", async () => {
|
||||
await using tmp = await tmpdir()
|
||||
const pings: number[] = []
|
||||
it.instance("subscriber does not receive events of other types", () =>
|
||||
Effect.gen(function* () {
|
||||
const bus = yield* Bus.Service
|
||||
const pings: number[] = []
|
||||
const done = yield* Deferred.make<void>()
|
||||
|
||||
await withInstance(tmp.path, async () => {
|
||||
Bus.subscribe(TestEvent.Ping, (evt) => {
|
||||
yield* bus.subscribeCallback(TestEvent.Ping, (evt) => {
|
||||
pings.push(evt.properties.value)
|
||||
Deferred.doneUnsafe(done, Effect.void)
|
||||
})
|
||||
await Bun.sleep(10)
|
||||
await Bus.publish(TestEvent.Pong, { message: "hello" })
|
||||
await Bus.publish(TestEvent.Ping, { value: 1 })
|
||||
await Bun.sleep(10)
|
||||
})
|
||||
yield* bus.publish(TestEvent.Pong, { message: "hello" })
|
||||
yield* bus.publish(TestEvent.Ping, { value: 1 })
|
||||
yield* Deferred.await(done).pipe(Effect.timeout("2 seconds"))
|
||||
|
||||
expect(pings).toEqual([1])
|
||||
})
|
||||
expect(pings).toEqual([1])
|
||||
}),
|
||||
)
|
||||
|
||||
test("publish with no subscribers does not throw", async () => {
|
||||
await using tmp = await tmpdir()
|
||||
|
||||
await withInstance(tmp.path, async () => {
|
||||
await Bus.publish(TestEvent.Ping, { value: 1 })
|
||||
})
|
||||
})
|
||||
it.instance("publish with no subscribers does not throw", () =>
|
||||
Effect.gen(function* () {
|
||||
const bus = yield* Bus.Service
|
||||
yield* bus.publish(TestEvent.Ping, { value: 1 })
|
||||
}),
|
||||
)
|
||||
})
|
||||
|
||||
describe("unsubscribe", () => {
|
||||
test("unsubscribe stops delivery", async () => {
|
||||
await using tmp = await tmpdir()
|
||||
const received: number[] = []
|
||||
it.instance("unsubscribe stops delivery", () =>
|
||||
Effect.gen(function* () {
|
||||
const bus = yield* Bus.Service
|
||||
const received: number[] = []
|
||||
const first = yield* Deferred.make<void>()
|
||||
|
||||
await withInstance(tmp.path, async () => {
|
||||
const unsub = Bus.subscribe(TestEvent.Ping, (evt) => {
|
||||
const unsub = yield* bus.subscribeCallback(TestEvent.Ping, (evt) => {
|
||||
received.push(evt.properties.value)
|
||||
if (evt.properties.value === 1) Deferred.doneUnsafe(first, Effect.void)
|
||||
})
|
||||
await Bun.sleep(10)
|
||||
await Bus.publish(TestEvent.Ping, { value: 1 })
|
||||
await Bun.sleep(10)
|
||||
unsub()
|
||||
await Bun.sleep(10)
|
||||
await Bus.publish(TestEvent.Ping, { value: 2 })
|
||||
await Bun.sleep(10)
|
||||
})
|
||||
yield* bus.publish(TestEvent.Ping, { value: 1 })
|
||||
yield* Deferred.await(first).pipe(Effect.timeout("2 seconds"))
|
||||
yield* Effect.sync(unsub)
|
||||
yield* bus.publish(TestEvent.Ping, { value: 2 })
|
||||
yield* Effect.sleep("10 millis")
|
||||
|
||||
expect(received).toEqual([1])
|
||||
})
|
||||
expect(received).toEqual([1])
|
||||
}),
|
||||
)
|
||||
})
|
||||
|
||||
describe("subscribeAll", () => {
|
||||
test("subscribeAll is live immediately after subscribe returns", async () => {
|
||||
await using tmp = await tmpdir()
|
||||
const received: string[] = []
|
||||
it.instance("subscribeAll is live immediately after subscribe returns", () =>
|
||||
Effect.gen(function* () {
|
||||
const bus = yield* Bus.Service
|
||||
const received: string[] = []
|
||||
const done = yield* Deferred.make<void>()
|
||||
|
||||
await withInstance(tmp.path, async () => {
|
||||
Bus.subscribeAll((evt) => {
|
||||
yield* bus.subscribeAllCallback((evt) => {
|
||||
received.push(evt.type)
|
||||
Deferred.doneUnsafe(done, Effect.void)
|
||||
})
|
||||
await Bus.publish(TestEvent.Ping, { value: 1 })
|
||||
await Bun.sleep(10)
|
||||
})
|
||||
yield* bus.publish(TestEvent.Ping, { value: 1 })
|
||||
yield* Deferred.await(done).pipe(Effect.timeout("2 seconds"))
|
||||
|
||||
expect(received).toEqual(["test.ping"])
|
||||
})
|
||||
expect(received).toEqual(["test.ping"])
|
||||
}),
|
||||
)
|
||||
|
||||
test("receives all event types", async () => {
|
||||
await using tmp = await tmpdir()
|
||||
const received: string[] = []
|
||||
it.instance("receives all event types", () =>
|
||||
Effect.gen(function* () {
|
||||
const bus = yield* Bus.Service
|
||||
const received: string[] = []
|
||||
const done = yield* Deferred.make<void>()
|
||||
|
||||
await withInstance(tmp.path, async () => {
|
||||
Bus.subscribeAll((evt) => {
|
||||
yield* bus.subscribeAllCallback((evt) => {
|
||||
received.push(evt.type)
|
||||
if (received.length === 2) Deferred.doneUnsafe(done, Effect.void)
|
||||
})
|
||||
await Bun.sleep(10)
|
||||
await Bus.publish(TestEvent.Ping, { value: 1 })
|
||||
await Bus.publish(TestEvent.Pong, { message: "hi" })
|
||||
await Bun.sleep(10)
|
||||
})
|
||||
yield* bus.publish(TestEvent.Ping, { value: 1 })
|
||||
yield* bus.publish(TestEvent.Pong, { message: "hi" })
|
||||
yield* Deferred.await(done).pipe(Effect.timeout("2 seconds"))
|
||||
|
||||
expect(received).toContain("test.ping")
|
||||
expect(received).toContain("test.pong")
|
||||
})
|
||||
expect(received).toContain("test.ping")
|
||||
expect(received).toContain("test.pong")
|
||||
}),
|
||||
)
|
||||
})
|
||||
|
||||
describe("multiple subscribers", () => {
|
||||
test("all subscribers for same event type are called", async () => {
|
||||
await using tmp = await tmpdir()
|
||||
const a: number[] = []
|
||||
const b: number[] = []
|
||||
it.instance("all subscribers for same event type are called", () =>
|
||||
Effect.gen(function* () {
|
||||
const bus = yield* Bus.Service
|
||||
const a: number[] = []
|
||||
const b: number[] = []
|
||||
const doneA = yield* Deferred.make<void>()
|
||||
const doneB = yield* Deferred.make<void>()
|
||||
|
||||
await withInstance(tmp.path, async () => {
|
||||
Bus.subscribe(TestEvent.Ping, (evt) => {
|
||||
yield* bus.subscribeCallback(TestEvent.Ping, (evt) => {
|
||||
a.push(evt.properties.value)
|
||||
Deferred.doneUnsafe(doneA, Effect.void)
|
||||
})
|
||||
Bus.subscribe(TestEvent.Ping, (evt) => {
|
||||
yield* bus.subscribeCallback(TestEvent.Ping, (evt) => {
|
||||
b.push(evt.properties.value)
|
||||
Deferred.doneUnsafe(doneB, Effect.void)
|
||||
})
|
||||
await Bun.sleep(10)
|
||||
await Bus.publish(TestEvent.Ping, { value: 7 })
|
||||
await Bun.sleep(10)
|
||||
})
|
||||
yield* bus.publish(TestEvent.Ping, { value: 7 })
|
||||
yield* Deferred.await(doneA).pipe(Effect.timeout("2 seconds"))
|
||||
yield* Deferred.await(doneB).pipe(Effect.timeout("2 seconds"))
|
||||
|
||||
expect(a).toEqual([7])
|
||||
expect(b).toEqual([7])
|
||||
})
|
||||
expect(a).toEqual([7])
|
||||
expect(b).toEqual([7])
|
||||
}),
|
||||
)
|
||||
})
|
||||
|
||||
describe("instance isolation", () => {
|
||||
test("events in one directory do not reach subscribers in another", async () => {
|
||||
await using tmpA = await tmpdir()
|
||||
await using tmpB = await tmpdir()
|
||||
const receivedA: number[] = []
|
||||
const receivedB: number[] = []
|
||||
it.live("events in one directory do not reach subscribers in another", () =>
|
||||
Effect.gen(function* () {
|
||||
const tmpA = yield* tmpdirScoped()
|
||||
const tmpB = yield* tmpdirScoped()
|
||||
const receivedA: number[] = []
|
||||
const receivedB: number[] = []
|
||||
const doneA = yield* Deferred.make<void>()
|
||||
const doneB = yield* Deferred.make<void>()
|
||||
|
||||
await withInstance(tmpA.path, async () => {
|
||||
Bus.subscribe(TestEvent.Ping, (evt) => {
|
||||
receivedA.push(evt.properties.value)
|
||||
})
|
||||
await Bun.sleep(10)
|
||||
})
|
||||
yield* Effect.gen(function* () {
|
||||
const bus = yield* Bus.Service
|
||||
yield* bus.subscribeCallback(TestEvent.Ping, (evt) => {
|
||||
receivedA.push(evt.properties.value)
|
||||
Deferred.doneUnsafe(doneA, Effect.void)
|
||||
})
|
||||
}).pipe(provideInstance(tmpA))
|
||||
|
||||
await withInstance(tmpB.path, async () => {
|
||||
Bus.subscribe(TestEvent.Ping, (evt) => {
|
||||
receivedB.push(evt.properties.value)
|
||||
})
|
||||
await Bun.sleep(10)
|
||||
})
|
||||
yield* Effect.gen(function* () {
|
||||
const bus = yield* Bus.Service
|
||||
yield* bus.subscribeCallback(TestEvent.Ping, (evt) => {
|
||||
receivedB.push(evt.properties.value)
|
||||
Deferred.doneUnsafe(doneB, Effect.void)
|
||||
})
|
||||
}).pipe(provideInstance(tmpB))
|
||||
|
||||
await withInstance(tmpA.path, async () => {
|
||||
await Bus.publish(TestEvent.Ping, { value: 1 })
|
||||
await Bun.sleep(10)
|
||||
})
|
||||
yield* Effect.gen(function* () {
|
||||
const bus = yield* Bus.Service
|
||||
yield* bus.publish(TestEvent.Ping, { value: 1 })
|
||||
}).pipe(provideInstance(tmpA))
|
||||
|
||||
await withInstance(tmpB.path, async () => {
|
||||
await Bus.publish(TestEvent.Ping, { value: 2 })
|
||||
await Bun.sleep(10)
|
||||
})
|
||||
yield* Effect.gen(function* () {
|
||||
const bus = yield* Bus.Service
|
||||
yield* bus.publish(TestEvent.Ping, { value: 2 })
|
||||
}).pipe(provideInstance(tmpB))
|
||||
|
||||
expect(receivedA).toEqual([1])
|
||||
expect(receivedB).toEqual([2])
|
||||
})
|
||||
yield* Deferred.await(doneA).pipe(Effect.timeout("2 seconds"))
|
||||
yield* Deferred.await(doneB).pipe(Effect.timeout("2 seconds"))
|
||||
|
||||
expect(receivedA).toEqual([1])
|
||||
expect(receivedB).toEqual([2])
|
||||
}),
|
||||
)
|
||||
})
|
||||
|
||||
describe("instance disposal", () => {
|
||||
test("InstanceDisposed is delivered to wildcard subscribers before stream ends", async () => {
|
||||
await using tmp = await tmpdir()
|
||||
const received: string[] = []
|
||||
it.live("InstanceDisposed is delivered to wildcard subscribers before stream ends", () =>
|
||||
Effect.gen(function* () {
|
||||
const tmp = yield* tmpdirScoped()
|
||||
const received: string[] = []
|
||||
const seen = yield* Deferred.make<void>()
|
||||
const disposed = yield* Deferred.make<void>()
|
||||
|
||||
await withInstance(tmp.path, async () => {
|
||||
Bus.subscribeAll((evt) => {
|
||||
received.push(evt.type)
|
||||
})
|
||||
await Bun.sleep(10)
|
||||
await Bus.publish(TestEvent.Ping, { value: 1 })
|
||||
await Bun.sleep(10)
|
||||
})
|
||||
yield* Effect.gen(function* () {
|
||||
const bus = yield* Bus.Service
|
||||
yield* bus.subscribeAllCallback((evt) => {
|
||||
received.push(evt.type)
|
||||
if (evt.type === TestEvent.Ping.type) Deferred.doneUnsafe(seen, Effect.void)
|
||||
if (evt.type === Bus.InstanceDisposed.type) Deferred.doneUnsafe(disposed, Effect.void)
|
||||
})
|
||||
yield* bus.publish(TestEvent.Ping, { value: 1 })
|
||||
yield* Deferred.await(seen).pipe(Effect.timeout("2 seconds"))
|
||||
}).pipe(provideInstance(tmp))
|
||||
|
||||
// disposeAllInstances triggers the finalizer which publishes InstanceDisposed
|
||||
await disposeAllInstances()
|
||||
await Bun.sleep(50)
|
||||
yield* Effect.promise(disposeAllInstances)
|
||||
yield* Deferred.await(disposed).pipe(Effect.timeout("2 seconds"))
|
||||
|
||||
expect(received).toContain("test.ping")
|
||||
expect(received).toContain(Bus.InstanceDisposed.type)
|
||||
})
|
||||
expect(received).toContain("test.ping")
|
||||
expect(received).toContain(Bus.InstanceDisposed.type)
|
||||
}),
|
||||
)
|
||||
})
|
||||
})
|
||||
|
||||
@@ -135,7 +135,7 @@ export function tmpdirScoped(options?: { git?: boolean; config?: Partial<Config.
|
||||
yield* git("config", "commit.gpgsign", "false")
|
||||
yield* git("config", "user.email", "test@opencode.test")
|
||||
yield* git("config", "user.name", "Test")
|
||||
yield* git("commit", "--allow-empty", "-m", "root commit")
|
||||
yield* git("commit", "--allow-empty", "-m", `root commit ${dir}`)
|
||||
}
|
||||
|
||||
if (options?.config) {
|
||||
|
||||
@@ -17,7 +17,7 @@ type Msg = {
|
||||
}
|
||||
|
||||
function run(msg: Msg) {
|
||||
return Process.run([process.execPath, worker, JSON.stringify(msg)], {
|
||||
return Process.runPromise([process.execPath, worker, JSON.stringify(msg)], {
|
||||
cwd: root,
|
||||
nothrow: true,
|
||||
})
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
@@ -12,7 +12,7 @@ const root = path.join(import.meta.dir, "../..")
|
||||
const worker = path.join(import.meta.dir, "../fixture/plugin-meta-worker.ts")
|
||||
|
||||
function run(input: { file: string; spec: string; target: string; id: string }) {
|
||||
return Process.run([process.execPath, worker, JSON.stringify(input)], {
|
||||
return Process.runPromise([process.execPath, worker, JSON.stringify(input)], {
|
||||
cwd: root,
|
||||
nothrow: true,
|
||||
})
|
||||
|
||||
@@ -1,13 +1,12 @@
|
||||
import { afterEach, describe, expect } from "bun:test"
|
||||
import { CrossSpawnSpawner } from "@opencode-ai/core/cross-spawn-spawner"
|
||||
import { Effect, Fiber, Layer } from "effect"
|
||||
import { Deferred, Effect, Fiber, Layer } from "effect"
|
||||
import { InstanceRef } from "../../src/effect/instance-ref"
|
||||
import { registerDisposer } from "../../src/effect/instance-registry"
|
||||
import { InstanceBootstrap } from "../../src/project/bootstrap-service"
|
||||
import { Instance } from "../../src/project/instance"
|
||||
import { WithInstance } from "../../src/project/with-instance"
|
||||
import { InstanceStore } from "../../src/project/instance-store"
|
||||
import { disposeAllInstances, tmpdirScoped } from "../fixture/fixture"
|
||||
import { disposeAllInstances, TestInstance, tmpdirScoped } from "../fixture/fixture"
|
||||
import { testEffect } from "../lib/effect"
|
||||
|
||||
let bootstrapRun: Effect.Effect<void> = Effect.void
|
||||
@@ -75,18 +74,18 @@ describe("InstanceStore", () => {
|
||||
Effect.gen(function* () {
|
||||
const dir = yield* tmpdirScoped({ git: true })
|
||||
const store = yield* InstanceStore.Service
|
||||
const started = Promise.withResolvers<void>()
|
||||
const release = Promise.withResolvers<void>()
|
||||
const started = yield* Deferred.make<void>()
|
||||
const release = yield* Deferred.make<void>()
|
||||
let initialized = 0
|
||||
|
||||
bootstrapRun = Effect.promise(async () => {
|
||||
bootstrapRun = Effect.gen(function* () {
|
||||
initialized++
|
||||
started.resolve()
|
||||
await release.promise
|
||||
yield* Deferred.succeed(started, undefined)
|
||||
yield* Deferred.await(release)
|
||||
})
|
||||
const first = yield* store.load({ directory: dir }).pipe(Effect.forkScoped)
|
||||
|
||||
yield* Effect.promise(() => started.promise)
|
||||
yield* Deferred.await(started)
|
||||
|
||||
bootstrapRun = Effect.sync(() => {
|
||||
initialized++
|
||||
@@ -94,7 +93,7 @@ describe("InstanceStore", () => {
|
||||
const second = yield* store.load({ directory: dir }).pipe(Effect.forkScoped)
|
||||
|
||||
expect(initialized).toBe(1)
|
||||
release.resolve()
|
||||
yield* Deferred.succeed(release, undefined)
|
||||
|
||||
const [firstCtx, secondCtx] = yield* Effect.all([Fiber.join(first), Fiber.join(second)])
|
||||
expect(secondCtx).toBe(firstCtx)
|
||||
@@ -147,8 +146,8 @@ describe("InstanceStore", () => {
|
||||
Effect.gen(function* () {
|
||||
const dir = yield* tmpdirScoped({ git: true })
|
||||
const store = yield* InstanceStore.Service
|
||||
const reloading = Promise.withResolvers<void>()
|
||||
const releaseReload = Promise.withResolvers<void>()
|
||||
const reloading = yield* Deferred.make<void>()
|
||||
const releaseReload = yield* Deferred.make<void>()
|
||||
const disposed: Array<string> = []
|
||||
const off = registerDisposer(async (directory) => {
|
||||
disposed.push(directory)
|
||||
@@ -156,15 +155,15 @@ describe("InstanceStore", () => {
|
||||
yield* Effect.addFinalizer(() => Effect.sync(off))
|
||||
|
||||
const first = yield* store.load({ directory: dir })
|
||||
bootstrapRun = Effect.promise(async () => {
|
||||
reloading.resolve()
|
||||
await releaseReload.promise
|
||||
bootstrapRun = Effect.gen(function* () {
|
||||
yield* Deferred.succeed(reloading, undefined)
|
||||
yield* Deferred.await(releaseReload)
|
||||
})
|
||||
const reload = yield* store.reload({ directory: dir }).pipe(Effect.forkScoped)
|
||||
|
||||
yield* Effect.promise(() => reloading.promise)
|
||||
yield* Deferred.await(reloading)
|
||||
const staleDispose = yield* store.dispose(first).pipe(Effect.forkScoped)
|
||||
releaseReload.resolve()
|
||||
yield* Deferred.succeed(releaseReload, undefined)
|
||||
|
||||
const second = yield* Fiber.join(reload)
|
||||
yield* Fiber.join(staleDispose)
|
||||
@@ -178,23 +177,23 @@ describe("InstanceStore", () => {
|
||||
Effect.gen(function* () {
|
||||
const dir = yield* tmpdirScoped({ git: true })
|
||||
const store = yield* InstanceStore.Service
|
||||
const disposing = Promise.withResolvers<void>()
|
||||
const releaseDispose = Promise.withResolvers<void>()
|
||||
const disposing = yield* Deferred.make<void>()
|
||||
const releaseDispose = yield* Deferred.make<void>()
|
||||
const disposed: Array<string> = []
|
||||
const off = registerDisposer(async (directory) => {
|
||||
disposed.push(directory)
|
||||
disposing.resolve()
|
||||
await releaseDispose.promise
|
||||
Deferred.doneUnsafe(disposing, Effect.void)
|
||||
await Effect.runPromise(Deferred.await(releaseDispose))
|
||||
})
|
||||
yield* Effect.addFinalizer(() => Effect.sync(off))
|
||||
|
||||
yield* store.load({ directory: dir })
|
||||
const first = yield* store.disposeAll().pipe(Effect.forkScoped)
|
||||
yield* Effect.promise(() => disposing.promise)
|
||||
yield* Deferred.await(disposing)
|
||||
const second = yield* store.disposeAll().pipe(Effect.forkScoped)
|
||||
|
||||
expect(disposed).toEqual([dir])
|
||||
releaseDispose.resolve()
|
||||
yield* Deferred.succeed(releaseDispose, undefined)
|
||||
yield* Effect.all([Fiber.join(first), Fiber.join(second)])
|
||||
expect(disposed).toEqual([dir])
|
||||
}),
|
||||
@@ -221,19 +220,19 @@ describe("InstanceStore", () => {
|
||||
}),
|
||||
)
|
||||
|
||||
it.live("provides legacy Promise callers with instance ALS", () =>
|
||||
Effect.gen(function* () {
|
||||
const dir = yield* tmpdirScoped({ git: true })
|
||||
it.instance(
|
||||
"provides legacy Promise callers with instance ALS",
|
||||
() =>
|
||||
Effect.gen(function* () {
|
||||
const test = yield* TestInstance
|
||||
const ctx = yield* InstanceRef
|
||||
if (!ctx) throw new Error("InstanceRef not provided")
|
||||
|
||||
const directory = yield* Effect.promise(() =>
|
||||
WithInstance.provide({
|
||||
directory: dir,
|
||||
fn: () => Instance.directory,
|
||||
}),
|
||||
)
|
||||
const directory = yield* Effect.promise(() => Promise.resolve(Instance.restore(ctx, () => Instance.directory)))
|
||||
|
||||
expect(directory).toBe(dir)
|
||||
expect(() => Instance.current).toThrow()
|
||||
}),
|
||||
expect(directory).toBe(test.directory)
|
||||
expect(() => Instance.current).toThrow()
|
||||
}),
|
||||
{ git: true },
|
||||
)
|
||||
})
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
@@ -2,7 +2,6 @@ import { afterEach, expect } from "bun:test"
|
||||
import { Cause, Effect, Exit, Fiber, Layer } from "effect"
|
||||
import { Question } from "../../src/question"
|
||||
import { Instance } from "../../src/project/instance"
|
||||
import { WithInstance } from "../../src/project/with-instance"
|
||||
import { InstanceRuntime } from "../../src/project/instance-runtime"
|
||||
import { QuestionID } from "../../src/question/schema"
|
||||
import { disposeAllInstances, provideInstance, reloadTestInstance, tmpdirScoped } from "../fixture/fixture"
|
||||
@@ -398,9 +397,8 @@ it.live("pending question rejects on instance dispose", () =>
|
||||
}).pipe(provideInstance(dir), Effect.forkScoped)
|
||||
|
||||
expect(yield* waitForPending(1).pipe(provideInstance(dir))).toHaveLength(1)
|
||||
yield* Effect.promise(() =>
|
||||
WithInstance.provide({ directory: dir, fn: () => InstanceRuntime.disposeInstance(Instance.current) }),
|
||||
)
|
||||
const ctx = yield* Effect.sync(() => Instance.current).pipe(provideInstance(dir))
|
||||
yield* Effect.promise(() => InstanceRuntime.disposeInstance(ctx))
|
||||
|
||||
const exit = yield* Fiber.await(fiber)
|
||||
expect(Exit.isFailure(exit)).toBe(true)
|
||||
|
||||
@@ -1,10 +1,11 @@
|
||||
import { describe, test, expect, beforeAll, afterAll } from "bun:test"
|
||||
import { describe, expect, beforeAll, afterAll } from "bun:test"
|
||||
import { Effect } from "effect"
|
||||
import { Discovery } from "../../src/skill/discovery"
|
||||
import { Global } from "@opencode-ai/core/global"
|
||||
import { Filesystem } from "@/util/filesystem"
|
||||
import { rm } from "fs/promises"
|
||||
import path from "path"
|
||||
import { testEffect } from "../lib/effect"
|
||||
|
||||
let CLOUDFLARE_SKILLS_URL: string
|
||||
let server: ReturnType<typeof Bun.serve>
|
||||
@@ -12,6 +13,7 @@ let downloadCount = 0
|
||||
|
||||
const fixturePath = path.join(import.meta.dir, "../fixture/skills")
|
||||
const cacheDir = path.join(Global.Path.cache, "skills")
|
||||
const it = testEffect(Discovery.defaultLayer)
|
||||
|
||||
beforeAll(async () => {
|
||||
await rm(cacheDir, { recursive: true, force: true })
|
||||
@@ -47,70 +49,85 @@ afterAll(async () => {
|
||||
})
|
||||
|
||||
describe("Discovery.pull", () => {
|
||||
const pull = (url: string) =>
|
||||
Effect.runPromise(Discovery.Service.use((s) => s.pull(url)).pipe(Effect.provide(Discovery.defaultLayer)))
|
||||
|
||||
test("downloads skills from cloudflare url", async () => {
|
||||
const dirs = await pull(CLOUDFLARE_SKILLS_URL)
|
||||
expect(dirs.length).toBeGreaterThan(0)
|
||||
for (const dir of dirs) {
|
||||
expect(dir).toStartWith(cacheDir)
|
||||
const md = path.join(dir, "SKILL.md")
|
||||
expect(await Filesystem.exists(md)).toBe(true)
|
||||
}
|
||||
const pull = Effect.fn("DiscoveryTest.pull")(function* (url: string) {
|
||||
return yield* Discovery.Service.use((s) => s.pull(url))
|
||||
})
|
||||
|
||||
test("url without trailing slash works", async () => {
|
||||
const dirs = await pull(CLOUDFLARE_SKILLS_URL.replace(/\/$/, ""))
|
||||
expect(dirs.length).toBeGreaterThan(0)
|
||||
for (const dir of dirs) {
|
||||
const md = path.join(dir, "SKILL.md")
|
||||
expect(await Filesystem.exists(md)).toBe(true)
|
||||
}
|
||||
})
|
||||
it.live("downloads skills from cloudflare url", () =>
|
||||
Effect.gen(function* () {
|
||||
const dirs = yield* pull(CLOUDFLARE_SKILLS_URL)
|
||||
expect(dirs.length).toBeGreaterThan(0)
|
||||
for (const dir of dirs) {
|
||||
expect(dir).toStartWith(cacheDir)
|
||||
const md = path.join(dir, "SKILL.md")
|
||||
expect(yield* Effect.promise(() => Filesystem.exists(md))).toBe(true)
|
||||
}
|
||||
}),
|
||||
)
|
||||
|
||||
test("returns empty array for invalid url", async () => {
|
||||
const dirs = await pull(`http://localhost:${server.port}/invalid-url/`)
|
||||
expect(dirs).toEqual([])
|
||||
})
|
||||
it.live("url without trailing slash works", () =>
|
||||
Effect.gen(function* () {
|
||||
const dirs = yield* pull(CLOUDFLARE_SKILLS_URL.replace(/\/$/, ""))
|
||||
expect(dirs.length).toBeGreaterThan(0)
|
||||
for (const dir of dirs) {
|
||||
const md = path.join(dir, "SKILL.md")
|
||||
expect(yield* Effect.promise(() => Filesystem.exists(md))).toBe(true)
|
||||
}
|
||||
}),
|
||||
)
|
||||
|
||||
test("returns empty array for non-json response", async () => {
|
||||
// any url not explicitly handled in server returns 404 text "Not Found"
|
||||
const dirs = await pull(`http://localhost:${server.port}/some-other-path/`)
|
||||
expect(dirs).toEqual([])
|
||||
})
|
||||
it.live("returns empty array for invalid url", () =>
|
||||
Effect.gen(function* () {
|
||||
const dirs = yield* pull(`http://localhost:${server.port}/invalid-url/`)
|
||||
expect(dirs).toEqual([])
|
||||
}),
|
||||
)
|
||||
|
||||
test("downloads reference files alongside SKILL.md", async () => {
|
||||
const dirs = await pull(CLOUDFLARE_SKILLS_URL)
|
||||
// find a skill dir that should have reference files (e.g. agents-sdk)
|
||||
const agentsSdk = dirs.find((d) => d.endsWith(path.sep + "agents-sdk"))
|
||||
expect(agentsSdk).toBeDefined()
|
||||
if (agentsSdk) {
|
||||
const refs = path.join(agentsSdk, "references")
|
||||
expect(await Filesystem.exists(path.join(agentsSdk, "SKILL.md"))).toBe(true)
|
||||
// agents-sdk has reference files per the index
|
||||
const refDir = await Array.fromAsync(new Bun.Glob("**/*.md").scan({ cwd: refs, onlyFiles: true }))
|
||||
expect(refDir.length).toBeGreaterThan(0)
|
||||
}
|
||||
})
|
||||
it.live("returns empty array for non-json response", () =>
|
||||
Effect.gen(function* () {
|
||||
// any url not explicitly handled in server returns 404 text "Not Found"
|
||||
const dirs = yield* pull(`http://localhost:${server.port}/some-other-path/`)
|
||||
expect(dirs).toEqual([])
|
||||
}),
|
||||
)
|
||||
|
||||
test("caches downloaded files on second pull", async () => {
|
||||
// clear dir and downloadCount
|
||||
await rm(cacheDir, { recursive: true, force: true })
|
||||
downloadCount = 0
|
||||
it.live("downloads reference files alongside SKILL.md", () =>
|
||||
Effect.gen(function* () {
|
||||
const dirs = yield* pull(CLOUDFLARE_SKILLS_URL)
|
||||
// find a skill dir that should have reference files (e.g. agents-sdk)
|
||||
const agentsSdk = dirs.find((d) => d.endsWith(path.sep + "agents-sdk"))
|
||||
expect(agentsSdk).toBeDefined()
|
||||
if (agentsSdk) {
|
||||
const refs = path.join(agentsSdk, "references")
|
||||
expect(yield* Effect.promise(() => Filesystem.exists(path.join(agentsSdk, "SKILL.md")))).toBe(true)
|
||||
// agents-sdk has reference files per the index
|
||||
const refDir = yield* Effect.promise(() =>
|
||||
Array.fromAsync(new Bun.Glob("**/*.md").scan({ cwd: refs, onlyFiles: true })),
|
||||
)
|
||||
expect(refDir.length).toBeGreaterThan(0)
|
||||
}
|
||||
}),
|
||||
)
|
||||
|
||||
// first pull to populate cache
|
||||
const first = await pull(CLOUDFLARE_SKILLS_URL)
|
||||
expect(first.length).toBeGreaterThan(0)
|
||||
const firstCount = downloadCount
|
||||
expect(firstCount).toBeGreaterThan(0)
|
||||
it.live("caches downloaded files on second pull", () =>
|
||||
Effect.gen(function* () {
|
||||
// clear dir and downloadCount
|
||||
yield* Effect.promise(() => rm(cacheDir, { recursive: true, force: true }))
|
||||
downloadCount = 0
|
||||
|
||||
// second pull should return same results from cache
|
||||
const second = await pull(CLOUDFLARE_SKILLS_URL)
|
||||
expect(second.length).toBe(first.length)
|
||||
expect(second.sort()).toEqual(first.sort())
|
||||
// first pull to populate cache
|
||||
const first = yield* pull(CLOUDFLARE_SKILLS_URL)
|
||||
expect(first.length).toBeGreaterThan(0)
|
||||
const firstCount = downloadCount
|
||||
expect(firstCount).toBeGreaterThan(0)
|
||||
|
||||
// second pull should NOT increment download count
|
||||
expect(downloadCount).toBe(firstCount)
|
||||
})
|
||||
// second pull should return same results from cache
|
||||
const second = yield* pull(CLOUDFLARE_SKILLS_URL)
|
||||
expect(second.length).toBe(first.length)
|
||||
expect(second.sort()).toEqual(first.sort())
|
||||
|
||||
// second pull should NOT increment download count
|
||||
expect(downloadCount).toBe(firstCount)
|
||||
}),
|
||||
)
|
||||
})
|
||||
|
||||
@@ -180,7 +180,7 @@ describe("tool.registry", () => {
|
||||
const promptTools = yield* registry.tools({
|
||||
providerID: ProviderID.opencode,
|
||||
modelID: ModelID.make("test"),
|
||||
agent: yield* agents.get(yield* agents.defaultAgent()),
|
||||
agent: yield* agents.defaultInfo(),
|
||||
})
|
||||
const promptTool = promptTools.find((tool) => tool.id === "sql")
|
||||
if (!promptTool) throw new Error("custom sql tool was not returned for prompts")
|
||||
|
||||
@@ -224,7 +224,7 @@ describe("Truncate", () => {
|
||||
)
|
||||
|
||||
test("loads truncate effect in a fresh process", async () => {
|
||||
const out = await Process.run([process.execPath, "run", path.join(ROOT, "src", "tool", "truncate.ts")], {
|
||||
const out = await Process.runPromise([process.execPath, "run", path.join(ROOT, "src", "tool", "truncate.ts")], {
|
||||
cwd: ROOT,
|
||||
})
|
||||
|
||||
|
||||
@@ -10,19 +10,19 @@ function node(script: string) {
|
||||
|
||||
describe("util.process", () => {
|
||||
test("captures stdout and stderr", async () => {
|
||||
const out = await Process.run(node('process.stdout.write("out");process.stderr.write("err")'))
|
||||
const out = await Process.runPromise(node('process.stdout.write("out");process.stderr.write("err")'))
|
||||
expect(out.code).toBe(0)
|
||||
expect(out.stdout.toString()).toBe("out")
|
||||
expect(out.stderr.toString()).toBe("err")
|
||||
})
|
||||
|
||||
test("returns code when nothrow is enabled", async () => {
|
||||
const out = await Process.run(node("process.exit(7)"), { nothrow: true })
|
||||
const out = await Process.runPromise(node("process.exit(7)"), { nothrow: true })
|
||||
expect(out.code).toBe(7)
|
||||
})
|
||||
|
||||
test("throws RunFailedError on non-zero exit", async () => {
|
||||
const err = await Process.run(node('process.stderr.write("bad");process.exit(3)')).catch((error) => error)
|
||||
const err = await Process.runPromise(node('process.stderr.write("bad");process.exit(3)')).catch((error) => error)
|
||||
expect(err).toBeInstanceOf(Process.RunFailedError)
|
||||
if (!(err instanceof Process.RunFailedError)) throw err
|
||||
expect(err.code).toBe(3)
|
||||
@@ -34,7 +34,7 @@ describe("util.process", () => {
|
||||
const started = Date.now()
|
||||
setTimeout(() => abort.abort(), 25)
|
||||
|
||||
const out = await Process.run(node("setInterval(() => {}, 1000)"), {
|
||||
const out = await Process.runPromise(node("setInterval(() => {}, 1000)"), {
|
||||
abort: abort.signal,
|
||||
nothrow: true,
|
||||
})
|
||||
@@ -50,7 +50,7 @@ describe("util.process", () => {
|
||||
const started = Date.now()
|
||||
setTimeout(() => abort.abort(), 25)
|
||||
|
||||
const out = await Process.run(node('process.on("SIGTERM", () => {}); setInterval(() => {}, 1000)'), {
|
||||
const out = await Process.runPromise(node('process.on("SIGTERM", () => {}); setInterval(() => {}, 1000)'), {
|
||||
abort: abort.signal,
|
||||
nothrow: true,
|
||||
timeout: 25,
|
||||
@@ -62,14 +62,14 @@ describe("util.process", () => {
|
||||
|
||||
test("uses cwd when spawning commands", async () => {
|
||||
await using tmp = await tmpdir()
|
||||
const out = await Process.run(node("process.stdout.write(process.cwd())"), {
|
||||
const out = await Process.runPromise(node("process.stdout.write(process.cwd())"), {
|
||||
cwd: tmp.path,
|
||||
})
|
||||
expect(out.stdout.toString()).toBe(tmp.path)
|
||||
})
|
||||
|
||||
test("merges environment overrides", async () => {
|
||||
const out = await Process.run(node('process.stdout.write(process.env.OPENCODE_TEST ?? "")'), {
|
||||
const out = await Process.runPromise(node('process.stdout.write(process.env.OPENCODE_TEST ?? "")'), {
|
||||
env: {
|
||||
OPENCODE_TEST: "set",
|
||||
},
|
||||
@@ -80,7 +80,7 @@ describe("util.process", () => {
|
||||
test("uses shell in run on Windows", async () => {
|
||||
if (process.platform !== "win32") return
|
||||
|
||||
const out = await Process.run(["set", "OPENCODE_TEST_SHELL"], {
|
||||
const out = await Process.runPromise(["set", "OPENCODE_TEST_SHELL"], {
|
||||
shell: true,
|
||||
env: {
|
||||
OPENCODE_TEST_SHELL: "ok",
|
||||
|
||||
Reference in New Issue
Block a user