Compare commits

...
6 changed files with 378 additions and 214 deletions
+4 -4
View File
@@ -1,8 +1,8 @@
{
"nodeModules": {
"x86_64-linux": "sha256-qAkjcbc1nJqOnCrNQ0bnsM4WG2ii5K1JWS9ohAYdjus=",
"aarch64-linux": "sha256-Nb+F0e3CvQv+uLzHzj9JKp5hV78mCnlSqFXzgIvgR24=",
"aarch64-darwin": "sha256-BIXALWWrjEZLUKZrY6l6+scjZmKFscFxX26TvWOvXGQ=",
"x86_64-darwin": "sha256-3uaFXl/n6je7AzIfsY1pvt3Ln/U1Oshx3z7ohuVPEs8="
"x86_64-linux": "sha256-FI1mX42vJuYdUDdWevlfHz+OcYkDn/I/HUbHE/jdQvs=",
"aarch64-linux": "sha256-3CQzzKnh/4Zf5vyn56yR5P3ULsW7K7Fr8/RQpekEJDk=",
"aarch64-darwin": "sha256-XPDVHMxlPpXlf43BRqNnwF809unk6iE8tvd0o92d0/w=",
"x86_64-darwin": "sha256-dFXTi13RSgL62lMsep1EoE/KSEPF7Oh31PVdxW1tkzg="
}
}
+47 -19
View File
@@ -118,6 +118,11 @@ interface TokenResponse {
expires_in?: number
}
interface CodexAuthPluginOptions {
issuer?: string
codexApiEndpoint?: string
}
async function exchangeCodeForTokens(code: string, redirectUri: string, pkce: PkceCodes): Promise<TokenResponse> {
const response = await fetch(`${ISSUER}/oauth/token`, {
method: "POST",
@@ -136,8 +141,8 @@ async function exchangeCodeForTokens(code: string, redirectUri: string, pkce: Pk
return response.json()
}
async function refreshAccessToken(refreshToken: string): Promise<TokenResponse> {
const response = await fetch(`${ISSUER}/oauth/token`, {
async function refreshAccessToken(refreshToken: string, issuer = ISSUER): Promise<TokenResponse> {
const response = await fetch(`${issuer}/oauth/token`, {
method: "POST",
headers: { "Content-Type": "application/x-www-form-urlencoded" },
body: new URLSearchParams({
@@ -364,7 +369,10 @@ function waitForOAuthCallback(pkce: PkceCodes, state: string): Promise<TokenResp
})
}
export async function CodexAuthPlugin(input: PluginInput): Promise<Hooks> {
export async function CodexAuthPlugin(input: PluginInput, options: CodexAuthPluginOptions = {}): Promise<Hooks> {
const issuer = options.issuer ?? ISSUER
const codexApiEndpoint = options.codexApiEndpoint ?? CODEX_API_ENDPOINT
return {
provider: {
id: "openai",
@@ -405,6 +413,13 @@ export async function CodexAuthPlugin(input: PluginInput): Promise<Hooks> {
const auth = await getAuth()
if (auth.type !== "oauth") return {}
let refreshPromise:
| Promise<{
access: string
accountId: string | undefined
}>
| undefined
return {
apiKey: OAUTH_DUMMY_KEY,
async fetch(requestInput: RequestInfo | URL, init?: RequestInit) {
@@ -429,21 +444,34 @@ export async function CodexAuthPlugin(input: PluginInput): Promise<Hooks> {
// Check if token needs refresh
if (!currentAuth.access || currentAuth.expires < Date.now()) {
log.info("refreshing codex access token")
const tokens = await refreshAccessToken(currentAuth.refresh)
const newAccountId = extractAccountId(tokens) || authWithAccount.accountId
await input.client.auth.set({
path: { id: "openai" },
body: {
type: "oauth",
refresh: tokens.refresh_token,
access: tokens.access_token,
expires: Date.now() + (tokens.expires_in ?? 3600) * 1000,
...(newAccountId && { accountId: newAccountId }),
},
})
currentAuth.access = tokens.access_token
authWithAccount.accountId = newAccountId
if (!refreshPromise) {
log.info("refreshing codex access token")
refreshPromise = refreshAccessToken(currentAuth.refresh, issuer)
.then(async (tokens) => {
const accountId = extractAccountId(tokens) || authWithAccount.accountId
await input.client.auth.set({
path: { id: "openai" },
body: {
type: "oauth",
refresh: tokens.refresh_token,
access: tokens.access_token,
expires: Date.now() + (tokens.expires_in ?? 3600) * 1000,
...(accountId && { accountId }),
},
})
return {
access: tokens.access_token,
accountId,
}
})
.finally(() => {
refreshPromise = undefined
})
}
const refreshed = await refreshPromise
currentAuth.access = refreshed.access
authWithAccount.accountId = refreshed.accountId
}
// Build headers
@@ -477,7 +505,7 @@ export async function CodexAuthPlugin(input: PluginInput): Promise<Hooks> {
: new URL(typeof requestInput === "string" ? requestInput : requestInput.url)
const url =
parsed.pathname.includes("/v1/responses") || parsed.pathname.includes("/chat/completions")
? new URL(CODEX_API_ENDPOINT)
? new URL(codexApiEndpoint)
: parsed
return fetch(url, {
+12 -190
View File
@@ -8,17 +8,15 @@ import * as Session from "./session"
import { Agent } from "../agent/agent"
import { Provider } from "@/provider/provider"
import { ModelID, ProviderID } from "../provider/schema"
import { type Tool as AITool, tool, jsonSchema, type ToolExecutionOptions, asSchema } from "ai"
import { type Tool as AITool, tool, jsonSchema } from "ai"
import type { JSONSchema7 } from "@ai-sdk/provider"
import { SessionCompaction } from "./compaction"
import { Bus } from "../bus"
import { ProviderTransform } from "@/provider/transform"
import { SystemPrompt } from "./system"
import { Instruction } from "./instruction"
import { Plugin } from "../plugin"
import MAX_STEPS from "../session/prompt/max-steps.txt"
import { ToolRegistry } from "@/tool/registry"
import { ToolJsonSchema } from "@/tool/json-schema"
import { MCP } from "../mcp"
import { LSP } from "@/lsp/lsp"
import { ulid } from "ulid"
@@ -48,7 +46,6 @@ import * as EffectLogger from "@opencode-ai/core/effect/logger"
import { InstanceState } from "@/effect/instance-state"
import { TaskTool, type TaskPromptOps } from "@/tool/task"
import { SessionRunState } from "./run-state"
import { EffectBridge } from "@/effect/bridge"
import { RuntimeFlags } from "@/effect/runtime-flags"
import { EventV2 } from "@opencode-ai/core/event"
import { EventV2Bridge } from "@/event-v2-bridge"
@@ -63,6 +60,7 @@ import * as Database from "@/storage/db"
import { SessionTable } from "./session.sql"
import { referencePromptMetadata, referenceTextPart } from "./prompt/reference"
import { SessionReminders } from "./reminders"
import { SessionTools } from "./tools"
import { LLMEvent } from "@opencode-ai/llm"
// @ts-ignore
@@ -126,9 +124,6 @@ export const layer = Layer.effect(
const references = yield* Reference.Service
const events = yield* EventV2Bridge.Service
const flags = yield* RuntimeFlags.Service
const runner = Effect.fn("SessionPrompt.runner")(function* () {
return yield* EffectBridge.make()
})
const ops = Effect.fn("SessionPrompt.ops")(function* () {
return {
cancel: (sessionID: SessionID) => cancel(sessionID),
@@ -301,186 +296,6 @@ export const layer = Layer.effect(
.pipe(Effect.catchCause((cause) => elog.error("failed to generate title", { error: Cause.squash(cause) })))
})
const resolveTools = Effect.fn("SessionPrompt.resolveTools")(function* (input: {
agent: Agent.Info
model: Provider.Model
session: Session.Info
tools?: Record<string, boolean>
processor: Pick<SessionProcessor.Handle, "message" | "updateToolCall" | "completeToolCall">
bypassAgentCheck: boolean
messages: MessageV2.WithParts[]
}) {
using _ = log.time("resolveTools")
const tools: Record<string, AITool> = {}
const run = yield* runner()
const promptOps = yield* ops()
const context = (args: any, options: ToolExecutionOptions): Tool.Context => ({
sessionID: input.session.id,
abort: options.abortSignal!,
messageID: input.processor.message.id,
callID: options.toolCallId,
extra: { model: input.model, bypassAgentCheck: input.bypassAgentCheck, promptOps },
agent: input.agent.name,
messages: input.messages,
metadata: (val) =>
input.processor.updateToolCall(options.toolCallId, (match) => {
if (!["running", "pending"].includes(match.state.status)) return match
return {
...match,
state: {
title: val.title,
metadata: val.metadata,
status: "running",
input: args,
time: { start: Date.now() },
},
}
}),
ask: (req) =>
permission
.ask({
...req,
sessionID: input.session.id,
tool: { messageID: input.processor.message.id, callID: options.toolCallId },
ruleset: Permission.merge(input.agent.permission, input.session.permission ?? []),
})
.pipe(Effect.orDie),
})
for (const item of yield* registry.tools({
modelID: ModelID.make(input.model.api.id),
providerID: input.model.providerID,
agent: input.agent,
})) {
const schema = ProviderTransform.schema(input.model, ToolJsonSchema.fromTool(item))
tools[item.id] = tool({
description: item.description,
inputSchema: jsonSchema(schema),
execute(args, options) {
return run.promise(
Effect.gen(function* () {
const ctx = context(args, options)
yield* plugin.trigger(
"tool.execute.before",
{ tool: item.id, sessionID: ctx.sessionID, callID: ctx.callID },
{ args },
)
const result = yield* item.execute(args, ctx)
const output = {
...result,
attachments: result.attachments?.map((attachment) => ({
...attachment,
id: PartID.ascending(),
sessionID: ctx.sessionID,
messageID: input.processor.message.id,
})),
}
yield* plugin.trigger(
"tool.execute.after",
{ tool: item.id, sessionID: ctx.sessionID, callID: ctx.callID, args },
output,
)
if (options.abortSignal?.aborted) {
yield* input.processor.completeToolCall(options.toolCallId, output)
}
return output
}),
)
},
})
}
for (const [key, item] of Object.entries(yield* mcp.tools())) {
const execute = item.execute
if (!execute) continue
const schema = yield* Effect.promise(() => Promise.resolve(asSchema(item.inputSchema).jsonSchema))
const transformed = ProviderTransform.schema(input.model, schema)
item.inputSchema = jsonSchema(transformed)
item.execute = (args, opts) =>
run.promise(
Effect.gen(function* () {
const ctx = context(args, opts)
yield* plugin.trigger(
"tool.execute.before",
{ tool: key, sessionID: ctx.sessionID, callID: opts.toolCallId },
{ args },
)
const result: Awaited<ReturnType<NonNullable<typeof execute>>> = yield* Effect.gen(function* () {
yield* ctx.ask({ permission: key, metadata: {}, patterns: ["*"], always: ["*"] })
return yield* Effect.promise(() => execute(args, opts))
}).pipe(
Effect.withSpan("Tool.execute", {
attributes: {
"tool.name": key,
"tool.call_id": opts.toolCallId,
"session.id": ctx.sessionID,
"message.id": input.processor.message.id,
},
}),
)
yield* plugin.trigger(
"tool.execute.after",
{ tool: key, sessionID: ctx.sessionID, callID: opts.toolCallId, args },
result,
)
const textParts: string[] = []
const attachments: Omit<MessageV2.FilePart, "id" | "sessionID" | "messageID">[] = []
for (const contentItem of result.content) {
if (contentItem.type === "text") textParts.push(contentItem.text)
else if (contentItem.type === "image") {
attachments.push({
type: "file",
mime: contentItem.mimeType,
url: `data:${contentItem.mimeType};base64,${contentItem.data}`,
})
} else if (contentItem.type === "resource") {
const { resource } = contentItem
if (resource.text) textParts.push(resource.text)
if (resource.blob) {
attachments.push({
type: "file",
mime: resource.mimeType ?? "application/octet-stream",
url: `data:${resource.mimeType ?? "application/octet-stream"};base64,${resource.blob}`,
filename: resource.uri,
})
}
}
}
const truncated = yield* truncate.output(textParts.join("\n\n"), {}, input.agent)
const metadata = {
...result.metadata,
truncated: truncated.truncated,
...(truncated.truncated && { outputPath: truncated.outputPath }),
}
const output = {
title: "",
metadata,
output: truncated.content,
attachments: attachments.map((attachment) => ({
...attachment,
id: PartID.ascending(),
sessionID: ctx.sessionID,
messageID: input.processor.message.id,
})),
content: result.content,
}
if (opts.abortSignal?.aborted) {
yield* input.processor.completeToolCall(opts.toolCallId, output)
}
return output
}),
)
tools[key] = item
}
return tools
})
const handleSubtask = Effect.fn("SessionPrompt.handleSubtask")(function* (input: {
task: MessageV2.SubtaskPart
model: Provider.Model
@@ -1552,16 +1367,23 @@ export const layer = Layer.effect(
const outcome: "break" | "continue" = yield* Effect.gen(function* () {
const lastUserMsg = msgs.findLast((m) => m.info.role === "user")
const bypassAgentCheck = lastUserMsg?.parts.some((p) => p.type === "agent") ?? false
const promptOps = yield* ops()
const tools = yield* resolveTools({
const tools = yield* SessionTools.resolve({
agent,
session,
model,
tools: lastUser.tools,
processor: handle,
bypassAgentCheck,
messages: msgs,
})
promptOps,
}).pipe(
Effect.provideService(Plugin.Service, plugin),
Effect.provideService(Permission.Service, permission),
Effect.provideService(ToolRegistry.Service, registry),
Effect.provideService(MCP.Service, mcp),
Effect.provideService(Truncate.Service, truncate),
)
if (lastUser.format?.type === "json_schema") {
tools["StructuredOutput"] = createStructuredOutputTool({
+208
View File
@@ -0,0 +1,208 @@
import { Agent } from "@/agent/agent"
import { Provider } from "@/provider/provider"
import { ProviderTransform } from "@/provider/transform"
import { MCP } from "@/mcp"
import { Permission } from "@/permission"
import { Tool } from "@/tool/tool"
import { ToolJsonSchema } from "@/tool/json-schema"
import { ToolRegistry } from "@/tool/registry"
import { Truncate } from "@/tool/truncate"
import { ModelID } from "@/provider/schema"
import { Plugin } from "@/plugin"
import type { TaskPromptOps } from "@/tool/task"
import { type Tool as AITool, tool, jsonSchema, type ToolExecutionOptions, asSchema } from "ai"
import { Effect } from "effect"
import { MessageV2 } from "./message-v2"
import * as Session from "./session"
import { SessionProcessor } from "./processor"
import { PartID } from "./schema"
import * as Log from "@opencode-ai/core/util/log"
import { EffectBridge } from "@/effect/bridge"
const log = Log.create({ service: "session.tools" })
export const resolve = Effect.fn("SessionTools.resolve")(function* (input: {
agent: Agent.Info
model: Provider.Model
session: Session.Info
processor: Pick<SessionProcessor.Handle, "message" | "updateToolCall" | "completeToolCall">
bypassAgentCheck: boolean
messages: MessageV2.WithParts[]
promptOps: TaskPromptOps
}) {
using _ = log.time("resolveTools")
const tools: Record<string, AITool> = {}
const run = yield* EffectBridge.make()
const plugin = yield* Plugin.Service
const permission = yield* Permission.Service
const registry = yield* ToolRegistry.Service
const mcp = yield* MCP.Service
const truncate = yield* Truncate.Service
const context = (args: Record<string, unknown>, options: ToolExecutionOptions): Tool.Context => ({
sessionID: input.session.id,
abort: options.abortSignal!,
messageID: input.processor.message.id,
callID: options.toolCallId,
extra: { model: input.model, bypassAgentCheck: input.bypassAgentCheck, promptOps: input.promptOps },
agent: input.agent.name,
messages: input.messages,
metadata: (val) =>
input.processor.updateToolCall(options.toolCallId, (match) => {
if (!["running", "pending"].includes(match.state.status)) return match
return {
...match,
state: {
title: val.title,
metadata: val.metadata,
status: "running",
input: args,
time: { start: Date.now() },
},
}
}),
ask: (req) =>
permission
.ask({
...req,
sessionID: input.session.id,
tool: { messageID: input.processor.message.id, callID: options.toolCallId },
ruleset: Permission.merge(input.agent.permission, input.session.permission ?? []),
})
.pipe(Effect.orDie),
})
for (const item of yield* registry.tools({
modelID: ModelID.make(input.model.api.id),
providerID: input.model.providerID,
agent: input.agent,
})) {
const schema = ProviderTransform.schema(input.model, ToolJsonSchema.fromTool(item))
tools[item.id] = tool({
description: item.description,
inputSchema: jsonSchema(schema),
execute(args, options) {
return run.promise(
Effect.gen(function* () {
const ctx = context(args, options)
yield* plugin.trigger(
"tool.execute.before",
{ tool: item.id, sessionID: ctx.sessionID, callID: ctx.callID },
{ args },
)
const result = yield* item.execute(args, ctx)
const output = {
...result,
attachments: result.attachments?.map((attachment) => ({
...attachment,
id: PartID.ascending(),
sessionID: ctx.sessionID,
messageID: input.processor.message.id,
})),
}
yield* plugin.trigger(
"tool.execute.after",
{ tool: item.id, sessionID: ctx.sessionID, callID: ctx.callID, args },
output,
)
if (options.abortSignal?.aborted) {
yield* input.processor.completeToolCall(options.toolCallId, output)
}
return output
}),
)
},
})
}
for (const [key, item] of Object.entries(yield* mcp.tools())) {
const execute = item.execute
if (!execute) continue
const schema = yield* Effect.promise(() => Promise.resolve(asSchema(item.inputSchema).jsonSchema))
const transformed = ProviderTransform.schema(input.model, schema)
item.inputSchema = jsonSchema(transformed)
item.execute = (args, opts) =>
run.promise(
Effect.gen(function* () {
const ctx = context(args, opts)
yield* plugin.trigger(
"tool.execute.before",
{ tool: key, sessionID: ctx.sessionID, callID: opts.toolCallId },
{ args },
)
const result: Awaited<ReturnType<NonNullable<typeof execute>>> = yield* Effect.gen(function* () {
yield* ctx.ask({ permission: key, metadata: {}, patterns: ["*"], always: ["*"] })
return yield* Effect.promise(() => execute(args, opts))
}).pipe(
Effect.withSpan("Tool.execute", {
attributes: {
"tool.name": key,
"tool.call_id": opts.toolCallId,
"session.id": ctx.sessionID,
"message.id": input.processor.message.id,
},
}),
)
yield* plugin.trigger(
"tool.execute.after",
{ tool: key, sessionID: ctx.sessionID, callID: opts.toolCallId, args },
result,
)
const textParts: string[] = []
const attachments: Omit<MessageV2.FilePart, "id" | "sessionID" | "messageID">[] = []
for (const contentItem of result.content) {
if (contentItem.type === "text") textParts.push(contentItem.text)
else if (contentItem.type === "image") {
attachments.push({
type: "file",
mime: contentItem.mimeType,
url: `data:${contentItem.mimeType};base64,${contentItem.data}`,
})
} else if (contentItem.type === "resource") {
const { resource } = contentItem
if (resource.text) textParts.push(resource.text)
if (resource.blob) {
attachments.push({
type: "file",
mime: resource.mimeType ?? "application/octet-stream",
url: `data:${resource.mimeType ?? "application/octet-stream"};base64,${resource.blob}`,
filename: resource.uri,
})
}
}
}
const truncated = yield* truncate.output(textParts.join("\n\n"), {}, input.agent)
const metadata = {
...result.metadata,
truncated: truncated.truncated,
...(truncated.truncated && { outputPath: truncated.outputPath }),
}
const output = {
title: "",
metadata,
output: truncated.content,
attachments: attachments.map((attachment) => ({
...attachment,
id: PartID.ascending(),
sessionID: ctx.sessionID,
messageID: input.processor.message.id,
})),
content: result.content,
}
if (opts.abortSignal?.aborted) {
yield* input.processor.completeToolCall(opts.toolCallId, output)
}
return output
}),
)
tools[key] = item
}
return tools
})
export * as SessionTools from "./tools"
+106
View File
@@ -1,5 +1,6 @@
import { describe, expect, test } from "bun:test"
import {
CodexAuthPlugin,
parseJwtClaims,
extractAccountIdFromClaims,
extractAccountId,
@@ -120,4 +121,109 @@ describe("plugin.codex", () => {
).toBe("acc-123")
})
})
test("deduplicates concurrent Codex token refreshes", async () => {
let auth = {
type: "oauth" as const,
refresh: "refresh-old",
access: "",
expires: 0,
}
const authUpdates: Array<{
body: { refresh: string; access: string; expires: number; accountId?: string }
}> = []
let resolveRefresh: (() => void) | undefined
const refreshReady = new Promise<void>((resolve) => {
resolveRefresh = resolve
})
let refreshRequests = 0
const apiRequests: { authorization: string | null; accountId: string | null }[] = []
using server = Bun.serve({
port: 0,
async fetch(request) {
const url = new URL(request.url)
if (url.pathname === "/oauth/token") {
expect(await request.text()).toContain("refresh_token=refresh-old")
refreshRequests += 1
await refreshReady
return Response.json({
id_token: createTestJwt({ chatgpt_account_id: "acc-123" }),
access_token: "access-new",
refresh_token: "refresh-new",
expires_in: 3600,
})
}
if (url.pathname === "/backend-api/codex/responses") {
apiRequests.push({
authorization: request.headers.get("authorization"),
accountId: request.headers.get("ChatGPT-Account-Id"),
})
return new Response("{}", { status: 200 })
}
return new Response("unexpected request", { status: 500 })
},
})
const hooks = await CodexAuthPlugin(
{
client: {
auth: {
async set(input: { body: { refresh: string; access: string; expires: number; accountId?: string } }) {
authUpdates.push(input)
auth = {
type: "oauth",
refresh: input.body.refresh,
access: input.body.access,
expires: input.body.expires,
...(input.body.accountId && { accountId: input.body.accountId }),
}
},
},
} as never,
project: {} as never,
directory: "",
worktree: "",
experimental_workspace: {
register() {},
},
serverUrl: new URL("https://example.com"),
$: {} as never,
},
{
issuer: server.url.origin,
codexApiEndpoint: new URL("/backend-api/codex/responses", server.url).toString(),
},
)
const loaded = await hooks.auth!.loader!(async () => auth as never, {} as never)
const first = loaded.fetch!("https://api.openai.com/v1/responses")
const second = loaded.fetch!("https://api.openai.com/v1/responses")
await waitFor(() => refreshRequests === 1)
expect(apiRequests).toHaveLength(0)
resolveRefresh!()
await Promise.all([first, second])
expect(refreshRequests).toBe(1)
expect(authUpdates).toHaveLength(1)
expect(authUpdates[0]?.body.refresh).toBe("refresh-new")
expect(authUpdates[0]?.body.access).toBe("access-new")
expect(authUpdates[0]?.body.accountId).toBe("acc-123")
expect(apiRequests).toEqual([
{ authorization: "Bearer access-new", accountId: "acc-123" },
{ authorization: "Bearer access-new", accountId: "acc-123" },
])
})
})
async function waitFor(predicate: () => boolean) {
const started = Date.now()
while (!predicate()) {
if (Date.now() - started > 1_000) throw new Error("timed out waiting for condition")
await new Promise((resolve) => setTimeout(resolve, 1))
}
}
+1 -1
View File
@@ -1563,7 +1563,7 @@ PART_MAPPING["reasoning"] = function ReasoningPartDisplay(props) {
const streaming = createMemo(
() => props.message.role === "assistant" && typeof (props.message as AssistantMessage).time.completed !== "number",
)
const text = () => (data.store.part_text_accum_delta?.[part().id] ?? part().text).trim()
const text = () => (data.store.part_text_accum_delta?.[part().id] ?? part().text ?? "").trim()
return (
<Show when={text()}>