Compare commits

...
44 changed files with 857 additions and 337 deletions

No files matched your search

+39 -40
View File
@@ -584,7 +584,7 @@ const serverToolResultType = (name: string): AnthropicServerToolResultType | und
return undefined
}
const lowerServerToolResult = Effect.fn("AnthropicMessages.lowerServerToolResult")(function* (
const lowerServerToolResult = Effect.fnUntraced(function* (
part: ToolResultPart,
providerMetadataKey: string,
) {
@@ -657,7 +657,7 @@ const citationsFromMetadata = (metadata: MediaPart["metadata"]): AnthropicDocume
const isHttpUrl = (value: string) => /^https?:\/\//i.test(value.trim())
const lowerMedia = Effect.fn("AnthropicMessages.lowerMedia")(function* (
const lowerMedia = Effect.fnUntraced(function* (
part: MediaPart,
breakpoints?: Cache.Breakpoints,
) {
@@ -847,7 +847,7 @@ const splitsLocalToolResults = (messages: LLMRequest["messages"], index: number)
return pending.size > 0
}
const lowerNativeSystemUpdate = Effect.fn("AnthropicMessages.lowerNativeSystemUpdate")(function* (
const lowerNativeSystemUpdate = Effect.fnUntraced(function* (
message: LLMRequest["messages"][number],
breakpoints: Cache.Breakpoints,
) {
@@ -862,7 +862,7 @@ const lowerNativeSystemUpdate = Effect.fn("AnthropicMessages.lowerNativeSystemUp
}
})
const lowerMessages = Effect.fn("AnthropicMessages.lowerMessages")(function* (
const lowerMessages = Effect.fnUntraced(function* (
request: LLMRequest,
breakpoints: Cache.Breakpoints,
) {
@@ -1300,32 +1300,29 @@ const onContentBlockStart = (
return [{ ...state, lifecycle: Lifecycle.stepStart(state.lifecycle, events) }, [...events, result]]
}
const onContentBlockDelta = Effect.fn("AnthropicMessages.onContentBlockDelta")(function* (
const onContentBlockDelta = (
state: ParserState,
event: AnthropicEvent & { readonly delta: AnthropicStreamDelta },
) {
): StepResult | AIError => {
const delta = event.delta
if (delta.type === "compaction_delta") {
if (event.index === undefined || !(event.index in state.compactions) || delta.content === undefined)
return yield* ProviderShared.eventError(ADAPTER, "Compaction delta is missing its block or content")
return [
{ ...state, compactions: { ...state.compactions, [event.index]: delta.content } },
NO_EVENTS,
] satisfies StepResult
return ProviderShared.eventError(ADAPTER, "Compaction delta is missing its block or content")
return [{ ...state, compactions: { ...state.compactions, [event.index]: delta.content } }, NO_EVENTS]
}
if (delta.type === "text_delta" && delta.text) {
if (!state.lifecycle.text.has(`text-${event.index ?? 0}`)) return [state, NO_EVENTS] satisfies StepResult
if (!state.lifecycle.text.has(`text-${event.index ?? 0}`)) return [state, NO_EVENTS]
const events: LLMEvent[] = []
return [
{ ...state, lifecycle: Lifecycle.textDelta(state.lifecycle, events, `text-${event.index ?? 0}`, delta.text) },
events,
] satisfies StepResult
]
}
if (delta.type === "thinking_delta" && delta.thinking) {
if (!state.lifecycle.reasoning.has(`reasoning-${event.index ?? 0}`)) return [state, NO_EVENTS] satisfies StepResult
if (!state.lifecycle.reasoning.has(`reasoning-${event.index ?? 0}`)) return [state, NO_EVENTS]
const events: LLMEvent[] = []
return [
{
@@ -1333,24 +1330,24 @@ const onContentBlockDelta = Effect.fn("AnthropicMessages.onContentBlockDelta")(f
lifecycle: Lifecycle.reasoningDelta(state.lifecycle, events, `reasoning-${event.index ?? 0}`, delta.thinking),
},
events,
] satisfies StepResult
]
}
if (delta.type === "signature_delta" && delta.signature) {
const index = event.index ?? 0
if (!state.lifecycle.reasoning.has(`reasoning-${index}`)) return [state, NO_EVENTS] satisfies StepResult
if (!state.lifecycle.reasoning.has(`reasoning-${index}`)) return [state, NO_EVENTS]
return [
{
...state,
reasoningSignatures: { ...state.reasoningSignatures, [index]: delta.signature },
},
NO_EVENTS,
] satisfies StepResult
]
}
if (delta.type === "input_json_delta" && event.index !== undefined) {
if (!delta.partial_json) return [state, NO_EVENTS] satisfies StepResult
if (!state.tools[event.index]) return [state, NO_EVENTS] satisfies StepResult
if (!delta.partial_json) return [state, NO_EVENTS]
if (!state.tools[event.index]) return [state, NO_EVENTS]
const result = ToolStream.appendExisting(
ADAPTER,
state.tools,
@@ -1358,21 +1355,18 @@ const onContentBlockDelta = Effect.fn("AnthropicMessages.onContentBlockDelta")(f
delta.partial_json,
"Anthropic Messages tool argument delta is missing its tool call",
)
if (ToolStream.isError(result)) return yield* result
if (ToolStream.isError(result)) return result
const events: LLMEvent[] = []
const lifecycle = result.events.length ? Lifecycle.stepStart(state.lifecycle, events) : state.lifecycle
events.push(...result.events)
return [{ ...state, lifecycle, tools: result.tools }, events] satisfies StepResult
return [{ ...state, lifecycle, tools: result.tools }, events]
}
return [state, NO_EVENTS] satisfies StepResult
})
return [state, NO_EVENTS]
}
const onContentBlockStop = Effect.fn("AnthropicMessages.onContentBlockStop")(function* (
state: ParserState,
event: AnthropicEvent,
) {
if (event.index === undefined) return [state, NO_EVENTS] satisfies StepResult
const onContentBlockStop = (state: ParserState, event: AnthropicEvent): StepResult | AIError => {
if (event.index === undefined) return [state, NO_EVENTS]
if (event.index in state.compactions) {
const { [event.index]: content, ...compactions } = state.compactions
const events: LLMEvent[] = []
@@ -1383,9 +1377,10 @@ const onContentBlockStop = Effect.fn("AnthropicMessages.onContentBlockStop")(fun
text: content,
}),
)
return [{ ...state, compactions, lifecycle }, events] satisfies StepResult
return [{ ...state, compactions, lifecycle }, events]
}
const result = yield* ToolStream.finish(ADAPTER, state.tools, event.index)
const result = ToolStream.finish(ADAPTER, state.tools, event.index)
if (ToolStream.isError(result)) return result
const events: LLMEvent[] = []
const resultEvents = result.events ?? []
const signature = state.reasoningSignatures[event.index]
@@ -1400,8 +1395,8 @@ const onContentBlockStop = Effect.fn("AnthropicMessages.onContentBlockStop")(fun
events.push(...resultEvents)
const reasoningSignatures = { ...state.reasoningSignatures }
delete reasoningSignatures[event.index]
return [{ ...state, lifecycle, tools: result.tools, reasoningSignatures }, events] satisfies StepResult
})
return [{ ...state, lifecycle, tools: result.tools, reasoningSignatures }, events]
}
const onMessageDelta = (
state: ParserState,
@@ -1439,10 +1434,11 @@ const onMessageDelta = (
]
}
const onMessageStop = Effect.fn("AnthropicMessages.onMessageStop")(function* (state: ParserState) {
const onMessageStop = (state: ParserState): StepResult | AIError => {
if (Object.keys(state.compactions).length)
return yield* ProviderShared.eventError(ADAPTER, "Response ended with an incomplete compaction block")
const result = yield* ToolStream.finishAll(ADAPTER, state.tools)
return ProviderShared.eventError(ADAPTER, "Response ended with an incomplete compaction block")
const result = ToolStream.finishAll(ADAPTER, state.tools)
if (ToolStream.isError(result)) return result
const events: LLMEvent[] = []
const lifecycle = result.events.length ? Lifecycle.stepStart(state.lifecycle, events) : state.lifecycle
events.push(...result.events)
@@ -1464,8 +1460,8 @@ const onMessageStop = Effect.fn("AnthropicMessages.onMessageStop")(function* (st
usage: state.usage,
providerMetadata: state.pendingFinish?.providerMetadata,
})
return [{ ...state, lifecycle: finished, tools: result.tools }, events] satisfies StepResult
})
return [{ ...state, lifecycle: finished, tools: result.tools }, events]
}
// Prefix `error.type` so overloads, rate limits, and quota errors are visible
// even when the provider message is generic or empty.
@@ -1511,6 +1507,9 @@ const invalidStreamEvent = (event: AnthropicEvent) =>
),
)
const stepOutcome = (result: StepResult | AIError) =>
result instanceof AIError ? Effect.fail(result) : Effect.succeed(result)
const step = (state: ParserState, event: AnthropicEvent) => {
if (!SSE_EVENTS.has(event.type)) return Effect.succeed<StepResult>([state, NO_EVENTS])
if (
@@ -1556,15 +1555,15 @@ const step = (state: ParserState, event: AnthropicEvent) => {
return Effect.succeed<StepResult>([state, NO_EVENTS])
const decoded = decodeAnthropicStreamDelta(event.delta)
if (Option.isNone(decoded)) return invalidStreamEvent(event)
return onContentBlockDelta(state, { ...event, delta: decoded.value })
return stepOutcome(onContentBlockDelta(state, { ...event, delta: decoded.value }))
}
if (event.type === "content_block_stop") return onContentBlockStop(state, event)
if (event.type === "content_block_stop") return stepOutcome(onContentBlockStop(state, event))
if (event.type === "message_delta") {
const decoded = decodeAnthropicStreamDelta(event.delta)
if (Option.isNone(decoded)) return invalidStreamEvent(event)
return Effect.succeed(onMessageDelta(state, { ...event, delta: decoded.value }))
}
if (event.type === "message_stop") return onMessageStop(state)
if (event.type === "message_stop") return stepOutcome(onMessageStop(state))
if (event.type === "error") return onError(event)
return Effect.succeed<StepResult>([state, NO_EVENTS])
}
@@ -285,7 +285,7 @@ const lowerToolCall = (part: ToolCallPart, normalizeID: (id: string) => string):
},
})
const lowerToolResultContent = Effect.fn("BedrockConverse.lowerToolResultContent")(function* (
const lowerToolResultContent = Effect.fnUntraced(function* (
part: ToolResultPart,
documentNames: Set<string>,
) {
@@ -305,7 +305,7 @@ const lowerToolResultContent = Effect.fn("BedrockConverse.lowerToolResultContent
return content
})
const lowerToolResult = Effect.fn("BedrockConverse.lowerToolResult")(function* (
const lowerToolResult = Effect.fnUntraced(function* (
part: ToolResultPart,
documentNames: Set<string>,
normalizeID: (id: string) => string,
@@ -322,7 +322,7 @@ const lowerToolResult = Effect.fn("BedrockConverse.lowerToolResult")(function* (
// Keep Claude and Nova tool-result images inline; put other models' images beside the result.
const keepToolImagesInline = (id: string) => id.includes("anthropic.claude-") || id.includes("amazon.nova-")
const lowerMessages = Effect.fn("BedrockConverse.lowerMessages")(function* (
const lowerMessages = Effect.fnUntraced(function* (
request: LLMRequest,
breakpoints: BedrockCache.Breakpoints,
) {
@@ -722,7 +722,8 @@ const step = (state: ParserState, event: BedrockEvent) =>
if (event.contentBlockStop) {
const index = event.contentBlockStop.contentBlockIndex
const result = yield* ToolStream.finish(ADAPTER, state.tools, index)
const result = ToolStream.finish(ADAPTER, state.tools, index)
if (ToolStream.isError(result)) return yield* result
const events: LLMEvent[] = []
const resultEvents = result.events ?? []
const lifecycle = (() => {
+5 -3
View File
@@ -252,7 +252,7 @@ const mapUsage = (usage: typeof NativeUsage.Type) =>
})
// Lifecycle deltas open blocks on demand and ends are no-ops for closed blocks, so content-start needs no handling.
const step = Effect.fn("CohereChat.step")(function* (state: State, event: Event) {
const step = Effect.fnUntraced(function* (state: State, event: Event) {
const events: LLMEvent[] = []
switch (event.type) {
case "message-start":
@@ -292,11 +292,13 @@ const step = Effect.fn("CohereChat.step")(function* (state: State, event: Event)
return [{ ...state, tools: result.tools }, result.events] as const
}
case "tool-call-end": {
const result = yield* ToolStream.finish(ADAPTER, state.tools, event.index)
const result = ToolStream.finish(ADAPTER, state.tools, event.index)
if (ToolStream.isError(result)) return yield* result
return [{ ...state, tools: result.tools }, result.events ?? []] as const
}
case "message-end": {
const pending = yield* ToolStream.finishAll(ADAPTER, state.tools)
const pending = ToolStream.finishAll(ADAPTER, state.tools)
if (ToolStream.isError(pending)) return yield* pending
events.push(...pending.events)
const lifecycle = Lifecycle.finish(state.lifecycle, events, {
reason: finishReason(event.delta.finish_reason),
+2 -2
View File
@@ -283,7 +283,7 @@ const lowerToolConfig = (toolChoice: NonNullable<LLMRequest["toolChoice"]>) =>
tool: (name) => ({ functionCallingConfig: { mode: "ANY" as const, allowedFunctionNames: [name] } }),
})
const lowerContentPart = Effect.fn("Gemini.lowerContentPart")(function* (part: TextPart | MediaPart) {
const lowerContentPart = Effect.fnUntraced(function* (part: TextPart | MediaPart) {
if (part.type === "text") return { text: part.text }
return yield* GeminiGenerateContent.mediaPart("Gemini", part.media)
})
@@ -302,7 +302,7 @@ const lowerToolCall = (part: ToolCallPart, omitIds: boolean, metadataKey: string
thoughtSignature: thoughtSignature(part.providerMetadata, metadataKey),
})
const lowerMessages = Effect.fn("Gemini.lowerMessages")(function* (request: LLMRequest) {
const lowerMessages = Effect.fnUntraced(function* (request: LLMRequest) {
const contents: GeminiContent[] = []
const metadataKey = request.model.route.providerMetadataKey ?? String(request.model.provider)
const omitCallIds = omitsFunctionCallIds(request.model.id)
@@ -182,7 +182,7 @@ type StepResult = readonly [ParserState, ReadonlyArray<LLMEvent>]
// =============================================================================
// Request Body Construction
// =============================================================================
const mediaContent = Effect.fn("GoogleInteractions.mediaContent")(function* (asset: Media.Asset) {
const mediaContent = Effect.fnUntraced(function* (asset: Media.Asset) {
if (
asset.kind !== "image" &&
asset.kind !== "audio" &&
@@ -208,7 +208,7 @@ const signature = (metadata: ProviderMetadata | undefined, key: string) => {
: undefined
}
const lowerMessages = Effect.fn("GoogleInteractions.lowerMessages")(function* (request: LLMRequest) {
const lowerMessages = Effect.fnUntraced(function* (request: LLMRequest) {
const steps: InputStep[] = []
const key = request.model.route.providerMetadataKey ?? String(request.model.provider)
for (const message of request.messages) {
@@ -275,7 +275,7 @@ const lowerMessages = Effect.fn("GoogleInteractions.lowerMessages")(function* (r
return steps
})
const lowerToolResult = Effect.fn("GoogleInteractions.lowerToolResult")(function* (part: ToolResultPart) {
const lowerToolResult = Effect.fnUntraced(function* (part: ToolResultPart) {
if (part.result.type === "json" && ProviderShared.isRecord(part.result.value)) return part.result.value
if (part.result.type !== "content") return ProviderShared.toolResultText(part)
@@ -364,7 +364,7 @@ const mapUsage = (usage: RawUsage | undefined, key: string) => {
})
}
const onStart = Effect.fn("GoogleInteractions.onStart")(function* (
const onStart = Effect.fnUntraced(function* (
state: ParserState,
index: number,
step: OutputStep,
@@ -402,7 +402,7 @@ const onStart = Effect.fn("GoogleInteractions.onStart")(function* (
return [{ ...state, lifecycle, tools, steps: { ...state.steps, [index]: step } }, events] satisfies StepResult
})
const onDelta = Effect.fn("GoogleInteractions.onDelta")(function* (
const onDelta = Effect.fnUntraced(function* (
state: ParserState,
index: number,
delta: typeof Delta.Type,
@@ -447,7 +447,7 @@ const onDelta = Effect.fn("GoogleInteractions.onDelta")(function* (
return yield* ProviderShared.eventError(ADAPTER, `Unsupported Interactions delta: ${delta.type}`, encodeJson(delta))
})
const onStop = Effect.fn("GoogleInteractions.onStop")(function* (state: ParserState, index: number) {
const onStop = Effect.fnUntraced(function* (state: ParserState, index: number) {
const step = state.steps[index]
if (!step) return yield* ProviderShared.eventError(ADAPTER, "Interactions step.stop without step.start")
const events: LLMEvent[] = []
@@ -461,11 +461,12 @@ const onStop = Effect.fn("GoogleInteractions.onStop")(function* (state: ParserSt
{ ...state, lifecycle: Lifecycle.textEnd(state.lifecycle, events, String(index)) },
events,
] satisfies StepResult
const result = yield* ToolStream.finish(ADAPTER, state.tools, index)
const result = ToolStream.finish(ADAPTER, state.tools, index)
if (ToolStream.isError(result)) return yield* result
return [{ ...state, tools: result.tools }, result.events ?? []] satisfies StepResult
})
const step = Effect.fn("GoogleInteractions.step")(function* (state: ParserState, event: Event) {
const step = Effect.fnUntraced(function* (state: ParserState, event: Event) {
switch (event.event_type) {
case "step.start":
return yield* onStart(state, event.index, event.step)
@@ -500,7 +501,8 @@ const step = Effect.fn("GoogleInteractions.step")(function* (state: ParserState,
`Unexpected terminal Interactions status: ${interaction.status}`,
encodeJson(event),
)
const pending = yield* ToolStream.finishAll(ADAPTER, state.tools)
const pending = ToolStream.finishAll(ADAPTER, state.tools)
if (ToolStream.isError(pending)) return yield* pending
const events = [...pending.events]
const lifecycle = Lifecycle.finish(state.lifecycle, events, {
reason: {
+3 -3
View File
@@ -95,7 +95,7 @@ const HOSTED_TOOLS = {
image_generation_call: {
name: "image_generation",
input: () => ({}),
result: Effect.fn("MetaResponses.imageResult")(function* (raw: ResponsesHostedTools.Item) {
result: Effect.fnUntraced(function* (raw: ResponsesHostedTools.Item) {
const item = yield* Schema.decodeUnknownEffect(ImageItem)(raw).pipe(
Effect.mapError((cause) =>
ProviderShared.eventError(
@@ -136,7 +136,7 @@ const HOSTED_TOOLS = {
},
} satisfies ResponsesHostedTools.Definitions
const onEvent = Effect.fn("MetaResponses.onEvent")(function* (
const onEvent = Effect.fnUntraced(function* (
state: OpenResponses.ParserState,
input: OpenResponses.Event,
) {
@@ -173,7 +173,7 @@ const onEvent = Effect.fn("MetaResponses.onEvent")(function* (
] satisfies OpenResponses.StepResult
})
const step = Effect.fn("MetaResponses.step")(function* (state: ParserState, input: OpenResponses.Event) {
const step = Effect.fnUntraced(function* (state: ParserState, input: OpenResponses.Event) {
const completedItems = new Set(state.completedItems)
const event = OpenResponses.normalize(state, input)
if (event.type === "response.output_item.done" && event.item && completedItems.has(event.item.id))
+10 -9
View File
@@ -226,7 +226,7 @@ const MistralEvent = Schema.StructWithRest(
type MistralEvent = Schema.Schema.Type<typeof MistralEvent>
const MistralStreamEvent = Schema.Union([Schema.Literal(DONE), Protocol.jsonEvent(MistralEvent)])
const lowerMedia = Effect.fn("MistralChat.lowerMedia")(function* (part: MediaPart) {
const lowerMedia = Effect.fnUntraced(function* (part: MediaPart) {
const mime = part.media.mediaType.toLowerCase()
const url =
ProviderShared.mediaUrl(part.media) ??
@@ -236,7 +236,7 @@ const lowerMedia = Effect.fn("MistralChat.lowerMedia")(function* (part: MediaPar
return yield* ProviderShared.invalidRequest(`Mistral Chat does not support media type ${part.media.mediaType}`)
})
const lowerUser = Effect.fn("MistralChat.lowerUser")(function* (message: LLMRequest["messages"][number]) {
const lowerUser = Effect.fnUntraced(function* (message: LLMRequest["messages"][number]) {
const content: MistralUserContent[] = []
for (const part of message.content) {
if (part.type === "text") {
@@ -260,7 +260,7 @@ const lowerToolCall = (part: ToolCallPart, normalizeID: (id: string) => string):
function: { name: part.name, arguments: ProviderShared.encodeJson(part.input) },
})
const lowerAssistant = Effect.fn("MistralChat.lowerAssistant")(function* (
const lowerAssistant = Effect.fnUntraced(function* (
message: LLMRequest["messages"][number],
normalizeID: (id: string) => string,
prefix: boolean,
@@ -298,7 +298,7 @@ const lowerAssistant = Effect.fn("MistralChat.lowerAssistant")(function* (
}
})
const lowerToolResults = Effect.fn("MistralChat.lowerToolResults")(function* (
const lowerToolResults = Effect.fnUntraced(function* (
message: LLMRequest["messages"][number],
normalizeID: (id: string) => string,
) {
@@ -335,7 +335,7 @@ const lowerToolResults = Effect.fn("MistralChat.lowerToolResults")(function* (
return output
})
const lowerMessages = Effect.fn("MistralChat.lowerMessages")(function* (request: LLMRequest) {
const lowerMessages = Effect.fnUntraced(function* (request: LLMRequest) {
const normalizeID = MistralToolID.normalizer(request)
const messages: MistralMessage[] =
request.system.length === 0
@@ -596,7 +596,7 @@ const toolText = (tool: MistralToolDelta) => {
return value === null || value === undefined ? "" : ProviderShared.encodeJson(value)
}
const appendTools = Effect.fn("MistralChat.appendTools")(function* (
const appendTools = Effect.fnUntraced(function* (
initial: ParserState,
events: LLMEvent[],
deltas: ReadonlyArray<MistralToolDelta>,
@@ -662,7 +662,7 @@ const hasLateContent = (event: MistralEvent) => {
)
}
const step = Effect.fn("MistralChat.step")(function* (state: ParserState, event: MistralEvent) {
const step = Effect.fnUntraced(function* (state: ParserState, event: MistralEvent) {
if (event.error) {
const body = ProviderShared.encodeJson(event)
return yield* new AIError({
@@ -712,8 +712,9 @@ const step = Effect.fn("MistralChat.step")(function* (state: ParserState, event:
)
const finished =
!incomplete && Object.keys(withTools.tools).length > 0
? yield* ToolStream.finishAll(ADAPTER, withTools.tools)
? ToolStream.finishAll(ADAPTER, withTools.tools)
: undefined
if (ToolStream.isError(finished)) return yield* finished
return [
{
...withTools,
@@ -726,7 +727,7 @@ const step = Effect.fn("MistralChat.step")(function* (state: ParserState, event:
] as const
})
const finishEvents = Effect.fn("MistralChat.finishEvents")(function* (state: ParserState) {
const finishEvents = Effect.fnUntraced(function* (state: ParserState) {
if (!state.finishReason)
return yield* new AIError({
reason: new InvalidProviderOutputError({
+51 -59
View File
@@ -442,7 +442,7 @@ interface ReasoningStreamItem {
// =============================================================================
// Request Lowering
// =============================================================================
export const lowerTool = Effect.fn("OpenResponses.lowerTool")(function* (protocolName: string, tool: ToolDefinition) {
export const lowerTool = Effect.fnUntraced(function* (protocolName: string, tool: ToolDefinition) {
if (tool.native !== undefined)
return yield* ProviderShared.invalidRequest(`${protocolName} does not support provider-native tool ${tool.name}`)
return {
@@ -507,7 +507,10 @@ const lowerReasoning = (part: ReasoningPart, providerMetadataKey: string): OpenR
}
}
const lowerMedia = Effect.fn("OpenResponses.lowerMedia")(function* (
const decodeImageDetail = ProviderShared.validateWith(Schema.decodeUnknownEffect(OpenResponsesInputImage.fields.detail))
const decodeMessageMetadata = ProviderShared.validateWith(Schema.decodeUnknownEffect(MessageMetadata))
const lowerMedia = Effect.fnUntraced(function* (
part: MediaPart,
request: LLMRequest,
adapter: ProviderAdapter,
@@ -516,9 +519,8 @@ const lowerMedia = Effect.fn("OpenResponses.lowerMedia")(function* (
const media = part.media.inline()
const providerMedia = adapter.lowerMedia?.({ part, media, request })
if (providerMedia) return providerMedia
const detail = yield* ProviderShared.validateWith(Schema.decodeUnknownEffect(OpenResponsesInputImage.fields.detail))(
part.providerMetadata?.[metadataKey(request.model)]?.detail,
)
const rawDetail = part.providerMetadata?.[metadataKey(request.model)]?.detail
const detail = rawDetail === undefined ? undefined : yield* decodeImageDetail(rawDetail)
const mime = part.media.mediaType.toLowerCase()
const url = ProviderShared.mediaUrl(part.media)
const location = url ?? (yield* ProviderShared.requireInlineMedia(adapter.name, part.media)).dataUrl
@@ -591,7 +593,7 @@ const lowerToolResultOutput = Effect.fnUntraced(function* (
const DEFAULT_EFFORT = "medium"
const lowerMessages = Effect.fn("OpenResponses.lowerMessages")(function* (
const lowerMessages = Effect.fnUntraced(function* (
request: LLMRequest,
adapter: ProviderAdapter,
) {
@@ -599,9 +601,8 @@ const lowerMessages = Effect.fn("OpenResponses.lowerMessages")(function* (
const providerMetadataKey = metadataKey(request.model)
for (const message of request.messages) {
const metadata = yield* ProviderShared.validateWith(
Schema.decodeUnknownEffect(Schema.UndefinedOr(MessageMetadata)),
)(message.providerMetadata?.[providerMetadataKey])
const rawMetadata = message.providerMetadata?.[providerMetadataKey]
const metadata = rawMetadata === undefined ? undefined : yield* decodeMessageMetadata(rawMetadata)
if (message.role === "system") {
const update = effortUpdate(message)
if (update) {
@@ -755,7 +756,7 @@ const lowerMessages = Effect.fn("OpenResponses.lowerMessages")(function* (
return input
})
export const lowerConversation = Effect.fn("OpenResponses.lowerConversation")(function* (
export const lowerConversation = Effect.fnUntraced(function* (
request: LLMRequest,
adapter: ProviderAdapter,
) {
@@ -1142,23 +1143,16 @@ const onReasoningSummaryPartDone = (state: ParserState, event: Event): StepResul
]
}
const onFunctionCallArgumentsDelta = Effect.fn("OpenResponses.onFunctionCallArgumentsDelta")(function* (
state: ParserState,
event: Event,
) {
if (event.item_id === undefined) return [state, NO_EVENTS] satisfies StepResult
const onFunctionCallArgumentsDelta = (state: ParserState, event: Event): StepResult | AIError => {
if (event.item_id === undefined) return [state, NO_EVENTS]
const tool = state.tools[event.item_id]
if (!tool) return [state, NO_EVENTS] satisfies StepResult
if (!tool) return [state, NO_EVENTS]
const final = event.type === "response.function_call_arguments.done" ? event.arguments : undefined
if (event.type === "response.function_call_arguments.done" && final === undefined)
return [state, NO_EVENTS] satisfies StepResult
if (event.type === "response.function_call_arguments.done" && final === undefined) return [state, NO_EVENTS]
if (final !== undefined && !final.startsWith(tool.input))
return [
{ ...state, tools: ToolStream.start(state.tools, event.item_id, { ...tool, input: final }) },
NO_EVENTS,
] satisfies StepResult
return [{ ...state, tools: ToolStream.start(state.tools, event.item_id, { ...tool, input: final }) }, NO_EVENTS]
const delta = final === undefined ? event.delta : final.slice(tool.input.length)
if (!delta) return [state, NO_EVENTS] satisfies StepResult
if (!delta) return [state, NO_EVENTS]
const result = ToolStream.appendExisting(
state.id,
state.tools,
@@ -1166,23 +1160,20 @@ const onFunctionCallArgumentsDelta = Effect.fn("OpenResponses.onFunctionCallArgu
delta,
`${state.name} tool argument delta is missing its tool call`,
)
if (ToolStream.isError(result)) return yield* result
if (ToolStream.isError(result)) return result
const events: LLMEvent[] = []
const lifecycle = result.events.length ? Lifecycle.stepStart(state.lifecycle, events) : state.lifecycle
events.push(...result.events)
return [{ ...state, lifecycle, tools: result.tools }, events] satisfies StepResult
})
return [{ ...state, lifecycle, tools: result.tools }, events]
}
const onOutputItemDone = Effect.fn("OpenResponses.onOutputItemDone")(function* (
state: ParserState,
item: NormalizedEvent["item"],
) {
if (!item) return [state, NO_EVENTS] satisfies StepResult
const onOutputItemDone = (state: ParserState, item: NormalizedEvent["item"]): StepResult | AIError => {
if (!item) return [state, NO_EVENTS]
if (item.type === "compaction") {
if (typeof item.encrypted_content !== "string")
return yield* ProviderShared.eventError(state.id, "Compaction output is missing its encrypted content")
if (state.completedCompactions.has(item.id)) return [state, NO_EVENTS] satisfies StepResult
return ProviderShared.eventError(state.id, "Compaction output is missing its encrypted content")
if (state.completedCompactions.has(item.id)) return [state, NO_EVENTS]
const events: LLMEvent[] = []
const lifecycle = Lifecycle.stepStart(state.lifecycle, events)
events.push(
@@ -1192,10 +1183,7 @@ const onOutputItemDone = Effect.fn("OpenResponses.onOutputItemDone")(function* (
encrypted: item.encrypted_content,
}),
)
return [
{ ...state, lifecycle, completedCompactions: new Set([...state.completedCompactions, item.id]) },
events,
] satisfies StepResult
return [{ ...state, lifecycle, completedCompactions: new Set([...state.completedCompactions, item.id]) }, events]
}
if (item.type === "message") {
@@ -1220,11 +1208,11 @@ const onOutputItemDone = Effect.fn("OpenResponses.onOutputItemDone")(function* (
message: active ? undefined : state.message,
},
events,
] satisfies StepResult
]
}
if (item.type === "function_call") {
if (!item.call_id || !item.name) return [state, NO_EVENTS] satisfies StepResult
if (!item.call_id || !item.name) return [state, NO_EVENTS]
const metadata = providerMetadata(state, { itemId: item.id })
const registered = state.tools[item.id] !== undefined
const tools = registered
@@ -1235,10 +1223,8 @@ const onOutputItemDone = Effect.fn("OpenResponses.onOutputItemDone")(function* (
namespace: item.namespace,
providerMetadata: metadata,
})
const result =
item.arguments === undefined
? yield* ToolStream.finish(state.id, tools, item.id)
: yield* ToolStream.finishWithInput(state.id, tools, item.id, item.arguments)
const result = ToolStream.finish(state.id, tools, item.id, item.arguments)
if (ToolStream.isError(result)) return result
const events: LLMEvent[] = []
const finished = result.events ?? []
// A done-only call never streamed a start event, so open its lifecycle here.
@@ -1266,7 +1252,7 @@ const onOutputItemDone = Effect.fn("OpenResponses.onOutputItemDone")(function* (
tools: result.tools,
},
events,
] satisfies StepResult
]
}
if (item.type === "reasoning") {
@@ -1298,18 +1284,18 @@ const onOutputItemDone = Effect.fn("OpenResponses.onOutputItemDone")(function* (
}
const reasoningItems = { ...state.reasoningItems }
delete reasoningItems[item.id]
return [{ ...state, lifecycle, reasoningItems }, events] satisfies StepResult
return [{ ...state, lifecycle, reasoningItems }, events]
}
const lifecycle = Lifecycle.stepStart(state.lifecycle, events)
events.push(LLMEvent.reasoningStart({ id: item.id, providerMetadata: metadata }))
events.push(LLMEvent.reasoningEnd({ id: item.id, providerMetadata: metadata, text: itemText }))
return [{ ...state, lifecycle }, events] satisfies StepResult
return [{ ...state, lifecycle }, events]
}
return [state, NO_EVENTS] satisfies StepResult
})
return [state, NO_EVENTS]
}
const onResponseFinish = Effect.fn("OpenResponses.onResponseFinish")(function* (state: ParserState, event: Event) {
const onResponseFinish = (state: ParserState, event: Event): StepResult | AIError => {
let current = state
const events: LLMEvent[] = []
if (event.type === "response.completed") {
@@ -1317,19 +1303,21 @@ const onResponseFinish = Effect.fn("OpenResponses.onResponseFinish")(function* (
for (const item of (event.response?.output ?? []).map((item, index) => resolveItem(state, item, index))) {
// Terminal recovery cannot insert a checkpoint before already-emitted content.
if (item.type === "compaction" && state.lifecycle.stepStarted && !state.completedCompactions.has(item.id))
return yield* ProviderShared.eventError(
return ProviderShared.eventError(
state.id,
"Cannot recover a compaction checkpoint after output has been emitted",
)
const recoverable =
item.type === "compaction" || (item.type === "function_call" && current.tools[item.id] !== undefined)
if (!recoverable) continue
const [next, emitted] = yield* onOutputItemDone(current, item)
current = next
events.push(...emitted)
const done = onOutputItemDone(current, item)
if (done instanceof AIError) return done
current = done[0]
events.push(...done[1])
}
// Some compatible providers omit output_item.done even after completing the response.
const pending = yield* ToolStream.finishAll(current.id, current.tools)
const pending = ToolStream.finishAll(current.id, current.tools)
if (ToolStream.isError(pending)) return pending
current = {
...current,
tools: pending.tools,
@@ -1353,8 +1341,8 @@ const onResponseFinish = Effect.fn("OpenResponses.onResponseFinish")(function* (
})
: undefined,
})
return [{ ...current, lifecycle }, events] satisfies StepResult
})
return [{ ...current, lifecycle }, events]
}
/** Error code and message from wherever the frame put them; top-level fields win over nested ones. */
export const errorDetail = (event: Event) => {
@@ -1389,6 +1377,9 @@ export const providerFailure = (event: Event, fallback: string, body = ProviderS
return new AIError({ reason })
}
const stepOutcome = (result: StepResult | AIError) =>
result instanceof AIError ? Effect.fail(result) : Effect.succeed(result)
// Callers must pass events through `normalize` first. The OpenAPI requires
// string IDs but imposes no minLength; empty is not missing.
export const step = (state: ParserState, event: NormalizedEvent) => {
@@ -1448,10 +1439,11 @@ export const step = (state: ParserState, event: NormalizedEvent) => {
}
if (event.type === "response.function_call_arguments.delta" || event.type === "response.function_call_arguments.done")
return event.item_id !== undefined
? onFunctionCallArgumentsDelta(state, event)
? stepOutcome(onFunctionCallArgumentsDelta(state, event))
: ProviderShared.eventError(state.id, `${event.type} is missing item_id`)
if (event.type === "response.output_item.done") return onOutputItemDone(state, event.item)
if (event.type === "response.completed" || event.type === "response.incomplete") return onResponseFinish(state, event)
if (event.type === "response.output_item.done") return stepOutcome(onOutputItemDone(state, event.item))
if (event.type === "response.completed" || event.type === "response.incomplete")
return stepOutcome(onResponseFinish(state, event))
if (event.type === "response.failed") return providerFailure(event, `${state.name} response failed`)
if (event.type === "error") return providerFailure(event, `${state.name} stream error`)
return Effect.succeed<StepResult>([state, NO_EVENTS])
+15 -12
View File
@@ -360,7 +360,7 @@ const lowerToolCall = (
extra_content: decodeExtraContent(part.providerMetadata?.[options.providerMetadataKey]?.extraContent),
})
const lowerMedia = Effect.fn("OpenAIChat.lowerMedia")(function* (part: MediaPart) {
const lowerMedia = Effect.fnUntraced(function* (part: MediaPart) {
// Chat Completions accepts PDFs, and no other documents, as inline `file` parts; file URLs are not supported.
if (part.media.mediaType.toLowerCase() === "application/pdf")
return {
@@ -406,7 +406,7 @@ const lowerReasoningDetail = (detail: ReasoningDetail) => {
const isKimiDetail = (detail: { readonly type: string }) => detail.type === "summary" || detail.type === "encrypted"
const lowerUserMessage = Effect.fn("OpenAIChat.lowerUserMessage")(function* (
const lowerUserMessage = Effect.fnUntraced(function* (
message: OpenAIChatRequestMessage,
options: LoweringOptions,
) {
@@ -430,7 +430,7 @@ const lowerUserMessage = Effect.fn("OpenAIChat.lowerUserMessage")(function* (
return { role: "user" as const, content }
})
const lowerAssistantMessage = Effect.fn("OpenAIChat.lowerAssistantMessage")(function* (
const lowerAssistantMessage = Effect.fnUntraced(function* (
message: OpenAIChatRequestMessage,
configuredField: string | undefined,
requireReasoning: boolean,
@@ -495,7 +495,7 @@ const lowerAssistantMessage = Effect.fn("OpenAIChat.lowerAssistantMessage")(func
return { ...result, [field]: reasoningText }
})
const lowerToolMessages = Effect.fn("OpenAIChat.lowerToolMessages")(function* (
const lowerToolMessages = Effect.fnUntraced(function* (
message: OpenAIChatRequestMessage,
options: LoweringOptions,
) {
@@ -532,7 +532,7 @@ const toolMessage = (toolCallID: string, text: string, cacheControl: OpenAIChatC
content: cacheControl === undefined ? text : [{ type: "text" as const, text, cache_control: cacheControl }],
})
const lowerMessage = Effect.fn("OpenAIChat.lowerMessage")(function* (
const lowerMessage = Effect.fnUntraced(function* (
message: OpenAIChatRequestMessage,
reasoningField: string | undefined,
requireReasoning: boolean,
@@ -544,7 +544,7 @@ const lowerMessage = Effect.fn("OpenAIChat.lowerMessage")(function* (
return (yield* lowerToolMessages(message, options)).messages
})
const lowerMessages = Effect.fn("OpenAIChat.lowerMessages")(function* (request: LLMRequest, options: LoweringOptions) {
const lowerMessages = Effect.fnUntraced(function* (request: LLMRequest, options: LoweringOptions) {
const system: OpenAIChatMessage[] =
request.system.length === 0
? []
@@ -855,7 +855,7 @@ export const fromRequest = Effect.fn("OpenAIChat.fromRequest")(function* (
// Streaming parsers are small state machines: every event returns a new state
// plus the common `LLMEvent`s produced by that event. Tool calls are accumulated
// because OpenAI streams JSON arguments across multiple deltas.
const mapFinishReason = Effect.fn("OpenAIChat.mapFinishReason")(function* (event: OpenAIChatEvent, reason: string) {
const mapFinishReason = Effect.fnUntraced(function* (event: OpenAIChatEvent, reason: string) {
switch (reason) {
case "error":
return yield* new AIError({
@@ -1189,8 +1189,9 @@ const step = (state: ParserState, event: OpenAIChatEvent) =>
!incompleteTools &&
state.finishReason === undefined &&
Object.keys(tools).length > 0
? yield* ToolStream.finishAll(ADAPTER, tools)
? ToolStream.finishAll(ADAPTER, tools)
: undefined
if (ToolStream.isError(finished)) return yield* finished
return [
{
@@ -1214,7 +1215,7 @@ const step = (state: ParserState, event: OpenAIChatEvent) =>
] as const
})
const finishEvents = Effect.fn("OpenAIChat.finishEvents")(function* (state: ParserState) {
const finishEvents = Effect.fnUntraced(function* (state: ParserState) {
if (state.finishReason === undefined && state.requireFinishReason)
return yield* new AIError({
reason: new InvalidProviderOutputError({
@@ -1224,10 +1225,12 @@ const finishEvents = Effect.fn("OpenAIChat.finishEvents")(function* (state: Pars
}),
})
const events: LLMEvent[] = []
const toolCallEvents =
const finished =
state.finishReason === undefined && Object.keys(state.tools).length > 0
? (yield* ToolStream.finishAll(ADAPTER, state.tools)).events
: state.toolCallEvents
? ToolStream.finishAll(ADAPTER, state.tools)
: undefined
if (ToolStream.isError(finished)) return yield* finished
const toolCallEvents = finished?.events ?? state.toolCallEvents
const hasToolCalls = toolCallEvents.length > 0
const reason = state.finishReason
? {
@@ -157,7 +157,7 @@ const nativeImageTool = (tool: ToolDefinition) => {
return Schema.is(OpenAIResponsesImageGenerationTool)(native) ? native : undefined
}
const lowerTool = Effect.fn("OpenAIResponses.lowerTool")(function* (tool: ToolDefinition) {
const lowerTool = Effect.fnUntraced(function* (tool: ToolDefinition) {
const native = nativeImageToolInput(tool)
if (native !== undefined) {
if (Schema.is(OpenAIResponsesImageGenerationTool)(native)) return native
@@ -168,7 +168,7 @@ const lowerTool = Effect.fn("OpenAIResponses.lowerTool")(function* (tool: ToolDe
// Native namespaces hold only function tools, so deeper levels flatten into
// the leaf names the same way non-native protocols flatten the whole tree.
const lowerToolEntry = Effect.fn("OpenAIResponses.lowerToolEntry")(function* (tool: ToolEntry) {
const lowerToolEntry = Effect.fnUntraced(function* (tool: ToolEntry) {
if (tool.type === "tool") return yield* lowerTool(tool)
// OpenAI requires a namespace description; fall back to a generic one so a
// missing description never blocks the request.
@@ -237,7 +237,7 @@ const checkpointBody = {
}),
}
const hostedToolResult = Effect.fn("OpenAIResponses.hostedToolResult")(function* (item: ResponsesHostedTools.Item) {
const hostedToolResult = Effect.fnUntraced(function* (item: ResponsesHostedTools.Item) {
const isError = item.error !== undefined && item.error !== null
if (item.type === "image_generation_call" && item.result) {
yield* Effect.fromResult(Encoding.decodeBase64(item.result)).pipe(
+2 -2
View File
@@ -154,7 +154,7 @@ export const wrapSystemUpdate = (parts: ReadonlyArray<{ readonly text: string }>
* raw retrieved, tool, or web content into privileged updates: keep untrusted
* data in ordinary user/tool messages instead.
*/
export const systemUpdateText = Effect.fn("ProviderShared.systemUpdateText")(function* (
export const systemUpdateText = Effect.fnUntraced(function* (
route: string,
message: LLMRequest["messages"][number],
) {
@@ -167,7 +167,7 @@ export const systemUpdateText = Effect.fn("ProviderShared.systemUpdateText")(fun
})
/** Lower an unsupported privileged update into visible, in-order user text. */
export const wrappedSystemUpdate = Effect.fn("ProviderShared.wrappedSystemUpdate")(function* (
export const wrappedSystemUpdate = Effect.fnUntraced(function* (
route: string,
message: LLMRequest["messages"][number],
) {
@@ -76,7 +76,7 @@ function documentName(filename: string | undefined, names: Set<string>) {
return name
}
const mediaBase64 = Effect.fn("BedrockMedia.mediaBase64")(function* (part: MediaPart) {
const mediaBase64 = Effect.fnUntraced(function* (part: MediaPart) {
const media = yield* ProviderShared.requireInlineMedia("Bedrock Converse", part.media)
const bytes = yield* Effect.fromResult(Encoding.decodeBase64(media.base64)).pipe(
Effect.mapError((cause) =>
@@ -91,7 +91,7 @@ const mediaBase64 = Effect.fn("BedrockMedia.mediaBase64")(function* (part: Media
// document block. Image MIME types not in `IMAGE_FORMATS` (e.g. `image/svg+xml`)
// get an image-specific error so the caller knows it's a format-support issue,
// not a kind-detection issue.
export const lower = Effect.fn("BedrockMedia.lower")(function* (part: MediaPart, documentNames: Set<string>) {
export const lower = Effect.fnUntraced(function* (part: MediaPart, documentNames: Set<string>) {
const mime = part.media.mediaType.toLowerCase()
const imageFormat = IMAGE_FORMATS[mime as keyof typeof IMAGE_FORMATS]
if (imageFormat) {
@@ -11,7 +11,7 @@ interface State {
readonly responseID?: string
}
const onOutputItem = Effect.fn("ResponsesCheckpoint.onOutputItem")(function* (
const onOutputItem = Effect.fnUntraced(function* (
state: State,
input: OpenResponses.Event,
) {
@@ -63,7 +63,7 @@ export const make = <Body>(body: RouteBody<Body>): TriggerCompactOperation =>
checkpoints: {},
}),
terminal: OpenResponses.terminal,
step: Effect.fn("ResponsesCheckpoint.step")(function* (state: State, event: OpenResponses.Event) {
step: Effect.fnUntraced(function* (state: State, event: OpenResponses.Event) {
if (event.response?.id && state.responseID && event.response.id !== state.responseID)
return yield* ProviderShared.eventError(source.id, "Compaction response ID changed during execution")
if (event.type === "response.created") return [{ ...state, responseID: event.response?.id }, []] as const
@@ -33,7 +33,7 @@ export const onDone: (
state: OpenResponses.ParserState,
item: Item,
tools: Definitions,
) => Effect.Effect<OpenResponses.StepResult, AIError> = Effect.fn("ResponsesHostedTools.onDone")(
) => Effect.Effect<OpenResponses.StepResult, AIError> = Effect.fnUntraced(
function* (state, item, tools) {
const tool = tools[item.type]
if (!tool) return [state, []] satisfies OpenResponses.StepResult
+49 -66
View File
@@ -1,10 +1,11 @@
import { Effect, Option } from "effect"
import { Option, Result, Schema } from "effect"
import { AIError, LLMEvent, type ProviderMetadata, type ToolCall } from "../../schema/index.js"
import { eventError, parseToolInput, type ToolAccumulator } from "../shared.js"
import { Json, eventError, type ToolAccumulator } from "../shared.js"
import { parse } from "./partial-json.js"
type StreamKey = string | number
const parsePartialInput = Option.liftThrowable(parse)
const decodeInput = Schema.decodeUnknownResult(Json)
/**
* One pending streamed tool call. Providers emit the tool identity and JSON
@@ -69,31 +70,26 @@ const inputDelta = (tool: PendingTool, text: string) =>
input: Option.getOrElse(parsePartialInput(tool.input), () => ({})),
})
const toolCall = (route: string, tool: PendingTool, inputOverride?: string) => {
const toolCall = (route: string, tool: PendingTool, inputOverride?: string): ToolCall | AIError => {
const raw = inputOverride ?? tool.input
return parseToolInput(route, tool.name, raw).pipe(
Effect.catch((error) =>
tool.providerExecuted
? Effect.fail(error)
: Effect.succeed(
Option.getOrElse(
Option.map(parsePartialInput(raw), (input) => input ?? {}),
() => ({}),
),
),
),
Effect.map(
(input): ToolCall =>
LLMEvent.toolCall({
id: tool.id,
name: tool.name,
namespace: tool.namespace,
input,
providerExecuted: tool.providerExecuted ? true : undefined,
providerMetadata: tool.providerMetadata,
}),
),
)
const body = raw || "{}"
const parsed = decodeInput(body)
if (Result.isFailure(parsed) && tool.providerExecuted)
return eventError(route, `Invalid JSON input for ${route} tool call ${tool.name}`, body, parsed.failure)
const input = Result.isSuccess(parsed)
? parsed.success
: Option.getOrElse(
Option.map(parsePartialInput(raw), (value) => value ?? {}),
() => ({}),
)
return LLMEvent.toolCall({
id: tool.id,
name: tool.name,
namespace: tool.namespace,
input,
providerExecuted: tool.providerExecuted ? true : undefined,
providerMetadata: tool.providerMetadata,
})
}
const finishEvents = (tool: PendingTool, event: ToolCall): ReadonlyArray<LLMEvent> => [
@@ -123,8 +119,7 @@ const appendTool = <K extends StreamKey>(
}
}
export const isError = <K extends StreamKey>(result: AppendOutcome<K> | AIError): result is AIError =>
result instanceof AIError
export const isError = <T>(result: T | AIError): result is AIError => result instanceof AIError
/**
* Register a tool call whose start event arrived before any argument deltas.
@@ -198,52 +193,40 @@ export const appendExisting = <K extends StreamKey>(
): AppendOutcome<K> | AIError => append(tools, key, text) ?? eventError(route, missingToolMessage)
/**
* Finalize one pending tool call: parse the accumulated raw JSON, remove it
* from state, and recover incomplete local arguments when needed.
* Finalize one pending tool call: parse the accumulated raw JSON (or an
* authoritative final `input` override from `response.output_item.done`),
* remove it from state, and recover incomplete local arguments when needed.
* Missing keys are a no-op because some providers emit stop events for
* non-tool content blocks.
*/
export const finish = <K extends StreamKey>(route: string, tools: State<K>, key: K) =>
Effect.gen(function* () {
const tool = tools[key]
if (!tool) return { tools }
return {
tools: withoutTool(tools, key),
events: finishEvents(tool, yield* toolCall(route, tool)),
}
})
/**
* Finalize one pending tool call with an authoritative final input string.
* OpenAI Responses can send accumulated deltas and then repeat the completed
* arguments on `response.output_item.done`; the final value wins.
*/
export const finishWithInput = <K extends StreamKey>(route: string, tools: State<K>, key: K, input: string) =>
Effect.gen(function* () {
const tool = tools[key]
if (!tool) return { tools }
return {
tools: withoutTool(tools, key),
events: finishEvents(tool, yield* toolCall(route, tool, input)),
}
})
export const finish = <K extends StreamKey>(route: string, tools: State<K>, key: K, input?: string) => {
const tool = tools[key]
if (!tool) return { tools }
const event = toolCall(route, tool, input)
if (isError(event)) return event
return {
tools: withoutTool(tools, key),
events: finishEvents(tool, event),
}
}
/**
* Finalize every pending tool call at once. OpenAI Chat has this shape: it does
* not emit per-tool stop events, so all accumulated calls finish independently
* when the choice receives a terminal `finish_reason`.
*/
export const finishAll = <K extends StreamKey>(route: string, tools: State<K>) =>
Effect.gen(function* () {
const pending = Object.values<PendingTool | undefined>(tools).filter(
(tool): tool is PendingTool => tool !== undefined,
)
return {
tools: empty<K>(),
events: yield* Effect.forEach(pending, (tool) =>
toolCall(route, tool).pipe(Effect.map((event) => finishEvents(tool, event))),
).pipe(Effect.map((events) => events.flat())),
}
})
export const finishAll = <K extends StreamKey>(route: string, tools: State<K>) => {
const events: LLMEvent[] = []
for (const tool of Object.values<PendingTool | undefined>(tools)) {
if (!tool) continue
const event = toolCall(route, tool)
if (isError(event)) return event
events.push(...finishEvents(tool, event))
}
return {
tools: empty<K>(),
events,
}
}
export * as ToolStream from "./tool-stream.js"
+27 -31
View File
@@ -1,5 +1,5 @@
import { Effect, Stream } from "effect"
import { makeParser, type Event } from "effect/unstable/encoding/Sse"
import { makeParser } from "effect/unstable/encoding/Sse"
import { AIError, InvalidProviderOutputError } from "../schema/index.js"
/**
@@ -42,43 +42,39 @@ export const sseFraming = (
Stream.decodeText(),
Stream.mapAccumEffect(
() => {
const output: Event[] = []
const output: string[] = []
return {
output,
parser: makeParser((event) => {
if (event._tag === "Event") output.push(event)
if (
event._tag === "Event" &&
(events === undefined || events.has(event.event)) &&
event.data.length > 0 &&
// Some OpenAI-compatible proxies serialize an empty flush as a bare
// `data: null`, between events or after `[DONE]`. No protocol has a
// null event, so it carries nothing and must not abort the stream.
event.data !== "null" &&
// Vertex AI partner models (e.g. `xai/grok-4.6`) send their SSE
// keepalive comment as `data: : keepalive` while reasoning.
event.data !== ": keepalive" &&
(event.data !== "[DONE]" || includeDone || (events !== undefined && event.event !== "message"))
)
output.push(event.data)
}),
}
},
(state, chunk) =>
Effect.gen(function* () {
const error = state.parser.feed(chunk)
if (error)
return yield* new AIError({
reason: new InvalidProviderOutputError({
route: "sse",
message: error.message,
body: chunk,
cause: error,
}),
})
return [state, state.output.splice(0)] as const
}),
(state, chunk) => {
const error = state.parser.feed(chunk)
if (!error) return Effect.succeed([state, state.output.splice(0)] as const)
const reason = new InvalidProviderOutputError({
route: "sse",
message: error.message,
body: chunk,
cause: error,
})
return Effect.fail(new AIError({ reason }))
},
),
Stream.filter(
(event) =>
(events === undefined || events.has(event.event)) &&
event.data.length > 0 &&
// Some OpenAI-compatible proxies serialize an empty flush as a bare
// `data: null`, between events or after `[DONE]`. No protocol has a
// null event, so it carries nothing and must not abort the stream.
event.data !== "null" &&
// Vertex AI partner models (e.g. `xai/grok-4.6`) send their SSE
// keepalive comment as `data: : keepalive` while reasoning.
event.data !== ": keepalive" &&
(event.data !== "[DONE]" || includeDone || (events !== undefined && event.event !== "message")),
),
Stream.map((event) => event.data),
)
/** Server-Sent Events framing. Used by every JSON-streaming HTTP provider. */
+18 -10
View File
@@ -19,7 +19,8 @@ describe("ToolStream", () => {
if (ToolStream.isError(first)) return yield* first
const second = ToolStream.appendOrStart(ADAPTER, first.tools, 0, { text: ':"weather"}' }, "missing tool")
if (ToolStream.isError(second)) return yield* second
const finished = yield* ToolStream.finish(ADAPTER, second.tools, 0)
const finished = ToolStream.finish(ADAPTER, second.tools, 0)
if (ToolStream.isError(finished)) return yield* finished
expect(first.events).toEqual([
{ type: "tool-input-start", id: "call_1", name: "lookup" },
@@ -91,7 +92,8 @@ describe("ToolStream", () => {
"missing tool",
)
if (ToolStream.isError(second)) return yield* second
const finished = yield* ToolStream.finish(ADAPTER, second.tools, 0)
const finished = ToolStream.finish(ADAPTER, second.tools, 0)
if (ToolStream.isError(finished)) return yield* finished
expect(finished.events).toEqual([
{ type: "tool-input-end", id: "call_1", name: "lookup" },
@@ -114,7 +116,8 @@ describe("ToolStream", () => {
name: "lookup",
input: '{"query":"partial"}',
})
const finished = yield* ToolStream.finishWithInput(ADAPTER, tools, "item_1", '{"query":"final"}')
const finished = ToolStream.finish(ADAPTER, tools, "item_1", '{"query":"final"}')
if (ToolStream.isError(finished)) return yield* finished
expect(finished).toEqual({
tools: {},
@@ -133,7 +136,8 @@ describe("ToolStream", () => {
name: "lookup",
input: '{"query":"partial',
})
const finished = yield* ToolStream.finish(ADAPTER, tools, "item_1")
const finished = ToolStream.finish(ADAPTER, tools, "item_1")
if (ToolStream.isError(finished)) return yield* finished
expect(finished).toEqual({
tools: {},
@@ -152,7 +156,8 @@ describe("ToolStream", () => {
name: "lookup",
input: '{"path":"A\\H","text":"first\tsecond"}',
})
const finished = yield* ToolStream.finish(ADAPTER, tools, "item_1")
const finished = ToolStream.finish(ADAPTER, tools, "item_1")
if (ToolStream.isError(finished)) return yield* finished
expect(finished.events).toEqual([
{ type: "tool-input-end", id: "call_1", name: "lookup" },
@@ -168,7 +173,8 @@ describe("ToolStream", () => {
name: "lookup",
input: "invalid",
})
const finished = yield* ToolStream.finish(ADAPTER, tools, "item_1")
const finished = ToolStream.finish(ADAPTER, tools, "item_1")
if (ToolStream.isError(finished)) return yield* finished
expect(finished.events).toEqual([
{ type: "tool-input-end", id: "call_1", name: "lookup" },
@@ -189,7 +195,8 @@ describe("ToolStream", () => {
name: "lookup",
input: '{"query":"partial',
})
const finished = yield* ToolStream.finishAll(ADAPTER, tools)
const finished = ToolStream.finishAll(ADAPTER, tools)
if (ToolStream.isError(finished)) return yield* finished
expect(finished).toEqual({
tools: {},
@@ -211,9 +218,9 @@ describe("ToolStream", () => {
input: '{"query":"partial',
providerExecuted: true,
})
const result = yield* Effect.exit(ToolStream.finish(ADAPTER, tools, "item_1"))
const result = ToolStream.finish(ADAPTER, tools, "item_1")
expect(result._tag).toBe("Failure")
expect(result).toBeInstanceOf(AIError)
}),
)
@@ -230,7 +237,8 @@ describe("ToolStream", () => {
input: '{"query":"docs"}',
providerExecuted: true,
})
const finished = yield* ToolStream.finishAll(ADAPTER, tools)
const finished = ToolStream.finishAll(ADAPTER, tools)
if (ToolStream.isError(finished)) return yield* finished
expect(finished).toEqual({
tools: {},
@@ -23,17 +23,20 @@ const ModelKeySchema = Schema.Struct({
modelID: Schema.String,
variant: Schema.optional(Schema.String),
})
export type ModelKey = typeof ModelKeySchema.Type
const ChoiceSchema = Schema.Struct({
model: Persistence.optional(ModelKeySchema),
variant: Persistence.optional(Schema.NullOr(Schema.String)),
})
const StateSchema = Schema.Struct({
...ChoiceSchema.fields,
agent: Persistence.optional(Schema.String),
choices: Persistence.optional(Schema.Record(Schema.String, ChoiceSchema)),
})
type State = typeof StateSchema.Type
const SessionsSchema = Schema.Record(
@@ -61,12 +64,14 @@ export const ModelSelectionSchema = Persistence.migrate(
)
const WORKSPACE_KEY = "__workspace__"
const handoff = new Map<string, State>()
const handoffKey = (scope: ServerScope, dir: string, id: string) => ScopedKey.from(scope, dir, id)
const clone = (value: State | undefined) => {
if (!value) return
return {
...value,
model: value.model ? { ...value.model } : undefined,
@@ -86,11 +91,13 @@ export const { use: useLocal, provider: LocalProvider } = createSimpleContext({
const configuredModel = useConfiguredModel()
const id = createMemo(() => params.id || undefined)
const list = createMemo(() =>
normalizeAgentList(data.location.agent.list({ directory: sdk().directory }) ?? []).filter(
(item) => item.mode !== "subagent" && !item.hidden,
),
)
const agentsVisible = createMemo(() => settings.visibility.customAgents() || hasCustomAgent(list()))
const connected = createMemo(() => new Set(providers.connected().map((item) => item.id)))
@@ -111,13 +118,16 @@ export const { use: useLocal, provider: LocalProvider } = createSimpleContext({
const validModel = (model: ModelKey) => {
const provider = providers.all().get(model.providerID)
return !!provider?.models[model.modelID] && connected().has(model.providerID)
}
const firstModel = (...items: Array<() => ModelKey | undefined>) => {
for (const item of items) {
const model = item()
if (!model) continue
if (validModel(model)) return model
}
}
@@ -128,30 +138,39 @@ export const { use: useLocal, provider: LocalProvider } = createSimpleContext({
createEffect(() => {
const items = list()
if (items.length === 0) {
if (store.current !== undefined) setStore("current", undefined)
return
}
if (items.some((item) => item.name === store.current)) return
setStore("current", items[0]?.name)
})
const scope = createMemo<State | undefined>(() => {
const session = id()
if (!session) return store.draft ?? store.promoting
return saved.session[session] ?? handoff.get(handoffKey(serverSDK.scope, sdk().directory, session))
})
createEffect(() => {
const session = id()
if (!session) return
const key = handoffKey(serverSDK.scope, sdk().directory, session)
const next = handoff.get(key)
if (!next) return
if (saved.session[session] !== undefined) {
handoff.delete(key)
setStore("promoting", undefined)
return
}
@@ -169,15 +188,19 @@ export const { use: useLocal, provider: LocalProvider } = createSimpleContext({
const defaultModel = () => {
for (const provider of providers.connected()) {
const first = Object.values(provider.models)[0]
if (!first) continue
const model = { providerID: provider.id, modelID: first.id }
if (validModel(model)) return model
}
}
const fallback = createMemo(() => firstModel(configuredModel, recentModel, defaultModel))
const durable = () => {
const session = id()
return session ? data.session.get(session) : undefined
}
@@ -189,43 +212,55 @@ export const { use: useLocal, provider: LocalProvider } = createSimpleContext({
},
set(name: string | undefined) {
const item = pickAgent(name)
if (!item) {
setStore("current", undefined)
return
}
batch(() => {
const previous = snapshot()
if (previous.agent === item.name) return
const prev = scope()
const choices = {
...prev?.choices,
...(previous.agent ? { [previous.agent]: { model: previous.model, variant: previous.variant } } : {}),
}
const choices = { ...prev?.choices }
if (previous.agent) choices[previous.agent] = { model: previous.model, variant: previous.variant }
setStore("current", item.name)
const next = {
agent: item.name,
model: choices[item.name]?.model,
variant: choices[item.name]?.variant,
choices,
} satisfies State
write(next)
// Pin both choices while the agent and model acknowledgments arrive separately.
const selected = current()
if (selected) model.set({ providerID: selected.provider.id, modelID: selected.id })
})
},
move(direction: 1 | -1) {
const items = list()
if (items.length === 0) {
setStore("current", undefined)
return
}
let next = items.findIndex((item) => item.name === agent.current()?.name) + direction
if (next < 0) next = items.length - 1
if (next >= items.length) next = 0
const item = items[next]
if (!item) return
agent.set(item.name)
},
@@ -233,26 +268,33 @@ export const { use: useLocal, provider: LocalProvider } = createSimpleContext({
const current = createMemo(() => {
if (!configuredModel.ready()) return
const item = firstModel(
() => scope()?.model,
() => {
const session = durable()
if (session?.agent && session.agent !== agent.current()?.name) return
const model = session?.model
return model && { providerID: model.providerID, modelID: model.id }
},
() => agent.current()?.model,
fallback,
)
if (!item) return
return models.find(item)
})
const configured = () => {
const item = agent.current()
const model = current()
if (!item || !model) return
const global = configuredModel()
return (
getConfiguredAgentVariant({
agent: { model: item.model, variant: item.variant },
@@ -267,15 +309,19 @@ export const { use: useLocal, provider: LocalProvider } = createSimpleContext({
const selected = () => {
const draft = scope()
if (draft?.model && validModel(draft.model)) return draft.variant
const session = durable()
if (session?.agent && session.agent !== agent.current()?.name) return
const value = session?.model
if (value && validModel({ providerID: value.providerID, modelID: value.id })) return value.variant ?? null
}
const snapshot = () => {
const selected = current()
return {
agent: agent.current()?.name,
model: selected ? { providerID: selected.provider.id, modelID: selected.id } : undefined,
@@ -291,15 +337,19 @@ export const { use: useLocal, provider: LocalProvider } = createSimpleContext({
} satisfies State
const session = id()
if (session) {
setSaved("session", session, state)
return
}
setStore("draft", state)
}
const recent = createMemo(() => models.recent.list().filter(validModel).map(models.find).filter(Boolean))
const pending = new Map<string, State>()
const sameSelection = (a: State, b: State) =>
a.agent === b.agent &&
a.model?.providerID === b.model?.providerID &&
@@ -309,7 +359,9 @@ export const { use: useLocal, provider: LocalProvider } = createSimpleContext({
const reconcile = (sessionID: string) => {
const expected = pending.get(sessionID)
const session = data.session.get(sessionID)
if (!expected || !session?.model) return
if (
!sameSelection(expected, {
agent: session.agent,
@@ -320,9 +372,11 @@ export const { use: useLocal, provider: LocalProvider } = createSimpleContext({
return
pending.delete(sessionID)
const draft = saved.session[sessionID]
if (id() !== sessionID || !draft || !sameSelection(draft, expected)) return
setSaved("session", sessionID, { agent: undefined, model: undefined, variant: undefined })
}
onCleanup(serverSDK.event.on("session.model.selected", (event) => reconcile(event.data.sessionID)))
onCleanup(serverSDK.event.on("session.agent.selected", (event) => reconcile(event.data.sessionID)))
onCleanup(
@@ -340,6 +394,7 @@ export const { use: useLocal, provider: LocalProvider } = createSimpleContext({
trackSessionCommit(sessionID: string, selection: { agent: string; model: ModelKey; variant?: string }) {
pending.set(sessionID, selection)
reconcile(sessionID)
return () => {
if (pending.get(sessionID) === selection) pending.delete(sessionID)
}
@@ -347,14 +402,18 @@ export const { use: useLocal, provider: LocalProvider } = createSimpleContext({
cycle(direction: 1 | -1) {
const items = recent()
const item = current()
if (!item) return
const index = items.findIndex((entry) => entry?.provider.id === item.provider.id && entry?.id === item.id)
let next = index === -1 ? (direction === 1 ? 0 : items.length - 1) : index + direction
if (next < 0) next = items.length - 1
if (next >= items.length) next = 0
const entry = items[next]
if (!entry) return
model.set({ providerID: entry.provider.id, modelID: entry.id })
},
@@ -364,10 +423,13 @@ export const { use: useLocal, provider: LocalProvider } = createSimpleContext({
const previous = current()
const same = item && previous?.provider.id === item.providerID && previous.id === item.modelID
write({ model: item, variant: same ? (model.variant.current() ?? null) : undefined })
if (!item) return
// A session draft owns its variant even when preferences change in another session.
if (id() && !same) write({ variant: model.variant.current() ?? null })
models.setVisibility(item, true)
if (!options?.recent) return
models.recent.push(item)
})
@@ -383,6 +445,7 @@ export const { use: useLocal, provider: LocalProvider } = createSimpleContext({
selected,
current() {
const model = current()
return resolveModelVariant({
variants: this.list(),
selected: this.selected(),
@@ -392,12 +455,15 @@ export const { use: useLocal, provider: LocalProvider } = createSimpleContext({
},
list() {
const item = current()
if (!item?.variants) return []
return Object.keys(item.variants)
},
set(value: string | undefined) {
batch(() => {
const model = current()
if (!model) return
write({ model: { providerID: model.provider.id, modelID: model.id }, variant: value ?? null })
models.variant.set({ providerID: model.provider.id, modelID: model.id }, value)
@@ -405,6 +471,7 @@ export const { use: useLocal, provider: LocalProvider } = createSimpleContext({
},
cycle() {
const items = this.list()
if (items.length === 0) return
this.set(
cycleModelVariant({
@@ -428,11 +495,13 @@ export const { use: useLocal, provider: LocalProvider } = createSimpleContext({
},
promote(dir: string, session: string, state?: State) {
const next = clone(state ?? snapshot())
if (!next) return
// Creation already owns the active selection; keep only agent memory once it is in the read model.
// Otherwise a first-message command's configured overrides would stay hidden behind this handoff.
const created = data.session.get(session)
const selection = created?.model
const committed =
selection &&
sameSelection(next, {
@@ -442,6 +511,7 @@ export const { use: useLocal, provider: LocalProvider } = createSimpleContext({
})
? { choices: next.choices }
: next
const key = handoffKey(serverSDK.scope, dir, session)
handoff.set(key, committed)
@@ -454,10 +524,15 @@ export const { use: useLocal, provider: LocalProvider } = createSimpleContext({
},
restore(msg: { sessionID: string; agent: string; model: ModelKey }) {
const session = id()
if (!session) return
if (msg.sessionID !== session) return
if (durable()?.model) return
if (saved.session[session] !== undefined) return
if (handoff.has(handoffKey(serverSDK.scope, sdk().directory, session))) return
setSaved("session", session, {
@@ -468,6 +543,7 @@ export const { use: useLocal, provider: LocalProvider } = createSimpleContext({
},
},
}
return result
},
})
@@ -1,5 +1,5 @@
import { describe, expect, test } from "bun:test"
import { Effect, Schema, SchemaGetter } from "effect"
import { Effect, Predicate, Schema, SchemaGetter } from "effect"
import { Persistence } from "./schema"
describe("persistence schemas", () => {
@@ -12,6 +12,7 @@ describe("persistence schemas", () => {
}),
{ enabled: true, appearance: { width: 240, font: "default" }, variant: "high" },
)
const decode = Schema.decodeUnknownSync(schema)
expect(decode({ appearance: { width: 300 } })).toEqual({
enabled: true,
@@ -32,6 +33,7 @@ describe("persistence schemas", () => {
test("legacy migration observes missing fields before initial defaults are applied", () => {
const current = Persistence.struct({ mode: Schema.Literals(["compact", "full"]), enabled: Schema.Boolean })
const stored = Schema.Struct({
mode: Schema.optional(Schema.Unknown),
expanded: Schema.optional(Schema.Boolean),
@@ -45,6 +47,7 @@ describe("persistence schemas", () => {
encode: SchemaGetter.passthrough(),
}),
)
const schema = Persistence.withInitial(Persistence.migrate(current, stored), { mode: "compact", enabled: true })
const decode = Schema.decodeUnknownSync(schema)
expect(decode({ expanded: true, enabled: false })).toEqual({ mode: "full", enabled: false })
@@ -58,6 +61,7 @@ describe("persistence schemas", () => {
amount: Schema.NumberFromString.check(Schema.isFinite()),
items: Schema.mutable(Schema.Array(Schema.String)),
})
const schema = Persistence.withInitial(current, { amount: 7, items: ["initial"] })
const decode = Schema.decodeUnknownSync(schema)
expect(decode({ amount: "12", items: [] })).toEqual({ amount: 12, items: [] })
@@ -69,6 +73,7 @@ describe("persistence schemas", () => {
const schema = Persistence.struct({
enabled: Schema.Boolean.pipe(Schema.withDecodingDefault(Effect.succeed(true))),
})
const decode = Schema.decodeUnknownSync(schema)
expect(decode({})).toEqual({ enabled: true })
expect(decode({ enabled: undefined })).toEqual({ enabled: true })
@@ -84,6 +89,7 @@ describe("persistence schemas", () => {
label: Persistence.fallback(Schema.String, () => "default"),
amount: Persistence.fallback(Schema.NumberFromString.check(Schema.isFinite()), () => 7),
})
const decode = Schema.decodeUnknownSync(schema)
expect(decode({})).toEqual({ enabled: true, label: "default", amount: 7 })
expect(decode({ enabled: "false", label: "saved", amount: "invalid" })).toEqual({
@@ -142,14 +148,16 @@ describe("persistence schemas", () => {
test("recovers and migrates individual array entries", () => {
const current = Schema.Struct({ name: Schema.String })
const schema = Persistence.array(
Schema.Union([current, Schema.String]).pipe(
Schema.decodeTo(current, {
decode: SchemaGetter.transform((value) => (typeof value === "string" ? { name: value } : value)),
decode: SchemaGetter.transform((value) => (Predicate.isString(value) ? { name: value } : value)),
encode: SchemaGetter.passthrough(),
}),
),
)
const decode = Schema.decodeUnknownSync(schema)
const value = decode(["old", { name: "new" }, null, { name: false }])
expect(value).toEqual([{ name: "old" }, { name: "new" }])
+20 -1
View File
@@ -23,10 +23,13 @@ export function withInitial<S extends Schema.ConstraintCodec<object, unknown>>(
initial: NoInfer<S["Type"]>,
) {
const schema = isMigrated(definition) ? definition.current : definition
const read = isMigrated(definition)
? SchemaParser.decodeUnknownResult(definition.read, { onExcessProperty: "preserve" })
: Result.succeed<unknown>
const encode = Schema.encodeUnknownSync(schema)
return Schema.Unknown.pipe(
Schema.decode<Schema.Unknown>({
decode: SchemaGetter.transformOrFail((value) =>
@@ -39,25 +42,37 @@ export function withInitial<S extends Schema.ConstraintCodec<object, unknown>>(
}
// Object-level codecs own their recovery. Plain structs can recover fields independently.
// SAFETY: stored values and their defaults are untyped trees walked against `ast`; `withInitial` decodes the
// recovered and merged result with the current schema before anyone reads it.
// oxlint-disable-next-line anti-slop/no-unknown-parameters, anti-slop/no-unknown-returns -- see SAFETY above
function recover(ast: SchemaAST.AST, value: unknown, initial: unknown): unknown {
if (value === undefined) return initial
if (ast._tag === "Objects" && !ast.encoding && ast.indexSignatures.length === 0 && Predicate.isObject(value)) {
if (SchemaAST.isObjects(ast) && !ast.encoding && ast.indexSignatures.length === 0 && Predicate.isObject(value)) {
return Object.fromEntries(
ast.propertySignatures.flatMap((field) => {
const defaults = Predicate.isObject(initial) ? initial[field.name] : undefined
const next = recover(field.type, value[field.name], defaults)
if (next === undefined && !Object.hasOwn(value, field.name) && defaults === undefined) return []
return [[field.name, next]]
}),
)
}
const decoded = Schema.decodeUnknownOption(Schema.make<Schema.Codec<unknown, unknown>>(ast))(value)
return Option.isSome(decoded) ? decoded.value : initial
}
// SAFETY: like `recover`, merges untyped trees that `withInitial` decodes with the current schema afterwards.
// oxlint-disable-next-line anti-slop/no-unknown-parameters, anti-slop/no-unknown-returns -- see SAFETY above
function merge(initial: unknown, value: unknown): unknown {
if (value === undefined) return initial
if (!Predicate.isObject(initial) || !Predicate.isObject(value)) return value
return Object.fromEntries(
[...new Set([...Object.keys(initial), ...Object.keys(value)])].map((key) => [key, merge(initial[key], value[key])]),
)
@@ -66,11 +81,13 @@ function merge(initial: unknown, value: unknown): unknown {
// Unlike a decoding default, a fallback also replaces invalid persisted values.
export function fallback<S extends Schema.ConstraintCodec<unknown, unknown>>(schema: S, value: () => S["Type"]) {
const defaulted = Schema.withDecodingDefaultType<S>(Effect.sync(value))(schema)
return Schema.catchDecoding<typeof defaulted>(() => Effect.sync(() => Option.some(value())))(defaulted)
}
export function optional<S extends Schema.ConstraintCodec<unknown, unknown>>(schema: S) {
const field = Schema.optional(schema)
return Schema.catchDecoding<typeof field>(() => Effect.succeed(Option.none()))(field)
}
@@ -80,6 +97,7 @@ export function struct<const Fields extends Schema.Struct.Fields>(fields: Fields
export function record<S extends Schema.ConstraintCodec<unknown, unknown>>(schema: S) {
const entries = Schema.Record(Schema.String, Schema.mutableKey(schema))
return fallback(entries, () => Schema.decodeUnknownSync(entries)({}))
}
@@ -87,6 +105,7 @@ export function record<S extends Schema.ConstraintCodec<unknown, unknown>>(schem
export function array<S extends Schema.ConstraintCodec<unknown, unknown>>(schema: S) {
const decode = Schema.decodeUnknownOption(schema)
const encode = Schema.encodeSync(schema)
return fallback(
Schema.Array(Schema.Unknown).pipe(
Schema.decodeTo(Schema.mutable(Schema.Array(Schema.toType(schema))), {
+13 -2
View File
@@ -1,4 +1,4 @@
import { Effect, Option, Schema, SchemaGetter } from "effect"
import { Effect, Option, Predicate, Schema, SchemaGetter } from "effect"
import { Persistence } from "@/runtime/persistence/schema"
export const ServerKey = Schema.String.pipe(Schema.brand("ServerConnection.Key"))
@@ -19,8 +19,10 @@ export const ServerHttp = Persistence.struct({
const StoredServer = Schema.Union([ServerHttp, ServerHttpBase, Schema.String]).pipe(
Schema.decodeTo(ServerHttp, {
decode: SchemaGetter.transform((value) => {
if (typeof value === "string") return { type: "http", http: { url: value } }
if (Predicate.isString(value)) return { type: "http", http: { url: value } }
if ("http" in value) return value
return { type: "http", http: value }
}),
encode: SchemaGetter.transform((value) => value),
@@ -33,7 +35,9 @@ const ProjectList = Persistence.array(
expanded: Persistence.fallback(Schema.Boolean, () => true),
}),
)
const Projects = Persistence.record(ProjectList)
const LastProject = Persistence.record(Schema.String.pipe(Schema.catchDecoding(() => Effect.succeed(Option.none()))))
const State = Persistence.struct({
@@ -57,12 +61,15 @@ export function serverState(canonicalLocalServer: () => string | undefined = ()
Schema.decode({
decode: SchemaGetter.transform((value) => {
const canonical = canonicalLocalServer()
if (!canonical || canonical === "local") return value
const previous = value.projects[canonical]
const last = value.lastProject[canonical]
if (!previous && last === undefined) return value
const projects = { ...value.projects }
if (previous) {
const local = projects.local ?? []
const worktrees = new Set(local.map((project) => project.worktree))
@@ -71,16 +78,20 @@ export function serverState(canonicalLocalServer: () => string | undefined = ()
...previous.filter((project) => {
if (worktrees.has(project.worktree)) return false
worktrees.add(project.worktree)
return true
}),
]
delete projects[canonical]
}
const lastProject = { ...value.lastProject }
if (last !== undefined) {
lastProject.local ??= last
delete lastProject[canonical]
}
return { ...value, projects, lastProject }
}),
encode: SchemaGetter.transform((value) => value),
+11 -7
View File
@@ -95,7 +95,14 @@ export const ensure = Effect.fn("service.ensure")(function* (options: EnsureOpti
} else timeouts = undefined
if (service !== undefined) {
spawnDelay = timing.spawnDelay
const compatible = service.compatible && matchesVersion(service.version, options)
const versionMatches = matchesVersion(service.version, options)
const compatible = service.compatible && versionMatches
if (!service.compatible && versionMatches)
return yield* Effect.fail(
new Error(
"Background service uses an incompatible health protocol. Update this client or explicitly restart the service.",
),
)
if (compatible && service.state === "ready") {
yield* Effect.tryPromise(() => PtyHandoff.complete(options.file ?? fallback(), service.info))
return Option.some(service)
@@ -205,10 +212,7 @@ const probe = Effect.fnUntraced(function* (info: Info) {
return (yield* probeResult(info)).service
})
const probeResult = Effect.fnUntraced(function* (
info: Info,
timeout = defaultEnsureTiming.requestTimeout,
) {
const probeResult = Effect.fnUntraced(function* (info: Info, timeout = defaultEnsureTiming.requestTimeout) {
const endpoint = {
url: info.url,
auth:
@@ -230,8 +234,8 @@ const probeResult = Effect.fnUntraced(function* (
)
if ("cause" in result) return { service: undefined, timedOut: signal.aborted }
const response = result.value.response
// The previous V2 service exposes /api/status instead. Its authenticated 404 is enough
// to recognize the registered daemon as incompatible and route it through replacement.
// A missing health endpoint identifies protocol incompatibility, not an older
// version. Only an unmet version requirement lets ensure replace this owner.
if (response.status === 404)
return {
service: {
+8 -3
View File
@@ -76,7 +76,12 @@ export async function ensure(options: EnsureOptions = {}): Promise<Endpoint> {
if (registration.service !== undefined) {
spawnDelay = timing.spawnDelay
const service = registration.service
const compatible = service.compatible && matchesVersion(service.version, options)
const versionMatches = matchesVersion(service.version, options)
const compatible = service.compatible && versionMatches
if (!service.compatible && versionMatches)
throw new Error(
"Background service uses an incompatible health protocol. Update this client or explicitly restart the service.",
)
if (compatible && service.state === "ready") {
await PtyHandoff.complete(options.file ?? fallback(), service.info)
return service.endpoint
@@ -177,8 +182,8 @@ async function probeResult(info: Info, timeout = defaultEnsureTiming.requestTime
)
if ("cause" in result) return { service: undefined, timedOut: signal.aborted }
const response = result.value.response
// The previous V2 service exposes /api/status instead. Its authenticated 404 is enough
// to recognize the registered daemon as incompatible and route it through replacement.
// A missing health endpoint identifies protocol incompatibility, not an older
// version. Only an unmet version requirement lets ensure replace this owner.
if (response.status === 404)
return {
service: {
+7 -1
View File
@@ -44,8 +44,12 @@ const handoff = {
}
const server = Bun.serve({
port: 0,
async fetch(request) {
async fetch(request): Promise<Response> {
const pathname = new URL(request.url).pathname
if (mode === "protocol" && pathname === "/api/experimental/persistent-pty/handoff") {
await writeFile(registration + ".handoff-request", "")
return Response.json({ handoff: null })
}
if (pathname === "/api/experimental/persistent-pty/handoff" && mode === "handoff") {
if (request.method !== "POST" || request.headers.get("authorization") !== "Basic " + btoa("opencode:private"))
return new Response(null, { status: 401 })
@@ -55,6 +59,8 @@ const server = Bun.serve({
if (pathname === "/api/experimental/persistent-pty/handoff" && mode === "handoff-broken")
return new Response(null, { status: 500 })
if (pathname !== "/api/info") return new Response(null, { status: 404 })
if (mode === "protocol" && (await Bun.file(registration + ".missing-health").exists()))
return new Response(null, { status: 404 })
requests += 1
if (mode === "starting") await writeFile(registration + ".status-request", "")
if (mode === "hanging") {
@@ -0,0 +1,94 @@
import { NodeFileSystem } from "@effect/platform-node"
import { expect, test } from "bun:test"
import { Effect } from "effect"
import { Service } from "../src/effect/service"
import { ensure, stop } from "../src/promise/service"
import type { EnsureOptions, EnsureReason, StopOptions } from "../src/service"
import { serviceFixture } from "./fixture/service-fixture"
import { accelerate } from "./fixture/service-timing"
const effect = accelerate(Service.ensure)
const clients = [
{ name: "Promise", ensure: accelerate(ensure), stop },
{
name: "Effect",
ensure: (options: EnsureOptions) => Effect.runPromise(effect(options).pipe(Effect.provide(NodeFileSystem.layer))),
stop: (options: StopOptions) => Effect.runPromise(Service.stop(options).pipe(Effect.provide(NodeFileSystem.layer))),
},
]
for (const client of clients) {
for (const policy of [
{ name: "no version requirement", version: undefined },
{ name: "matching exact version", version: "test" },
{ name: "accepting version predicate", version: (version: string) => version === "test" },
]) {
test(`${client.name} reconnect preserves a protocol-incompatible owner with ${policy.name}`, async () => {
await using fixture = await serviceFixture()
const owner = fixture.spawn("protocol")
await fixture.waitForFile()
const original = await Bun.file(fixture.registration).json()
const starts: EnsureReason[] = []
const options = {
file: fixture.registration,
version: policy.version,
command: fixture.command("record-start"),
onStart: (reason: EnsureReason) => starts.push(reason),
}
expect((await client.ensure(options)).url).toBe(original.url)
// The same process remains alive. Only its health endpoint becomes unavailable.
await Bun.write(fixture.registration + ".missing-health", "")
const error = await client.ensure(options).catch((error: unknown) => error)
expect(await Bun.file(fixture.registration + ".signal").exists()).toBe(false)
expect(await Bun.file(fixture.registration + ".handoff-request").exists()).toBe(false)
expect(await Bun.file(fixture.registration + ".started").exists()).toBe(false)
expect(starts).toEqual([])
expect(owner.exitCode).toBe(null)
expect(await Bun.file(fixture.registration).json()).toEqual(original)
expect(error).toBeInstanceOf(Error)
if (!(error instanceof Error)) throw error
expect(error.message).toContain("incompatible health protocol")
await Bun.file(fixture.registration + ".missing-health").delete()
expect((await client.ensure(options)).url).toBe(original.url)
expect(owner.exitCode).toBe(null)
})
}
test(`${client.name} can explicitly replace a protocol-incompatible owner with a nonmatching version`, async () => {
await using fixture = await serviceFixture()
const owner = fixture.spawn("protocol")
await fixture.waitForFile()
await Bun.write(fixture.registration + ".missing-health", "")
const starts: EnsureReason[] = []
const endpoint = await client.ensure({
file: fixture.registration,
version: "2.1.0-next.1",
command: fixture.command("compatible"),
onStart: (reason) => starts.push(reason),
})
const replacement = await Bun.file(fixture.registration).json()
fixture.track(replacement.pid)
expect(await owner.exited).toBe(0)
expect(await Bun.file(fixture.registration + ".signal").text()).toBe("SIGTERM")
expect(await Bun.file(fixture.registration + ".handoff-request").exists()).toBe(true)
expect(starts).toEqual(["version-mismatch"])
expect(replacement.pid).not.toBe(owner.pid)
expect(replacement.version).toBe("2.1.0-next.1")
expect(endpoint.url).toBe(replacement.url)
})
test(`${client.name} can explicitly stop a protocol-incompatible owner`, async () => {
await using fixture = await serviceFixture()
const owner = fixture.spawn("protocol")
await fixture.waitForFile()
await Bun.write(fixture.registration + ".missing-health", "")
await client.stop({ file: fixture.registration })
expect(await owner.exited).toBe(0)
expect(await Bun.file(fixture.registration + ".signal").text()).toBe("SIGTERM")
expect(await Bun.file(fixture.registration).exists()).toBe(false)
})
}
+1 -1
View File
@@ -1 +1 @@
{"deepinfra":{"id":"deepinfra","env":["DEEPINFRA_API_KEY"],"npm":"@ai-sdk/deepinfra","name":"Deep Infra","doc":"https://deepinfra.com/models","models":{"tencent/Hy3":{"id":"tencent/Hy3","name":"Hy3","description":"Tencent Hy reasoning model for coding, instruction following, and agent tasks","family":"Hy","attachment":false,"reasoning":true,"reasoning_options":[],"tool_call":true,"structured_output":true,"temperature":true,"release_date":"2026-07-06","last_updated":"2026-07-06","modalities":{"input":["text"],"output":["text"]},"open_weights":true,"limit":{"context":262144,"input":192000,"output":128000},"cost":{"input":0.13,"output":0.53,"cache_read":0.033},"canonical_model_id":"tencent/hy3"},"tencent/Hy4-preview":{"id":"tencent/Hy4-preview","name":"Hy4 preview","description":"A next-generation productivity model with significantly enhanced Agent and complex task execution capabilities.","family":"Hy","attachment":false,"reasoning":true,"reasoning_options":[{"type":"effort","values":["none","high"]}],"tool_call":true,"structured_output":true,"temperature":true,"release_date":"2026-08-28","last_updated":"2026-08-28","modalities":{"input":["text"],"output":["text"]},"open_weights":true,"limit":{"context":1048576,"output":64000},"cost":{"input":0.834,"output":2.501,"cache_read":0.042},"canonical_model_id":"tencent/hy4-preview"},"meta-llama/Llama-3.3-70B-Instruct-Turbo":{"id":"meta-llama/Llama-3.3-70B-Instruct-Turbo","name":"Llama 3.3 70B Turbo","description":"Compact Llama instruction model for fast chat and local deployment","family":"llama","attachment":false,"reasoning":false,"tool_call":true,"structured_output":true,"release_date":"2024-12-06","last_updated":"2024-12-06","modalities":{"input":["text"],"output":["text"]},"open_weights":true,"limit":{"context":131072,"output":16384},"cost":{"input":0.1,"output":0.32}},"meta-llama/Llama-4-Scout-17B-16E-Instruct":{"id":"meta-llama/Llama-4-Scout-17B-16E-Instruct","name":"Llama 4 Scout 17B","description":"Open multimodal Llama model for long-context analysis and efficient agents","family":"llama","attachment":true,"reasoning":false,"tool_call":true,"structured_output":true,"release_date":"2025-04-05","last_updated":"2025-04-05","modalities":{"input":["text","image"],"output":["text"]},"open_weights":true,"limit":{"context":327680,"output":16384},"cost":{"input":0.1,"output":0.3}},"meta-llama/Llama-4-Maverick-17B-128E-Instruct-FP8":{"id":"meta-llama/Llama-4-Maverick-17B-128E-Instruct-FP8","name":"Llama 4 Maverick 17B FP8","description":"Open multimodal Llama model for strong reasoning and fast responses","family":"llama","attachment":true,"reasoning":false,"tool_call":false,"structured_output":true,"release_date":"2025-04-05","last_updated":"2025-04-05","modalities":{"input":["text","image"],"output":["text"]},"open_weights":true,"limit":{"context":1048576,"output":16384},"status":"deprecated","cost":{"input":0.2,"output":0.8}},"XiaomiMiMo/MiMo-V2.6-Pro":{"id":"XiaomiMiMo/MiMo-V2.6-Pro","name":"MiMo-V2.6-Pro","description":"Stronger MiMo Pro tier for multimodal reasoning and coding-agent execution","family":"mimo","attachment":true,"reasoning":true,"reasoning_options":[{"type":"toggle"}],"tool_call":true,"structured_output":true,"temperature":true,"release_date":"2026-09-22","last_updated":"2026-09-22","modalities":{"input":["text","image","audio","video"],"output":["text"]},"open_weights":true,"limit":{"context":1048576,"output":131072},"cost":{"input":0.43,"output":0.87,"cache_read":0.0036},"canonical_model_id":"xiaomi/mimo-v2.6-pro"},"XiaomiMiMo/MiMo-V2.5-Pro":{"id":"XiaomiMiMo/MiMo-V2.5-Pro","name":"MiMo-V2.5-Pro","description":"Stronger MiMo Pro tier for multimodal reasoning and coding-agent execution","family":"mimo","attachment":true,"reasoning":true,"reasoning_options":[{"type":"toggle"}],"tool_call":true,"interleaved":{"field":"reasoning_content"},"structured_output":true,"temperature":true,"knowledge":"2024-12","release_date":"2026-04-22","last_updated":"2026-04-22","modalities":{"input":["text","audio"],"output":["text"]},"open_weights":true,"limit":{"context":1048576,"output":16384},"status":"deprecated","cost":{"input":1,"output":3,"cache_read":0.2},"canonical_model_id":"xiaomi/mimo-v2.5-pro"},"XiaomiMiMo/MiMo-V2.6-Flash":{"id":"XiaomiMiMo/MiMo-V2.6-Flash","name":"MiMo-V2.6-Flash","description":"MiMo Flash model for multimodal coding agents and long-context automation","family":"mimo","attachment":true,"reasoning":true,"reasoning_options":[{"type":"toggle"}],"tool_call":true,"structured_output":true,"temperature":true,"release_date":"2026-09-22","last_updated":"2026-09-22","modalities":{"input":["text","image","audio","video"],"output":["text"]},"open_weights":true,"limit":{"context":1048576,"output":131072},"cost":{"input":0.14,"output":0.28,"cache_read":0.0028},"canonical_model_id":"xiaomi/mimo-v2.6-flash"},"XiaomiMiMo/MiMo-V2.5":{"id":"XiaomiMiMo/MiMo-V2.5","name":"MiMo-V2.5","description":"Open MiMo model for multimodal coding agents and long-coLine truncated
{"deepinfra":{"id":"deepinfra","env":["DEEPINFRA_API_KEY"],"npm":"@ai-sdk/deepinfra","name":"Deep Infra","doc":"https://deepinfra.com/models","models":{"tencent/Hy3":{"id":"tencent/Hy3","name":"Hy3","description":"Tencent Hy reasoning model for coding, instruction following, and agent tasks","family":"Hy","attachment":false,"reasoning":true,"reasoning_options":[],"tool_call":true,"structured_output":true,"temperature":true,"release_date":"2026-07-06","last_updated":"2026-07-06","modalities":{"input":["text"],"output":["text"]},"open_weights":true,"limit":{"context":262144,"input":192000,"output":128000},"cost":{"input":0.13,"output":0.53,"cache_read":0.033},"canonical_model_id":"tencent/hy3"},"tencent/Hy4-preview":{"id":"tencent/Hy4-preview","name":"Hy4 preview","description":"A next-generation productivity model with significantly enhanced Agent and complex task execution capabilities.","family":"Hy","attachment":false,"reasoning":true,"reasoning_options":[{"type":"effort","values":["none","high"]}],"tool_call":true,"structured_output":true,"temperature":true,"release_date":"2026-08-28","last_updated":"2026-08-28","modalities":{"input":["text"],"output":["text"]},"open_weights":true,"limit":{"context":1048576,"output":64000},"cost":{"input":0.834,"output":2.501,"cache_read":0.042},"canonical_model_id":"tencent/hy4-preview"},"meta-llama/Llama-3.3-70B-Instruct-Turbo":{"id":"meta-llama/Llama-3.3-70B-Instruct-Turbo","name":"Llama 3.3 70B Turbo","description":"Compact Llama instruction model for fast chat and local deployment","family":"llama","attachment":false,"reasoning":false,"tool_call":true,"structured_output":true,"release_date":"2024-12-06","last_updated":"2024-12-06","modalities":{"input":["text"],"output":["text"]},"open_weights":true,"limit":{"context":131072,"output":16384},"cost":{"input":0.1,"output":0.32}},"meta-llama/Llama-4-Scout-17B-16E-Instruct":{"id":"meta-llama/Llama-4-Scout-17B-16E-Instruct","name":"Llama 4 Scout 17B","description":"Open multimodal Llama model for long-context analysis and efficient agents","family":"llama","attachment":true,"reasoning":false,"tool_call":true,"structured_output":true,"release_date":"2025-04-05","last_updated":"2025-04-05","modalities":{"input":["text","image"],"output":["text"]},"open_weights":true,"limit":{"context":327680,"output":16384},"cost":{"input":0.1,"output":0.3}},"meta-llama/Llama-4-Maverick-17B-128E-Instruct-FP8":{"id":"meta-llama/Llama-4-Maverick-17B-128E-Instruct-FP8","name":"Llama 4 Maverick 17B FP8","description":"Open multimodal Llama model for strong reasoning and fast responses","family":"llama","attachment":true,"reasoning":false,"tool_call":false,"structured_output":true,"release_date":"2025-04-05","last_updated":"2025-04-05","modalities":{"input":["text","image"],"output":["text"]},"open_weights":true,"limit":{"context":1048576,"output":16384},"status":"deprecated","cost":{"input":0.2,"output":0.8}},"XiaomiMiMo/MiMo-V2.6-Pro":{"id":"XiaomiMiMo/MiMo-V2.6-Pro","name":"MiMo-V2.6-Pro","description":"Stronger MiMo Pro tier for multimodal reasoning and coding-agent execution","family":"mimo","attachment":true,"reasoning":true,"reasoning_options":[{"type":"toggle"}],"tool_call":true,"structured_output":true,"temperature":true,"release_date":"2026-09-22","last_updated":"2026-09-22","modalities":{"input":["text","image","audio","video"],"output":["text"]},"open_weights":true,"limit":{"context":1048576,"output":131072},"cost":{"input":0.43,"output":0.87,"cache_read":0.0036},"canonical_model_id":"xiaomi/mimo-v2.6-pro"},"XiaomiMiMo/MiMo-V2.5-Pro":{"id":"XiaomiMiMo/MiMo-V2.5-Pro","name":"MiMo-V2.5-Pro","description":"Stronger MiMo Pro tier for multimodal reasoning and coding-agent execution","family":"mimo","attachment":true,"reasoning":true,"reasoning_options":[{"type":"toggle"}],"tool_call":true,"interleaved":{"field":"reasoning_content"},"structured_output":true,"temperature":true,"knowledge":"2024-12","release_date":"2026-04-22","last_updated":"2026-04-22","modalities":{"input":["text","audio"],"output":["text"]},"open_weights":true,"limit":{"context":1048576,"output":16384},"status":"deprecated","cost":{"input":1,"output":3,"cache_read":0.2},"canonical_model_id":"xiaomi/mimo-v2.5-pro"},"XiaomiMiMo/MiMo-V2.6-Flash":{"id":"XiaomiMiMo/MiMo-V2.6-Flash","name":"MiMo-V2.6-Flash","description":"MiMo Flash model for multimodal coding agents and long-context automation","family":"mimo","attachment":true,"reasoning":true,"reasoning_options":[{"type":"toggle"}],"tool_call":true,"structured_output":true,"temperature":true,"release_date":"2026-09-22","last_updated":"2026-09-22","modalities":{"input":["text","image","audio","video"],"output":["text"]},"open_weights":true,"limit":{"context":1048576,"output":131072},"cost":{"input":0.14,"output":0.28,"cache_read":0.0028},"canonical_model_id":"xiaomi/mimo-v2.6-flash"},"XiaomiMiMo/MiMo-V2.5":{"id":"XiaomiMiMo/MiMo-V2.5","name":"MiMo-V2.5","description":"Open MiMo model for multimodal coding agents and long-coLine truncated
@@ -13,15 +13,19 @@ export function createPickedFileAuthorizations(
add(sender: number, paths: string[]) {
const token = randomUUID()
selections.set(token, { sender, paths: new Set(paths), remaining: MAX_ATTACHMENT_BYTES })
return token
},
read: Effect.fn("DesktopFiles.readPickedFile")(function* (sender: number, token: string, path: string) {
const selection = selections.get(token)
if (selection?.sender !== sender || !selection.paths.delete(path))
throw new Error(nativeT("desktop.picker.error.notSelected"))
const bytes = yield* read(path, selection.remaining)
selection.remaining -= bytes.byteLength
if (selection.paths.size === 0) selections.delete(token)
return bytes
}),
release(sender: number, token: string) {
@@ -32,6 +36,7 @@ export function createPickedFileAuthorizations(
export function assertAttachmentBudget(files: { size: number }[]) {
const total = files.reduce((sum, file) => sum + file.size, 0)
if (total <= MAX_ATTACHMENT_BYTES) return
throw new Error(nativeT("desktop.picker.error.sizeLimit", { limit: MAX_ATTACHMENT_BYTES / 1024 / 1024 }))
}
@@ -42,16 +47,20 @@ export function readAttachment(filePath: string, maxBytes = MAX_ATTACHMENT_BYTES
const fs = yield* FileSystem.FileSystem
const file = yield* fs.open(filePath, { flag: "r" })
const info = yield* file.stat
if (info.size > FileSystem.Size(maxBytes))
throw new Error(nativeT("desktop.picker.error.sizeLimit", { limit: MAX_ATTACHMENT_BYTES / 1024 / 1024 }))
const bytes = new Uint8Array(Number(info.size))
let offset = 0
while (offset < bytes.byteLength) {
const read = Number(yield* file.read(bytes.subarray(offset)))
if (read === 0) break
offset += read
}
return bytes.buffer.slice(0, offset)
}),
)
+58 -22
View File
@@ -2,7 +2,7 @@ import { describe, expect, test } from "bun:test"
import { EventEmitter } from "node:events"
import { MessageChannel } from "node:worker_threads"
import type { MessagePortMain, WebContents } from "electron"
import { Effect, Layer, ManagedRuntime, Schema, Stream } from "effect"
import { Effect, Layer, ManagedRuntime, Predicate, Schema, Stream } from "effect"
import { Rpc, RpcGroup, RpcMessage, RpcServer } from "effect/unstable/rpc"
import { Transferable } from "effect/unstable/workers"
import { omitUndefined } from "../shared/ipc-transport"
@@ -13,13 +13,16 @@ describe("desktop RPC transport", () => {
test("decodes renderer payloads whose optional fields are undefined", async () => {
let received: unknown
const rpcs = RpcGroup.make(FilesOpenFilePicker)
const handlers = rpcs.toLayer({
FilesOpenFilePicker: ({ options }) =>
Effect.sync(() => {
received = options
return null
}),
})
const live = RpcServer.layer(rpcs).pipe(Layer.provide(handlers), Layer.provideMerge(IpcServerProtocolLive))
const runtime = ManagedRuntime.make(live)
const handoff = await runtime.runPromise(IpcPortHandoff)
@@ -30,13 +33,10 @@ describe("desktop RPC transport", () => {
// Structured clone keeps a present-but-undefined key, so it reaches the JSON codec.
const rejected = await call(channel.port2, 0, "FilesOpenFilePicker", payload)
expect(rejected.exit).toMatchObject({
_tag: "Failure",
cause: [{ _tag: "Die", defect: expect.stringContaining('["options"]["title"]') }],
})
expect(rejected.exit).toMatchObject(died(expect.stringContaining('["options"]["title"]')))
const accepted = await call(channel.port2, 1, "FilesOpenFilePicker", omitUndefined(payload))
expect(accepted.exit).toEqual({ _tag: "Success", value: null })
expect(accepted.exit).toEqual(success(null))
expect(received).toEqual({ multiple: true, defaultPath: "C:\\project" })
channel.port2.close()
@@ -56,13 +56,16 @@ describe("desktop RPC transport", () => {
test("keeps multiple renderer ports independent", async () => {
let received: unknown
const handlers = TestRpcs.toLayer(
Effect.gen(function* () {
const handoff = yield* IpcPortHandoff
return TestRpcs.of({
"test.focused": (_request, context) => Effect.succeed(handoff.sender(context.client.id)?.id === 1),
"test.blob.put": ({ data }) => {
received = data
return Effect.succeed([...data].join(","))
},
"test.blob.get": () => Effect.succeed(new Uint8Array([3, 1, 4])),
@@ -70,6 +73,7 @@ describe("desktop RPC transport", () => {
})
}),
)
const live = RpcServer.layer(TestRpcs).pipe(Layer.provide(handlers), Layer.provideMerge(IpcServerProtocolLive))
const runtime = ManagedRuntime.make(live)
const handoff = await runtime.runPromise(IpcPortHandoff)
@@ -82,34 +86,36 @@ describe("desktop RPC transport", () => {
call(first.port2, 0, "test.focused", null),
call(second.port2, 0, "test.focused", null),
])
expect(focused.exit).toEqual({ _tag: "Success", value: true })
expect(unfocused.exit).toEqual({ _tag: "Success", value: false })
expect(focused.exit).toEqual(success(true))
expect(unfocused.exit).toEqual(success(false))
const put = await call(first.port2, 1, "test.blob.put", omitUndefined({ data: new Uint8Array([2, 7, 1]) }))
expect(put.exit).toEqual({ _tag: "Success", value: "2,7,1" })
expect(put.exit).toEqual(success("2,7,1"))
// Binary payloads arrive as bytes, not as base64 text or a plain object.
expect(received).toBeInstanceOf(Uint8Array)
expect((await call(first.port2, 2, "test.blob.get", null)).exit).toEqual({
_tag: "Success",
value: new Uint8Array([3, 1, 4]),
})
expect((await call(first.port2, 2, "test.blob.get", null)).exit).toEqual(success(new Uint8Array([3, 1, 4])))
expect((await call(first.port2, 3, "test.events", null)).chunks).toEqual([
{ _tag: "TestEvent", value: "session.new" },
Schema.encodeSync(TestEvent)(new TestEvent({ value: "session.new" })),
])
const reloaded = new MessageChannel()
handoff.bind(sender(1), serverPort(reloaded.port1))
const [reloadedFocused, stillUnfocused] = await Promise.all([
call(reloaded.port2, 0, "test.focused", null),
call(second.port2, 1, "test.focused", null),
])
expect(reloadedFocused.exit).toEqual({ _tag: "Success", value: true })
expect(stillUnfocused.exit).toEqual({ _tag: "Success", value: false })
expect(reloadedFocused.exit).toEqual(success(true))
expect(stillUnfocused.exit).toEqual(success(false))
for (const port of [first.port2, second.port2, reloaded.port2]) port.close()
await runtime.dispose()
})
})
class TestEvent extends Schema.TaggedClass<TestEvent>()("TestEvent", { value: Schema.String }) {}
const TestRpcs = RpcGroup.make(
Rpc.make("test.focused", { success: Schema.Boolean }),
Rpc.make("test.blob.put", { payload: { data: Transferable.Uint8Array }, success: Schema.String }),
@@ -117,28 +123,47 @@ const TestRpcs = RpcGroup.make(
Rpc.make("test.events", { success: TestEvent, stream: true }),
)
// SAFETY: these are the RPC wire format's plain encoded messages and exits, which the tests post and expect as is.
/* oxlint-disable anti-slop-effect/no-manual-tagged-construction -- see SAFETY above */
function success<A>(value: A) {
return { _tag: "Success", value } as const
}
function died<A>(defect: A) {
return { _tag: "Failure", cause: [{ _tag: "Die", defect }] } as const
}
// Speaks the wire format the way src/renderer/ipc-client.ts does: post a request, ack each chunk,
// and settle on the exit. The payload is posted as given so a test can send what omitUndefined drops.
function call(port: MessageChannel["port2"], id: number, tag: string, payload: unknown) {
function call(port: MessageChannel["port2"], id: number, tag: string, payload: RpcMessage.RequestEncoded["payload"]) {
const chunks: unknown[] = []
return new Promise<{ chunks: unknown[]; exit: RpcMessage.ResponseExitEncoded["exit"] }>((resolve) => {
const onMessage = (message: RpcMessage.FromServerEncoded) => {
if (!("requestId" in message) || Number(message.requestId) !== id) return
if (message._tag === "Chunk") {
if (Predicate.isTagged(message, "Chunk")) {
chunks.push(...message.values)
port.postMessage({ _tag: "Ack", requestId: message.requestId } satisfies RpcMessage.AckEncoded)
return
}
port.off("message", onMessage)
resolve({ chunks, exit: message.exit })
}
port.on("message", onMessage)
port.postMessage({ _tag: "Request", id, tag, payload, headers: [] })
})
}
/* oxlint-enable anti-slop-effect/no-manual-tagged-construction */
function sender(id: number) {
const events = new EventEmitter()
// SAFETY: the transport reads only a sender's `id` and `isDestroyed`, and its `destroyed` event.
// oxlint-disable-next-line anti-slop/no-chained-type-assertions -- see SAFETY above
return {
id,
isDestroyed: () => false,
@@ -148,27 +173,38 @@ function sender(id: number) {
}
function serverPort(port: MessageChannel["port1"]) {
const listeners = new Map<(event: Electron.MessageEvent) => void, (data: unknown) => void>()
return {
const listeners = new Map<(event: Electron.MessageEvent) => void, (data: Electron.MessageEvent["data"]) => void>()
const fake = {
on(event: string, listener: (event: Electron.MessageEvent) => void) {
if (event !== "message") {
port.on(event, listener)
return
}
const wrapped = (data: unknown) => listener({ data } as Electron.MessageEvent)
// SAFETY: the transport reads only the `data` of a message event.
const wrapped = (data: Electron.MessageEvent["data"]) => listener({ data } as Electron.MessageEvent)
listeners.set(listener, wrapped)
port.on("message", wrapped)
},
off(event: string, listener: (event: Electron.MessageEvent) => void) {
if (event !== "message") {
port.off(event, listener)
return
}
const wrapped = listeners.get(listener)
if (wrapped) port.off("message", wrapped)
},
postMessage: port.postMessage.bind(port),
start: port.start.bind(port),
close: port.close.bind(port),
} as unknown as MessagePortMain
}
// SAFETY: the transport uses only these methods of a port.
// oxlint-disable-next-line anti-slop/no-chained-type-assertions -- see SAFETY above
return fake as unknown as MessagePortMain
}
@@ -39,8 +39,10 @@ export const IpcServerProtocolLive = Layer.unwrap(
const disconnect = Effect.fnUntraced(function* (id: number) {
const binding = bindings.get(id)
if (!binding) return
bindings.delete(id)
if (senderBindings.get(binding.sender.id) === id) senderBindings.delete(binding.sender.id)
binding.port.off("message", binding.onMessage)
binding.port.off("close", binding.onClose)
@@ -52,16 +54,23 @@ export const IpcServerProtocolLive = Layer.unwrap(
const bind = Effect.fnUntraced(function* (sender: WebContents, port: MessagePortMain) {
const previous = senderBindings.get(sender.id)
if (previous !== undefined) yield* disconnect(previous)
if (sender.isDestroyed()) {
port.close()
return
}
const id = nextClientId++
const onMessage = (event: Electron.MessageEvent) => {
// SAFETY: the other end of this port is the renderer's ipc-client, which posts only RPC client
// messages; the RPC server decodes each request's payload with that RPC's schema.
Queue.offerUnsafe(inbound, [id, event.data as RpcMessage.FromClientEncoded] as const)
}
const onClose = () => runFork(disconnect(id))
const unbindEvents = yield* bindIpcEvents(sender.id)
const binding = { id, sender, port, onMessage, onClose, unbindEvents }
+16 -4
View File
@@ -22,6 +22,7 @@ import { DesktopCli } from "./service/desktop-cli"
import { getLastFocusedWindow } from "./windows"
const services = Layer.mergeAll(DesktopFiles.layer, Extensions.layer)
const handlers = Layer.mergeAll(
appHandlers,
storageHandlers,
@@ -31,6 +32,7 @@ const handlers = Layer.mergeAll(
eventHandlers,
extensionHandlers,
)
export const layer = RpcServer.layer(DesktopRpcs, { disableFatalDefects: true }).pipe(
Layer.provide(handlers),
Layer.provideMerge(IpcServerProtocolLive),
@@ -42,9 +44,11 @@ export const registerIpcHandlers = Effect.gen(function* () {
const lifecycle = yield* ApplicationLifecycle.Service
const desktopCli = yield* DesktopCli.Service
const runFork = Effect.runForkWith(yield* Effect.context())
const menu = {
trigger: (id: string) => {
const win = getLastFocusedWindow()
if (win) sendMenuCommand(win, id)
},
installCli: () => runFork(showCliInstaller(desktopCli)),
@@ -52,26 +56,34 @@ export const registerIpcHandlers = Effect.gen(function* () {
openExternal: (url: string) => runFork(openExternalURL(url)),
relaunch: lifecycle.relaunch,
}
const wire = (_event: Electron.Event, win: BrowserWindow) => {
const wire = (win: BrowserWindow) => {
win.webContents.on("before-input-event", (_event, input) => {
if (input.type !== "keyDown" || input.key !== "Escape") return
win.webContents.send(DragCancelEvent)
})
const post = () => {
if (win.isDestroyed() || win.webContents.isDestroyed()) return
const channel = new MessageChannelMain()
handoff.bind(win.webContents, channel.port1)
win.webContents.postMessage(IpcTransportPort, null, [channel.port2])
}
win.webContents.on("did-finish-load", post)
// The first window starts loading before the layers exist and may already be done.
if (!win.webContents.isLoading() && win.webContents.getURL()) post()
}
const onWindowCreated = (_event: Electron.Event, win: BrowserWindow) => wire(win)
yield* Effect.sync(() => {
app.on("browser-window-created", wire)
BrowserWindow.getAllWindows().forEach((win) => wire({} as Electron.Event, win))
app.on("browser-window-created", onWindowCreated)
BrowserWindow.getAllWindows().forEach((win) => wire(win))
})
yield* Effect.addFinalizer(() => Effect.sync(() => app.off("browser-window-created", wire)))
yield* Effect.addFinalizer(() => Effect.sync(() => app.off("browser-window-created", onWindowCreated)))
return {
installMenu: () => createMenu(menu),
}
+55 -4
View File
@@ -11,33 +11,47 @@ import { IpcTransportPort, omitUndefined } from "../shared/ipc-transport"
// out of the renderer's initial module graph is worth about a third of its startup script.
type EventTag = DesktopEvent["_tag"]
type InvokeTag = Exclude<keyof DesktopRpcClient, "DesktopEvents">
type InvokeArgs<Tag extends InvokeTag> = Parameters<DesktopRpcClient[Tag]>
type InvokeResult<Tag extends InvokeTag> =
ReturnType<DesktopRpcClient[Tag]> extends Effect.Effect<infer Value, unknown> ? Value : never
type EventValue<Tag extends EventTag> = Extract<DesktopEvent, { readonly _tag: Tag }>
type Pending = {
// SAFETY: the reply is the RPC's success value as the trusted main process encoded it; see the note above.
// oxlint-disable-next-line anti-slop/no-unknown-parameters -- see SAFETY above
readonly resolve: (value: unknown) => void
readonly reject: (error: unknown) => void
readonly reject: (cause: unknown) => void
readonly chunk?: (values: ReadonlyArray<unknown>) => void
}
const pending = new Map<number, Pending>()
const listeners = new Map<EventTag, Set<(value: unknown) => void>>()
const listeners = new Map<EventTag, Set<(value: DesktopEvent) => void>>()
// SAFETY: pagehide only waits for these callbacks to settle and ignores what they resolve to.
// oxlint-disable-next-line anti-slop/no-unknown-returns -- see SAFETY above
const beforeDispose = new Set<() => Promise<unknown> | void>()
let nextId = 0
const port = new Promise<MessagePort>((resolve) => {
const onMessage = (event: MessageEvent) => {
if (event.source !== window || event.data !== IpcTransportPort) return
const value = event.ports[0]
if (!value) return
window.removeEventListener("message", onMessage)
// SAFETY: the port comes from the trusted main process, whose RPC server posts only its own messages.
value.addEventListener("message", (message) => receive(value, message.data as RpcMessage.FromServerEncoded))
value.start()
resolve(value)
}
window.addEventListener("message", onMessage)
})
@@ -49,15 +63,20 @@ window.addEventListener(
)
void request("DesktopEvents", null, (values) => {
// SAFETY: DesktopEvents streams desktop events, whose encoded fields are JSON-native and read as decoded.
for (const value of values as ReadonlyArray<DesktopEvent>) listeners.get(value._tag)?.forEach((fn) => fn(value))
})
// SAFETY: as for `beforeDispose`, the callback's result is only awaited, never read.
// oxlint-disable-next-line anti-slop/no-unknown-returns -- see SAFETY above
export function onBeforeDispose(callback: () => Promise<unknown> | void) {
beforeDispose.add(callback)
return () => beforeDispose.delete(callback)
}
export function invoke<Tag extends InvokeTag>(tag: Tag, ...payload: InvokeArgs<Tag>): Promise<InvokeResult<Tag>> {
// SAFETY: main replies with the success value of `tag`'s RPC; its encoding is JSON-native or bytes, as decoded.
return request(tag, payload[0] ?? null) as Promise<InvokeResult<Tag>>
}
@@ -71,37 +90,55 @@ export function cancellable<Tag extends InvokeTag>(
payload: InvokeArgs<Tag>[0],
signal: AbortSignal | undefined,
): Promise<InvokeResult<Tag>> {
// SAFETY: as in `invoke`, the reply is the success value of `tag`'s RPC.
return request(tag, payload ?? null, undefined, signal) as Promise<InvokeResult<Tag>>
}
export function listen<Tag extends EventTag>(tag: Tag, listener: (value: EventValue<Tag>) => void) {
const callback = listener as (value: unknown) => void
// SAFETY: events dispatch by tag, so this listener only receives `Tag` events.
const callback = listener as (value: DesktopEvent) => void
const callbacks = listeners.get(tag) ?? new Set()
callbacks.add(callback)
listeners.set(tag, callbacks)
return () => {
callbacks.delete(callback)
if (callbacks.size === 0) listeners.delete(tag)
}
}
function request(tag: string, payload: unknown, chunk?: Pending["chunk"], signal?: AbortSignal) {
// SAFETY: this side speaks the RPC wire format by hand, whose messages are plain tagged objects. Effect's Match,
// Predicate, and message constructors would load the Effect runtime into the renderer; see the note above.
/* oxlint-disable anti-slop-effect/no-manual-tag-comparison, anti-slop-effect/no-manual-tagged-construction -- see SAFETY above */
function request(
tag: InvokeTag | "DesktopEvents",
payload: InvokeArgs<InvokeTag>[0] | null,
chunk?: Pending["chunk"],
signal?: AbortSignal,
) {
const id = nextId++
return new Promise<unknown>((resolve, reject) => {
if (signal?.aborted) return reject(signal.reason)
const abort = () => {
if (!pending.delete(id)) return
reject(signal?.reason)
void port.then((p) => p.postMessage({ _tag: "Interrupt", requestId: id } satisfies RpcMessage.InterruptEncoded))
}
signal?.addEventListener("abort", abort, { once: true })
const settle =
<Value>(callback: (value: Value) => void) =>
(value: Value) => {
signal?.removeEventListener("abort", abort)
callback(value)
}
pending.set(id, { resolve: settle(resolve), reject: settle(reject), chunk })
const message: RpcMessage.RequestEncoded = {
_tag: "Request",
id,
@@ -109,6 +146,7 @@ function request(tag: string, payload: unknown, chunk?: Pending["chunk"], signal
payload: omitUndefined(payload),
headers: [],
}
void port.then((p) => p.postMessage(message))
})
}
@@ -118,24 +156,33 @@ function receive(p: MessagePort, message: RpcMessage.FromServerEncoded) {
case "Chunk": {
pending.get(Number(message.requestId))?.chunk?.(message.values)
p.postMessage({ _tag: "Ack", requestId: message.requestId } satisfies RpcMessage.AckEncoded)
return
}
case "Exit": {
const id = Number(message.requestId)
const entry = pending.get(id)
pending.delete(id)
if (!entry) return
if (message.exit._tag === "Success") return entry.resolve(message.exit.value)
return entry.reject(failure(message.exit.cause))
}
case "Defect": {
const error = new Error("Desktop IPC defect", { cause: message.defect })
pending.forEach((entry) => entry.reject(error))
pending.clear()
return
}
case "ClientProtocolError": {
console.error("[desktop-ipc] protocol error", message.error)
return
}
}
@@ -145,8 +192,12 @@ function receive(p: MessagePort, message: RpcMessage.FromServerEncoded) {
// and interrupts surface as errors.
function failure(cause: ReadonlyArray<{ readonly _tag: string; readonly error?: unknown; readonly defect?: unknown }>) {
const failed = cause.find((item) => item._tag === "Fail")
if (failed) return failed.error
const died = cause.find((item) => item._tag === "Die")
if (died) return new Error("Desktop IPC handler failed", { cause: died.defect })
return new Error("Desktop IPC request interrupted")
}
/* oxlint-enable anti-slop-effect/no-manual-tag-comparison, anti-slop-effect/no-manual-tagged-construction */
+7
View File
@@ -8,12 +8,19 @@ import { StorageRpcs } from "./ipc-rpc/storage"
import { WindowRpcs } from "./ipc-rpc/window"
export { AppRpcs } from "./ipc-rpc/app"
export { EventRpcs } from "./ipc-rpc/events"
export { ExtensionRpcs } from "./ipc-rpc/extensions"
export { FileRpcs } from "./ipc-rpc/files"
export { MenuRpcs } from "./ipc-rpc/menu"
export { StorageRpcs } from "./ipc-rpc/storage"
export { WindowRpcs } from "./ipc-rpc/window"
export const DesktopRpcs = AppRpcs.merge(StorageRpcs, FileRpcs, WindowRpcs, MenuRpcs, EventRpcs, ExtensionRpcs)
export type DesktopRpcClient = RpcClient.FromGroup<typeof DesktopRpcs, RpcClientError.RpcClientError>
@@ -6,38 +6,50 @@ const ServerReadyData = Schema.Struct({
})
export const AppAwaitInitialization = Rpc.make("AppAwaitInitialization", { success: ServerReadyData })
export const AppReconnectService = Rpc.make("AppReconnectService", { success: ServerReadyData })
export const AppConsumeInitialDeepLinks = Rpc.make("AppConsumeInitialDeepLinks", {
success: Schema.Array(Schema.String),
})
export const AppGetDefaultServerUrl = Rpc.make("AppGetDefaultServerUrl", {
success: Schema.NullOr(Schema.String),
})
export const AppSetDefaultServerUrl = Rpc.make("AppSetDefaultServerUrl", {
payload: { url: Schema.NullOr(Schema.String) },
})
export const AppIsFirstLaunchOnboardingPending = Rpc.make("AppIsFirstLaunchOnboardingPending", {
success: Schema.Boolean,
})
export const AppFinishFirstLaunchOnboarding = Rpc.make("AppFinishFirstLaunchOnboarding", {
payload: { createDefaultProject: Schema.Boolean },
success: Schema.NullOr(Schema.String),
})
export const AppCheckAppExists = Rpc.make("AppCheckAppExists", {
payload: { appName: Schema.String },
success: Schema.Boolean,
})
export const AppResolveAppPath = Rpc.make("AppResolveAppPath", {
payload: { appName: Schema.String },
success: Schema.NullOr(Schema.String),
})
export const AppSetBackgroundColor = Rpc.make("AppSetBackgroundColor", {
payload: { color: Schema.String },
})
export const AppExportDebugLogs = Rpc.make("AppExportDebugLogs", { success: Schema.String })
export const AppSetForceFocus = Rpc.make("AppSetForceFocus", {
payload: { enabled: Schema.Boolean },
})
export const AppRecordFatalRendererError = Rpc.make("AppRecordFatalRendererError", {
payload: {
error: Schema.Struct({
@@ -49,10 +61,13 @@ export const AppRecordFatalRendererError = Rpc.make("AppRecordFatalRendererError
}),
},
})
export const AppSetNativeTranslations = Rpc.make("AppSetNativeTranslations", {
payload: { value: Schema.Unknown },
})
export const AppRelaunch = Rpc.make("AppRelaunch")
export const AppRpcs = RpcGroup.make(
AppAwaitInitialization,
AppReconnectService,
@@ -3,64 +3,80 @@ import { Rpc, RpcGroup } from "effect/unstable/rpc"
import { Transferable } from "effect/unstable/workers"
const OptionalString = Schema.optional(Schema.String)
const PickerOptions = Schema.Struct({
multiple: Schema.optional(Schema.Boolean),
title: OptionalString,
defaultPath: OptionalString,
})
const FilePickerOptions = Schema.Struct({
multiple: Schema.optional(Schema.Boolean),
title: OptionalString,
defaultPath: OptionalString,
extensions: Schema.optional(Schema.Array(Schema.String)),
})
const SavePickerOptions = Schema.Struct({ title: OptionalString, defaultPath: OptionalString })
const PickedFiles = Schema.Struct({
token: Schema.String,
files: Schema.Array(Schema.Struct({ path: Schema.String, name: Schema.String, size: Schema.Number })),
})
const ClipboardImage = Schema.Struct({ buffer: Transferable.Uint8Array, width: Schema.Number, height: Schema.Number })
export const FilesOpenDirectoryPicker = Rpc.make("FilesOpenDirectoryPicker", {
payload: { options: Schema.optional(PickerOptions) },
success: Schema.NullOr(Schema.Union([Schema.String, Schema.Array(Schema.String)])),
})
export const FilesOpenFilePicker = Rpc.make("FilesOpenFilePicker", {
payload: { options: Schema.optional(FilePickerOptions) },
success: Schema.NullOr(PickedFiles),
})
export const FilesReadPickedFile = Rpc.make("FilesReadPickedFile", {
payload: { token: Schema.String, path: Schema.String },
success: Transferable.Uint8Array,
})
export const FilesReleasePickedFiles = Rpc.make("FilesReleasePickedFiles", {
payload: { token: Schema.String },
})
export const FilesSaveFile = Rpc.make("FilesSaveFile", {
payload: { options: SavePickerOptions, content: Schema.String },
success: Schema.Boolean,
})
export const FilesOpenExternal = Rpc.make("FilesOpenExternal", {
payload: { url: Schema.String },
})
export const FilesOpenBrowser = Rpc.make("FilesOpenBrowser", {
payload: { url: Schema.String },
success: Schema.Boolean,
})
export const FilesOpenLocalFile = Rpc.make("FilesOpenLocalFile", {
payload: { url: Schema.String },
})
export const FilesOpenPath = Rpc.make("FilesOpenPath", {
payload: { path: Schema.String, application: Schema.optional(Schema.String) },
success: Schema.NullOr(Schema.String),
})
export const FilesRevealPath = Rpc.make("FilesRevealPath", {
payload: { path: Schema.String },
success: Schema.Boolean,
})
export const FilesReadClipboardImage = Rpc.make("FilesReadClipboardImage", {
success: Schema.NullOr(ClipboardImage),
})
export const FilesWriteClipboardText = Rpc.make("FilesWriteClipboardText", {
payload: { text: Schema.String },
})
@@ -26,4 +26,5 @@ const DesktopMenuAction = Schema.Literals([
export const MenuRunAction = Rpc.make("MenuRunAction", {
payload: { action: DesktopMenuAction },
})
export const MenuRpcs = RpcGroup.make(MenuRunAction)
@@ -6,6 +6,7 @@ export const StorageItems = Rpc.make("StorageItems", {
payload: { name: Schema.String },
success: Schema.Struct({ items: Schema.Record(Schema.String, Schema.String), revision: Schema.Number }),
})
export const StorageUpdate = Rpc.make("StorageUpdate", {
payload: {
name: Schema.String,
@@ -14,20 +15,26 @@ export const StorageUpdate = Rpc.make("StorageUpdate", {
},
success: Schema.Number,
})
export const StorageClear = Rpc.make("StorageClear", { payload: { name: Schema.String } })
export const DraftsGet = Rpc.make("DraftsGet", {
payload: { key: Schema.String },
success: Schema.NullOr(Schema.String),
})
export const DraftsSet = Rpc.make("DraftsSet", {
payload: { key: Schema.String, value: Schema.String, strict: Schema.Boolean },
success: Schema.Array(Schema.String),
})
export const DraftsDelete = Rpc.make("DraftsDelete", { payload: { key: Schema.String } })
export const DraftsPutBlob = Rpc.make("DraftsPutBlob", {
payload: { data: Transferable.Uint8Array },
success: Schema.String,
})
export const DraftsGetBlob = Rpc.make("DraftsGetBlob", {
payload: { id: Schema.String },
success: Schema.NullOr(Transferable.Uint8Array),
@@ -2,20 +2,29 @@ import { Schema } from "effect"
import { Rpc, RpcGroup } from "effect/unstable/rpc"
export const WindowThemeReady = Rpc.make("WindowThemeReady")
export const WindowGetFocused = Rpc.make("WindowGetFocused", { success: Schema.Boolean })
export const WindowGetFullscreen = Rpc.make("WindowGetFullscreen", { success: Schema.Boolean })
export const WindowSetFocus = Rpc.make("WindowSetFocus")
export const WindowShow = Rpc.make("WindowShow")
export const WindowGetZoomFactor = Rpc.make("WindowGetZoomFactor", { success: Schema.Number })
export const WindowSetZoomFactor = Rpc.make("WindowSetZoomFactor", {
payload: { factor: Schema.Number },
})
export const WindowGetPinchZoomEnabled = Rpc.make("WindowGetPinchZoomEnabled", {
success: Schema.Boolean,
})
export const WindowSetPinchZoomEnabled = Rpc.make("WindowSetPinchZoomEnabled", {
payload: { enabled: Schema.Boolean },
})
export const WindowSetTitlebar = Rpc.make("WindowSetTitlebar", {
payload: {
theme: Schema.Struct({
@@ -24,6 +33,7 @@ export const WindowSetTitlebar = Rpc.make("WindowSetTitlebar", {
}),
},
})
export const WindowRpcs = RpcGroup.make(
WindowThemeReady,
WindowGetFocused,
@@ -14,6 +14,7 @@ export const createBrowserNetwork = Effect.fn("BrowserNetwork.create")(function*
partition: string
}) {
const options = { location: input.location }
const proxy = yield* Effect.acquireRelease(
Effect.tryPromise(() =>
BrowserProxy.make({
@@ -37,6 +38,7 @@ export const createBrowserNetwork = Effect.fn("BrowserNetwork.create")(function*
),
(proxy) => Effect.promise(() => proxy.close()),
)
const partition = session.fromPartition(input.partition)
yield* Effect.addFinalizer(() => Effect.promise(() => partition.closeAllConnections()))
// This is the browser's private partition, not the app/API connection. Never
@@ -45,6 +47,7 @@ export const createBrowserNetwork = Effect.fn("BrowserNetwork.create")(function*
partition.setProxy({ mode: "fixed_servers", proxyRules: proxy.url, proxyBypassRules: "<-loopback>" }),
)
yield* Effect.tryPromise(() => partition.closeAllConnections())
return {
attach(contents: WebContents) {
const login = (
@@ -64,8 +67,10 @@ export const createBrowserNetwork = Effect.fn("BrowserNetwork.create")(function*
event.preventDefault()
callback(proxy.credentials.username, proxy.credentials.password)
}
contents.on("login", login)
contents.setWebRTCIPHandlingPolicy("disable_non_proxied_udp")
return () => contents.off("login", login)
},
}
@@ -25,6 +25,7 @@ const request = Effect.fn("test.askpass.request")(function* (
],
{ concurrency: "unbounded" },
)
return result.text
}, Effect.scoped)
@@ -32,11 +33,13 @@ it.live(
"per-prompt replies are isolated, including confirmation and OTP",
Effect.gen(function* () {
const prompts = yield* Queue.unbounded<{ id: string; text: string; confirm: boolean }>()
const bridge = yield* createAskpass({
binary: "unused",
prompt: (prompt) => Queue.offer(prompts, prompt).pipe(Effect.asVoid),
clear: () => Effect.void,
})
const password = yield* request(bridge.env, "Password:").pipe(Effect.forkScoped)
const first = yield* Queue.take(prompts)
expect(first.text).toBe("Password:")
@@ -56,11 +59,13 @@ it.live(
const parent = yield* Scope.Scope
const scope = yield* Scope.fork(parent)
const prompted = yield* Deferred.make<void>()
const bridge = yield* createAskpass({
binary: "unused",
prompt: () => Deferred.succeed(prompted, undefined).pipe(Effect.asVoid),
clear: () => Effect.void,
}).pipe(Scope.provide(scope))
expect(yield* request({ ...bridge.env, OPENCODE_SSH_ASKPASS_TOKEN: "incorrect" }, "Password:")).toBe("")
const reply = yield* request(bridge.env, "Trust fingerprint?", true).pipe(Effect.forkScoped)
yield* Deferred.await(prompted)
+10 -2
View File
@@ -1,5 +1,5 @@
import { NodeSocketServer } from "@effect/platform-node"
import { Deferred, Effect, Fiber, Schema, Semaphore } from "effect"
import { Deferred, Effect, Fiber, Predicate, Schema, Semaphore } from "effect"
import { randomUUID } from "node:crypto"
import { SshFailure } from "./command"
@@ -16,24 +16,31 @@ export const createAskpass = Effect.fn("Ssh.askpass")(function* (input: {
const pending = new Map<string, Deferred.Deferred<string>>()
const prompts = yield* Semaphore.make(1)
const server = yield* NodeSocketServer.make({ host: "127.0.0.1", port: 0 }).pipe(Effect.mapError(SshFailure.from))
if (server.address._tag !== "TcpAddress") return yield* Effect.fail(new SshFailure("connection"))
if (!Predicate.isTagged(server.address, "TcpAddress")) return yield* Effect.fail(new SshFailure("connection"))
const serving = yield* server
.run((socket) =>
Effect.gen(function* () {
const request = yield* Deferred.make<string, SshFailure>()
const state = { buffer: "", received: false }
const reader = yield* socket
.runString((chunk) => {
if (state.received) return Effect.fail(new SshFailure("connection"))
state.buffer += chunk
if (state.buffer.length > 16_384) return Effect.fail(new SshFailure("connection"))
if (!state.buffer.includes("\n")) return Effect.void
state.received = true
return Deferred.succeed(request, state.buffer.trim())
})
.pipe(Effect.ensuring(Deferred.fail(request, new SshFailure("connection"))), Effect.forkScoped)
const message = yield* Deferred.await(request).pipe(Effect.flatMap(Schema.decodeUnknownEffect(Request)))
if (message.token !== token) return
// One scoped waiter per helper invocation. Disconnecting a helper or closing
@@ -75,6 +82,7 @@ export const createAskpass = Effect.fn("Ssh.askpass")(function* (input: {
closed: Fiber.join(serving),
respond: Effect.fn("Ssh.askpass.respond")(function* (id: string, value: string) {
const response = pending.get(id)
if (response) yield* Deferred.succeed(response, value)
}),
}
+23 -2
View File
@@ -1,4 +1,4 @@
import { Effect, Schema } from "effect"
import { Effect, Option, Schema } from "effect"
import { HttpClient } from "effect/unstable/http"
import { parseTarget, quote, runSsh, sshArgs, SshFailure } from "./command"
import { RemoteCli } from "./remote-cli"
@@ -42,13 +42,17 @@ export function parseRegistration(output: string) {
.split(/\r?\n/)
.findLast((line) => line.startsWith("OPENCODE_SSH_STATUS="))
?.slice("OPENCODE_SSH_STATUS=".length)
if (!status) return undefined
for (const match of output.matchAll(
/OPENCODE_SSH_REGISTRATION_BEGIN\r?\n([\s\S]*?)\r?\nOPENCODE_SSH_REGISTRATION_END/g,
)) {
const result = Schema.decodeUnknownOption(Registration)(match[1])
if (result._tag === "Some" && result.value.url === status) return result.value
if (Option.isSome(result) && result.value.url === status) return result.value
}
return undefined
}
@@ -58,8 +62,10 @@ export function binaryPath(version: string) {
function connectionAddress(address: string, password: string) {
const url = new URL(address)
if (url.protocol !== "http:" || !["127.0.0.1", "localhost", "0.0.0.0", "[::]", "[::1]"].includes(url.hostname))
throw new SshFailure("service")
return {
host: url.hostname === "[::1]" ? "[::1]" : "127.0.0.1",
port: Number(url.port || 80),
@@ -81,30 +87,39 @@ export const bootstrap = Effect.fn("Ssh.bootstrap")(function* (input: {
env: input.env,
stdin: script,
})
yield* input.stage("checking")
const registered = parseRegistration(yield* run(discoverScript))
if (registered && (input.development || registered.version === input.version)) {
yield* input.stage("starting")
return yield* Effect.try({
try: () => connectionAddress(registered.url, registered.password),
catch: SshFailure.from,
})
}
if (registered && !input.replace) return yield* Effect.fail(new SshFailure("version", registered.version))
const destination = yield* Effect.try({ try: () => binaryPath(input.version), catch: SshFailure.from })
const existing = yield* run(RemoteCli.versionScript(`"${destination}"`))
const staged = RemoteCli.parseVersion(existing) === input.version
// Source worktree versions are unpublished. Use the installer's beta channel
// while retaining support for explicitly staged, matching development builds.
const version =
input.development && !staged ? yield* RemoteCli.latestBeta().pipe(Effect.mapError(SshFailure.from)) : input.version
const setup = { version, directory: `.opencode/desktop-ssh/${version}` }
if (!staged) {
const output = yield* run(RemoteCli.probeScript).pipe(Effect.mapError(() => new SshFailure("platform")))
const target = output
.split(/\r?\n/)
.findLast((line) => line.startsWith("OPENCODE_REMOTE_TARGET="))
?.split("=")[1]
const url = yield* Effect.try({ try: () => RemoteCli.archiveUrl(target ?? "", version), catch: SshFailure.from })
yield* input.stage("downloading")
yield* run(RemoteCli.installScript({ ...setup, source: { type: "download", url } })).pipe(
@@ -113,6 +128,7 @@ export const bootstrap = Effect.fn("Ssh.bootstrap")(function* (input: {
yield* input.stage("uploading")
const http = yield* HttpClient.HttpClient
const response = yield* http.get(url).pipe(Effect.mapError(SshFailure.from))
if (response.status < 200 || response.status >= 300)
return yield* Effect.fail(
new SshFailure(
@@ -121,6 +137,7 @@ export const bootstrap = Effect.fn("Ssh.bootstrap")(function* (input: {
),
)
const archive = new Uint8Array(yield* response.arrayBuffer.pipe(Effect.mapError(SshFailure.from)))
// The upload uses stdin; the script itself must be the remote command.
return yield* runSsh({
args: [
@@ -135,11 +152,15 @@ export const bootstrap = Effect.fn("Ssh.bootstrap")(function* (input: {
),
)
}
yield* input.stage("starting")
const registration = parseRegistration(yield* run(startScript(version, input.replace)))
if (!registration) return yield* Effect.fail(new SshFailure("service"))
if (!input.development && registration.version !== input.version)
return yield* Effect.fail(new SshFailure("version", registration.version))
return yield* Effect.try({
try: () => connectionAddress(registration.url, registration.password),
catch: SshFailure.from,
+40 -11
View File
@@ -1,4 +1,4 @@
import { Effect, PlatformError, Schema, Stream } from "effect"
import { Effect, PlatformError, Predicate, Schema, Stream } from "effect"
import { ChildProcess, ChildProcessSpawner } from "effect/unstable/process"
import { RemoteCli } from "./remote-cli"
@@ -23,17 +23,19 @@ export class SshFailure extends Schema.TaggedError<SshFailure>()("SshFailure", {
return this.detail
}
static from(this: void, error: unknown) {
if (error instanceof RemoteCli.Failure) return new SshFailure(error.code, error.detail)
static from(this: void, cause: unknown) {
if (cause instanceof RemoteCli.Failure) return new SshFailure(cause.code, cause.detail)
if (
error instanceof PlatformError.PlatformError &&
error.reason._tag === "NotFound" &&
error.reason.method === "spawn"
cause instanceof PlatformError.PlatformError &&
Predicate.isTagged(cause.reason, "NotFound") &&
cause.reason.method === "spawn"
)
return new SshFailure("ssh-missing", error.message)
return error instanceof SshFailure
? error
: new SshFailure("connection", error instanceof Error ? error.message : String(error))
return new SshFailure("ssh-missing", cause.message)
return cause instanceof SshFailure
? cause
: new SshFailure("connection", cause instanceof Error ? cause.message : String(cause))
}
}
@@ -44,37 +46,48 @@ export function quote(value: string) {
export function parseTarget(input: string) {
const tokens: string[] = []
const state = { word: "", quote: "", started: false }
for (let i = 0; i < input.length; i++) {
const c = input[i] ?? ""
if (c === "\n" || c === "\r" || c === "\0") throw new SshFailure("input")
if (c === "\\" && state.quote !== "'" && i + 1 < input.length && /[\s\\"']/.test(input[i + 1] ?? "")) {
state.word += input[++i]
state.started = true
continue
}
if (state.quote) {
if (c === state.quote) state.quote = ""
else state.word += c
continue
}
if (c === "'" || c === '"') {
state.quote = c
state.started = true
continue
}
if (/\s/.test(c)) {
if (state.started) tokens.push(state.word)
state.word = ""
state.started = false
continue
}
state.word += c
state.started = true
}
if (state.quote) throw new SshFailure("input")
if (state.started) tokens.push(state.word)
if (tokens[0] === "ssh") tokens.shift()
const args: string[] = []
const options = new Set([
"hostname",
"user",
@@ -87,25 +100,36 @@ export function parseTarget(input: string) {
"connecttimeout",
"addressfamily",
])
while (tokens[0]?.startsWith("-")) {
const token = tokens.shift() ?? ""
if (["-4", "-6", "-C", "-A", "-a"].includes(token)) {
args.push(token)
continue
}
const flag = token.slice(0, 2)
if (!["-p", "-l", "-i", "-F", "-J", "-o"].includes(flag)) throw new SshFailure("input")
const value = token.length > 2 ? token.slice(2) : tokens.shift()
if (!value || value.startsWith("-")) throw new SshFailure("input")
if (flag === "-p" && (!/^\d+$/.test(value) || Number(value) < 1 || Number(value) > 65535))
throw new SshFailure("input")
if (flag === "-o" && !options.has((value.split(/[=\s]/)[0] ?? "").toLowerCase())) throw new SshFailure("input")
args.push(flag, value)
}
const host = tokens[0]
if (tokens.length !== 1 || !host || !/^[a-zA-Z0-9_@.:[\]%-]+$/.test(host) || host.startsWith("-"))
throw new SshFailure("input")
if (host.includes("@") && host.slice(0, host.lastIndexOf("@")).includes(":")) throw new SshFailure("input")
return { host, args }
}
@@ -159,6 +183,7 @@ export const runSsh = Effect.fn("Ssh.run")(function* (input: {
timeout?: number
}) {
const spawner = yield* ChildProcessSpawner.ChildProcessSpawner
return yield* Effect.gen(function* () {
const child = yield* spawner.spawn(
ChildProcess.make(sshExecutable(), input.args, {
@@ -172,12 +197,13 @@ export const runSsh = Effect.fn("Ssh.run")(function* (input: {
? "ignore"
: {
stream: Stream.make(
typeof input.stdin === "string" ? new TextEncoder().encode(input.stdin) : input.stdin,
Predicate.isString(input.stdin) ? new TextEncoder().encode(input.stdin) : input.stdin,
),
endOnDone: true,
},
}),
)
const output = yield* Effect.all(
{
stdout: child.stdout.pipe(
@@ -198,8 +224,10 @@ export const runSsh = Effect.fn("Ssh.run")(function* (input: {
},
{ concurrency: "unbounded" },
)
if (output.code !== 0)
return yield* Effect.fail(new SshFailure("connection", commandFailureDetail(output.code, output)))
return output.stdout
}).pipe(Effect.scoped, Effect.timeout(input.timeout ?? 600_000), Effect.mapError(SshFailure.from))
})
@@ -210,5 +238,6 @@ export function commandFailureDetail(code: number | null, output: { stdout: stri
const stdout = output.stdout
.replace(/OPENCODE_SSH_REGISTRATION_BEGIN[\s\S]*?(?:OPENCODE_SSH_REGISTRATION_END|$)/g, "")
.trim()
return [output.stderr.trim(), stdout].filter(Boolean).join("\n") || JSON.stringify({ exitCode: code })
}
+72 -13
View File
@@ -8,6 +8,7 @@ import {
Fiber,
FileSystem,
Path,
Predicate,
PubSub,
Ref,
Schedule,
@@ -27,6 +28,7 @@ type Connection = {
ready: Deferred.Deferred<SshHttp | null>
respond?: (id: string, value: string) => Effect.Effect<void>
}
type Attempt = Connection & { fiber: Fiber.Fiber<void> }
export const createSshController = Effect.fn("Ssh.controller")(function* (input: {
@@ -49,12 +51,14 @@ export const createSshController = Effect.fn("Ssh.controller")(function* (input:
const items = new Map<string, SshItem>(
input.configs.map((config) => [config.id, { config, saved: true, stage: "disconnected", detail: "" }]),
)
const configs = new Map(input.configs.map((config) => [config.id, config]))
const attempts = new Map<string, Attempt>()
const paused = new Set(items.keys())
const failures = new Map<string, number>()
const lifecycle = { closed: false }
const emit = PubSub.publish(changed, undefined).pipe(Effect.asVoid)
const state = (owner?: number): Effect.Effect<Pick<SshState, "servers">> =>
Effect.sync(() => ({
servers: [...items.values()].map((item) => ({
@@ -66,12 +70,22 @@ export const createSshController = Effect.fn("Ssh.controller")(function* (input:
attempts.get(item.config.id)?.owner !== owner,
})),
}))
const update = Effect.fnUntraced(function* (id: string, value: Partial<SshItem>) {
const item = items.get(id)
if (!item) return
items.set(id, { ...item, ...value })
yield* emit
})
// A saved server's item takes its current saved config along with the change.
const withSavedConfig = (id: string, value: Partial<SshItem>): Partial<SshItem> => {
const config = configs.get(id)
return config ? { ...value, config } : value
}
const run = (options: Parameters<typeof runSsh>[0]) =>
runSsh(options).pipe(Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, spawner))
@@ -79,14 +93,17 @@ export const createSshController = Effect.fn("Ssh.controller")(function* (input:
const target = yield* Effect.try({ try: () => parseTarget(config.target), catch: SshFailure.from })
const directory = yield* fs.makeTempDirectoryScoped({ prefix: "oc-ssh-" })
const control = path.join(directory, "s")
const helper =
input.command && input.command.length > 1 && process.platform !== "win32"
? path.join(directory, "askpass")
: input.binary
if (helper !== input.binary)
yield* fs.writeFileString(helper, `#!/bin/sh\nexec ${input.command?.map(quote).join(" ")} "$@"\n`, {
mode: 0o700,
})
if (process.platform !== "win32") {
target.args.unshift("-o", "ControlMaster=auto", "-o", "ControlPersist=60", "-o", `ControlPath=${control}`)
// Close only our local SSH master. The remote OpenCode service owns its
@@ -97,6 +114,7 @@ export const createSshController = Effect.fn("Ssh.controller")(function* (input:
}
const authentication = yield* Deferred.make<void>()
const askpass = yield* createAskpass({
binary: helper,
prompt: Effect.fnUntraced(function* (prompt) {
@@ -104,8 +122,10 @@ export const createSshController = Effect.fn("Ssh.controller")(function* (input:
paused.add(config.id)
yield* update(config.id, { stage: "authentication" })
yield* Deferred.succeed(authentication, undefined)
return
}
yield* update(config.id, { stage: "authentication", prompt })
}),
clear: (id) =>
@@ -113,22 +133,27 @@ export const createSshController = Effect.fn("Ssh.controller")(function* (input:
? update(config.id, { prompt: undefined, stage: "connecting" })
: Effect.void,
})
connection.respond = askpass.respond
yield* Effect.gen(function* () {
const resolved = yield* run({ args: [...sshArgs(target), "-G", target.host], timeout: 10_000 }).pipe(
Effect.orElseSucceed(() => ""),
)
const fields = new Map(
resolved.split(/\r?\n/).map((line) => {
const separator = line.indexOf(" ")
return [line.slice(0, separator), line.slice(separator + 1)] as const
}),
)
if (fields.has("hostname"))
yield* update(config.id, {
destination: `${fields.get("user") ?? ""}@${fields.get("hostname")}:${fields.get("port") ?? "22"}`,
})
const remote = yield* bootstrap({
target,
version: input.version,
@@ -140,8 +165,10 @@ export const createSshController = Effect.fn("Ssh.controller")(function* (input:
Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, spawner),
Effect.provideService(HttpClient.HttpClient, httpClient),
)
const port = yield* freePort
const http = { url: `http://127.0.0.1:${port}`, password: remote.password }
const tunnel = yield* spawner.spawn(
ChildProcess.make(sshExecutable(), tunnelArgs(target, port, remote), {
env: askpass.env,
@@ -153,19 +180,24 @@ export const createSshController = Effect.fn("Ssh.controller")(function* (input:
forceKillAfter: "2 seconds",
}),
)
const detail = yield* Ref.make("")
const stderr = yield* tunnel.stderr.pipe(
Stream.decodeText(),
Stream.runForEach((text) => Ref.update(detail, (tail) => (tail + text).slice(-8192))),
Effect.forkScoped,
)
const closed = Effect.gen(function* () {
const exitCode = yield* tunnel.exitCode
yield* Fiber.join(stderr)
return yield* Effect.fail(
new SshFailure("connection", (yield* Ref.get(detail)) || JSON.stringify({ exitCode })),
)
})
yield* waitReady(http, () => items.get(config.id)?.stage === "authentication").pipe(
Effect.provideService(HttpClient.HttpClient, httpClient),
Effect.catch(() =>
@@ -188,20 +220,26 @@ export const createSshController = Effect.fn("Ssh.controller")(function* (input:
const start = Effect.fn("Ssh.start")(function* (request: SshStart, owner?: number) {
const id = request.id
if (lifecycle.closed || !/^[a-zA-Z0-9-]{1,80}$/.test(id)) return
const previous = attempts.get(id)
// A second window must not replace an interactive attempt while its owner
// is connecting or answering a challenge.
if (previous?.owner !== undefined && previous.owner !== owner && items.get(id)?.stage !== "ready") return
const config = { id, target: request.target.trim(), name: request.name.trim() }
const connection: Connection = {
owner,
before: items.get(id),
ready: yield* Deferred.make<SshHttp | null>(),
}
const admitted = yield* Deferred.make<void>()
const fiber = yield* Effect.gen(function* () {
yield* Deferred.await(admitted)
if (previous) yield* Fiber.interrupt(previous.fiber)
yield* connect(config, connection, request.replace)
}).pipe(
@@ -210,29 +248,34 @@ export const createSshController = Effect.fn("Ssh.controller")(function* (input:
if (Cause.hasInterruptsOnly(cause) || paused.has(id) || attempts.get(id)?.ready !== connection.ready) return
const failure = SshFailure.from(Cause.squash(cause))
failures.set(id, (failures.get(id) ?? 0) + 1)
const code = /REMOTE HOST IDENTIFICATION HAS CHANGED|Host key verification failed/.test(failure.message)
? "host-key"
: /spawn .*ENOENT/.test(failure.message)
? "ssh-missing"
: failure.code
if (
["version", "input", "unpublished", "platform", "host-key", "ssh-missing"].includes(code) ||
/Permission denied/.test(failure.message) ||
(failures.get(id) ?? 0) >= 5
)
paused.add(id)
yield* update(id, {
stage: code === "version" ? "incompatible" : "failed",
error: code,
detail: failure.message,
prompt: undefined,
...(configs.has(id) ? { config: configs.get(id) } : {}),
})
yield* update(
id,
withSavedConfig(id, {
stage: code === "version" ? "incompatible" : "failed",
error: code,
detail: failure.message,
prompt: undefined,
}),
)
}),
),
Effect.ensuring(
Effect.gen(function* () {
yield* Deferred.succeed(connection.ready, null)
if (attempts.get(id)?.ready !== connection.ready) return
attempts.delete(id)
yield* update(id, { prompt: undefined })
@@ -240,6 +283,7 @@ export const createSshController = Effect.fn("Ssh.controller")(function* (input:
),
Effect.forkIn(lifetime, { uninterruptible: false }),
)
attempts.set(id, Object.assign(connection, { fiber }))
items.set(id, {
config,
@@ -249,6 +293,7 @@ export const createSshController = Effect.fn("Ssh.controller")(function* (input:
detail: "",
})
paused.delete(id)
if (!request.background) failures.delete(id)
yield* emit
yield* Deferred.succeed(admitted, undefined)
@@ -257,15 +302,14 @@ export const createSshController = Effect.fn("Ssh.controller")(function* (input:
const disconnect = Effect.fn("Ssh.disconnect")(function* (id: string) {
paused.add(id)
const attempt = attempts.get(id)
yield* update(id, {
stage: "disconnected",
prompt: undefined,
...(configs.has(id) ? { config: configs.get(id) } : {}),
})
yield* update(id, withSavedConfig(id, { stage: "disconnected", prompt: undefined }))
if (attempt) yield* Fiber.interrupt(attempt.fiber)
})
const cancel = Effect.fn("Ssh.cancel")(function* (id: string, owner: number) {
const attempt = attempts.get(id)
if (!attempt || attempt.owner !== owner) return
paused.add(id)
// Restore before interrupting: askpass cleanup must not transition the
@@ -283,12 +327,14 @@ export const createSshController = Effect.fn("Ssh.controller")(function* (input:
})
yield* Fiber.interrupt(attempt.fiber)
})
const close = Effect.gen(function* () {
if (lifecycle.closed) return
lifecycle.closed = true
yield* Effect.forEach([...attempts.keys()], disconnect, { concurrency: "unbounded", discard: true })
yield* Scope.close(lifetime, Exit.void)
})
yield* Effect.addFinalizer(() => close)
return {
@@ -297,22 +343,29 @@ export const createSshController = Effect.fn("Ssh.controller")(function* (input:
start,
resolve: Effect.fn("Ssh.resolve")(function* (id: string) {
const item = items.get(id)
if (lifecycle.closed || !item || paused.has(id)) return null
if (item.stage === "ready" && item.http) {
const healthy = yield* checkHealth(item.http).pipe(Effect.provideService(HttpClient.HttpClient, httpClient))
if (lifecycle.closed || paused.has(id)) return null
// Another window may already have replaced this tunnel during the probe.
if (items.get(id) === item) {
if (healthy) return item.http
yield* start({ ...item.config, background: true })
}
}
if (!attempts.has(id) && items.has(id)) yield* start({ ...item.config, background: true })
const attempt = attempts.get(id)
return attempt ? yield* Deferred.await(attempt.ready) : null
}),
respond: Effect.fn("Ssh.respond")(function* (id: string, prompt: string, value: string, owner: number) {
const attempt = attempts.get(id)
if (attempt?.owner === owner && attempt.respond) yield* attempt.respond(prompt, value)
}),
disconnect,
@@ -337,7 +390,9 @@ export const createSshController = Effect.fn("Ssh.controller")(function* (input:
const freePort = Effect.gen(function* () {
const server = yield* NodeSocketServer.make({ host: "127.0.0.1", port: 0 })
if (server.address._tag !== "TcpAddress") return yield* Effect.fail(new SshFailure("connection"))
if (!Predicate.isTagged(server.address, "TcpAddress")) return yield* Effect.fail(new SshFailure("connection"))
return server.address.port
}).pipe(Effect.scoped)
@@ -345,14 +400,18 @@ const waitReady = Effect.fn("Ssh.waitReady")(function* (http: SshHttp, authentic
const clock = { deadline: (yield* Clock.currentTimeMillis) + 30_000 }
yield* Effect.gen(function* () {
const now = yield* Clock.currentTimeMillis
if (authenticating()) clock.deadline = now + 30_000
if (now >= clock.deadline) return yield* Effect.fail(new SshFailure("service"))
return yield* checkHealth(http)
}).pipe(Effect.repeat({ until: (ready) => ready, schedule: Schedule.spaced(100) }))
})
const checkHealth = Effect.fn("Ssh.checkHealth")(function* (http: SshHttp) {
const client = yield* HttpClient.HttpClient
return yield* client
.get(`${http.url}/api/info`, {
headers: { authorization: `Basic ${Buffer.from(`opencode:${http.password}`).toString("base64")}` },
@@ -19,6 +19,7 @@ export function quote(value: string) {
export function requireVersion(version: string) {
if (version !== "local" && !/^[0-9][a-zA-Z0-9.+-]*$/.test(version))
throw new Failure({ code: "version", detail: version })
return version
}
@@ -48,10 +49,13 @@ export function parseVersion(output: string) {
.split(/\r?\n/)
.find((line) => line.trim())
?.trim()
if (!line) return null
const marker = line.lastIndexOf(" v")
const version = marker === -1 ? line : line.slice(marker + 2)
if (!version) throw new Error("V2 CLI did not provide a version")
return version
}
@@ -71,6 +75,7 @@ printf 'OPENCODE_REMOTE_TARGET=%s\\n' "$target"
export function archiveUrl(target: string, version: string) {
if (!/^(linux|darwin)-(x64-baseline|arm64)(-musl)?$/.test(target))
throw new Failure({ code: "platform", detail: target })
return `https://registry.npmjs.org/@opencode/cli-${target}/-/cli-${target}-${requireVersion(version)}.tgz`
}
@@ -78,6 +83,7 @@ type Source = { type: "download"; url: string } | { type: "archive" } | { type:
export function installScript(input: { version: string; directory?: string; source: Source }) {
const version = requireVersion(input.version)
// The managed CLI installer also configures the user's shell PATH. Private
// installations use archives so their destination and shell setup stay isolated.
if (input.source.type === "installer")
@@ -85,6 +91,7 @@ export function installScript(input: { version: string; directory?: string; sour
curl -fsSL https://raw.githubusercontent.com/anomalyco/opencode/v2/install | bash -s -- ${input.source.binary ? `--binary ${input.source.binary}` : `--version ${quote(version)}`}
${verifyScript('"$HOME/.opencode/bin/opencode"', version)}
`
return `set -eu
umask 077
destination="$HOME"/${quote(`${input.directory ?? ".opencode/bin"}/opencode`)}
@@ -100,6 +107,7 @@ mv "$stage/package/bin/opencode" "$destination"
function stageBinary(source: Exclude<Source, { type: "installer" }>) {
if (source.type === "archive") return 'cat > "$stage/archive.tgz"\ntar -xzf "$stage/archive.tgz" -C "$stage"'
return `url=${quote(source.url)}
if command -v curl >/dev/null 2>&1; then
curl -fsSL --connect-timeout 15 --max-time 180 "$url" -o "$stage/archive.tgz"
@@ -117,11 +125,13 @@ const Beta = Schema.Struct({ version: Schema.String.check(Schema.isPattern(/^0\.
export const latestBeta = Effect.fn("RemoteCli.latestBeta")(function* () {
const http = yield* HttpClient.HttpClient
const metadata = yield* http.get("https://registry.npmjs.org/@opencode%2fcli/beta").pipe(
Effect.flatMap(HttpClientResponse.filterStatusOk),
Effect.flatMap(HttpClientResponse.schemaBodyJson(Beta)),
Effect.timeout("30 seconds"),
Effect.mapError(() => new Failure({ code: "install", detail: "https://registry.npmjs.org/@opencode%2fcli/beta" })),
)
return metadata.version
})