diff --git a/packages/opencode/src/question/index.ts b/packages/opencode/src/question/index.ts index e03af848b070..4ded2aec253a 100644 --- a/packages/opencode/src/question/index.ts +++ b/packages/opencode/src/question/index.ts @@ -1,4 +1,4 @@ -import { Deferred, Effect, Layer, Schema, Context } from "effect" +import { Cache, Context, Deferred, Duration, Effect, Layer, Schema } from "effect" import { Bus } from "@/bus" import { BusEvent } from "@/bus/bus-event" import { InstanceState } from "@/effect/instance-state" @@ -75,6 +75,19 @@ export const Reply = Schema.Struct({ }).annotate({ identifier: "QuestionReply" }) export type Reply = Schema.Schema.Type +export const Result = Schema.Union([ + Schema.Struct({ + status: Schema.Literal("answered"), + answers: Schema.Array(Answer).annotate({ + description: "User answers in order of questions (each answer is an array of selected labels)", + }), + }), + Schema.Struct({ + status: Schema.Literal("rejected"), + }), +]).annotate({ discriminator: "status", identifier: "QuestionResult" }) +export type Result = Schema.Schema.Type + const Replied = Schema.Struct({ sessionID: SessionID, requestID: QuestionID, @@ -104,21 +117,28 @@ export class NotFoundError extends Schema.TaggedErrorClass()("Que interface PendingEntry { info: Request - deferred: Deferred.Deferred, RejectedError> + deferred: Deferred.Deferred } interface State { pending: Map + completed: Cache.Cache } // Service export interface Interface { + readonly create: (input: { + sessionID: SessionID + questions: ReadonlyArray + tool?: Tool + }) => Effect.Effect readonly ask: (input: { sessionID: SessionID questions: ReadonlyArray tool?: Tool }) => Effect.Effect, RejectedError> + readonly wait: (requestID: QuestionID) => Effect.Effect readonly reply: (input: { requestID: QuestionID answers: ReadonlyArray @@ -137,12 +157,17 @@ export const layer = Layer.effect( Effect.fn("Question.state")(function* () { const state = { pending: new Map(), + completed: yield* Cache.make({ + capacity: 10_000, + timeToLive: Duration.minutes(30), + lookup: (requestID) => Effect.fail(new NotFoundError({ requestID })), + }), } yield* Effect.addFinalizer(() => Effect.gen(function* () { for (const item of state.pending.values()) { - yield* Deferred.fail(item.deferred, new RejectedError()) + yield* Deferred.succeed(item.deferred, { status: "rejected" }) } state.pending.clear() }), @@ -152,7 +177,7 @@ export const layer = Layer.effect( }), ) - const ask = Effect.fn("Question.ask")(function* (input: { + const create = Effect.fn("Question.create")(function* (input: { sessionID: SessionID questions: ReadonlyArray tool?: Tool @@ -161,7 +186,7 @@ export const layer = Layer.effect( const id = QuestionID.ascending() log.info("asking", { id, questions: input.questions.length }) - const deferred = yield* Deferred.make, RejectedError>() + const deferred = yield* Deferred.make() const info: Request = { id, sessionID: input.sessionID, @@ -170,49 +195,79 @@ export const layer = Layer.effect( } pending.set(id, { info, deferred }) yield* bus.publish(Event.Asked, info) + return info + }) + + const wait = Effect.fn("Question.wait")(function* (requestID: QuestionID) { + const info = yield* InstanceState.get(state) + const pending = info.pending.get(requestID) + if (pending) return yield* Deferred.await(pending.deferred) + return yield* Cache.get(info.completed, requestID).pipe( + Effect.tapError(() => Cache.invalidateWhen(info.completed, requestID, () => true)), + ) + }) + + const settle = Effect.fn("Question.settle")(function* (requestID: QuestionID, result: Result) { + const info = yield* InstanceState.get(state) + const existing = info.pending.get(requestID) + if (!existing) return false + + yield* Cache.set(info.completed, requestID, result) + info.pending.delete(requestID) + if (result.status === "answered") { + yield* bus.publish(Event.Replied, { + sessionID: existing.info.sessionID, + requestID: existing.info.id, + answers: result.answers, + }) + } else { + yield* bus.publish(Event.Rejected, { + sessionID: existing.info.sessionID, + requestID: existing.info.id, + }) + } + yield* Deferred.succeed(existing.deferred, result) + return true + }) - return yield* Effect.ensuring( - Deferred.await(deferred), - Effect.sync(() => { - pending.delete(id) - }), + const ask = Effect.fn("Question.ask")(function* (input: { + sessionID: SessionID + questions: ReadonlyArray + tool?: Tool + }) { + const request = yield* create(input) + + const result = yield* Effect.ensuring( + wait(request.id).pipe(Effect.orDie), + settle(request.id, { status: "rejected" }).pipe(Effect.asVoid), ) + if (result.status === "answered") return result.answers + return yield* new RejectedError() }) const reply = Effect.fn("Question.reply")(function* (input: { requestID: QuestionID answers: ReadonlyArray }) { - const pending = (yield* InstanceState.get(state)).pending - const existing = pending.get(input.requestID) - if (!existing) { + const result: Result = { + status: "answered", + answers: input.answers.map((a) => [...a]), + } + const ok = yield* settle(input.requestID, result) + if (!ok) { log.warn("reply for unknown request", { requestID: input.requestID }) return yield* new NotFoundError({ requestID: input.requestID }) } - pending.delete(input.requestID) log.info("replied", { requestID: input.requestID, answers: input.answers }) - yield* bus.publish(Event.Replied, { - sessionID: existing.info.sessionID, - requestID: existing.info.id, - answers: input.answers.map((a) => [...a]), - }) - yield* Deferred.succeed(existing.deferred, input.answers) }) const reject = Effect.fn("Question.reject")(function* (requestID: QuestionID) { - const pending = (yield* InstanceState.get(state)).pending - const existing = pending.get(requestID) - if (!existing) { + const ok = yield* settle(requestID, { status: "rejected" }) + if (!ok) { log.warn("reject for unknown request", { requestID }) return yield* new NotFoundError({ requestID }) } - pending.delete(requestID) log.info("rejected", { requestID }) - yield* bus.publish(Event.Rejected, { - sessionID: existing.info.sessionID, - requestID: existing.info.id, - }) - yield* Deferred.fail(existing.deferred, new RejectedError()) }) const list = Effect.fn("Question.list")(function* () { @@ -220,7 +275,7 @@ export const layer = Layer.effect( return Array.from(pending.values(), (x) => x.info) }) - return Service.of({ ask, reply, reject, list }) + return Service.of({ create, ask, wait, reply, reject, list }) }), ) diff --git a/packages/opencode/src/server/routes/instance/httpapi/groups/question.ts b/packages/opencode/src/server/routes/instance/httpapi/groups/question.ts index e9e63429db20..5004939dcba3 100644 --- a/packages/opencode/src/server/routes/instance/httpapi/groups/question.ts +++ b/packages/opencode/src/server/routes/instance/httpapi/groups/question.ts @@ -1,5 +1,6 @@ import { Question } from "@/question" import { QuestionID } from "@/question/schema" +import { SessionID } from "@/session/schema" import { Schema } from "effect" import { HttpApi, HttpApiEndpoint, HttpApiError, HttpApiGroup, OpenApi } from "effect/unstable/httpapi" import { QuestionNotFoundError } from "../errors" @@ -15,10 +16,32 @@ const ReplyPayload = Schema.Struct({ }), }) +export const AskPayload = Schema.Struct({ + sessionID: SessionID, + questions: Schema.Array(Question.Info).annotate({ + description: "Questions to ask", + }), +}) + +const AskResponse = Schema.Struct({ + id: QuestionID, +}) + export const QuestionApi = HttpApi.make("question") .add( HttpApiGroup.make("question") .add( + HttpApiEndpoint.post("ask", root, { + query: WorkspaceRoutingQuery, + payload: AskPayload, + success: described(AskResponse, "Created question request"), + }).annotateMerge( + OpenApi.annotations({ + identifier: "question.ask", + summary: "Ask questions", + description: "Create a question request and return its id.", + }), + ), HttpApiEndpoint.get("list", root, { query: WorkspaceRoutingQuery, success: described(Schema.Array(Question.Request), "List of pending questions"), @@ -29,6 +52,18 @@ export const QuestionApi = HttpApi.make("question") description: "Get all pending question requests across all sessions.", }), ), + HttpApiEndpoint.get("wait", `${root}/:requestID/wait`, { + params: { requestID: QuestionID }, + query: WorkspaceRoutingQuery, + success: described(Question.Result, "Question result"), + error: [QuestionNotFoundError], + }).annotateMerge( + OpenApi.annotations({ + identifier: "question.wait", + summary: "Wait for question result", + description: "Wait for a question request to be answered or rejected.", + }), + ), HttpApiEndpoint.post("reply", `${root}/:requestID/reply`, { params: { requestID: QuestionID }, query: WorkspaceRoutingQuery, diff --git a/packages/opencode/src/server/routes/instance/httpapi/handlers/question.ts b/packages/opencode/src/server/routes/instance/httpapi/handlers/question.ts index e9ed8cc89154..c5d9c147a0c8 100644 --- a/packages/opencode/src/server/routes/instance/httpapi/handlers/question.ts +++ b/packages/opencode/src/server/routes/instance/httpapi/handlers/question.ts @@ -4,6 +4,7 @@ import { Effect } from "effect" import { HttpApiBuilder } from "effect/unstable/httpapi" import { InstanceHttpApi } from "../api" import { QuestionNotFoundError } from "../errors" +import { AskPayload } from "../groups/question" export const questionHandlers = HttpApiBuilder.group(InstanceHttpApi, "question", (handlers) => Effect.gen(function* () { @@ -13,6 +14,24 @@ export const questionHandlers = HttpApiBuilder.group(InstanceHttpApi, "question" return yield* svc.list() }) + const ask = Effect.fn("QuestionHttpApi.ask")(function* (ctx: { payload: typeof AskPayload.Type }) { + const request = yield* svc.create(ctx.payload) + return { id: request.id } + }) + + const wait = Effect.fn("QuestionHttpApi.wait")(function* (ctx: { params: { requestID: QuestionID } }) { + return yield* svc.wait(ctx.params.requestID).pipe( + Effect.catchTag("Question.NotFoundError", (error) => + Effect.fail( + new QuestionNotFoundError({ + requestID: String(error.requestID), + message: `Question request not found: ${error.requestID}`, + }), + ), + ), + ) + }) + const reply = Effect.fn("QuestionHttpApi.reply")(function* (ctx: { params: { requestID: QuestionID } payload: Question.Reply @@ -49,6 +68,11 @@ export const questionHandlers = HttpApiBuilder.group(InstanceHttpApi, "question" return true }) - return handlers.handle("list", list).handle("reply", reply).handle("reject", reject) + return handlers + .handle("ask", ask) + .handle("list", list) + .handle("wait", wait) + .handle("reply", reply) + .handle("reject", reject) }), ) diff --git a/packages/opencode/test/question/question.test.ts b/packages/opencode/test/question/question.test.ts index 5f6f87972ea4..eac3d81670b2 100644 --- a/packages/opencode/test/question/question.test.ts +++ b/packages/opencode/test/question/question.test.ts @@ -23,6 +23,20 @@ const askEffect = Effect.fn("QuestionTest.ask")(function* (input: { return yield* question.ask(input) }) +const createEffect = Effect.fn("QuestionTest.create")(function* (input: { + sessionID: SessionID + questions: ReadonlyArray + tool?: Question.Tool +}) { + const question = yield* Question.Service + return yield* question.create(input) +}) + +const waitEffect = Effect.fn("QuestionTest.wait")(function* (requestID: QuestionID) { + const question = yield* Question.Service + return yield* question.wait(requestID) +}) + const listEffect = Question.Service.use((svc) => svc.list()) const replyEffect = Effect.fn("QuestionTest.reply")(function* (input: { @@ -183,6 +197,64 @@ it.instance( { git: true }, ) +it.instance( + "wait - returns answers for pending and completed requests", + () => + Effect.gen(function* () { + const request = yield* createEffect({ + sessionID: SessionID.make("ses_test"), + questions: [ + { + question: "What would you like to do?", + header: "Action", + options: [ + { label: "Option 1", description: "First option" }, + { label: "Option 2", description: "Second option" }, + ], + }, + ], + }) + + const fiber = yield* waitEffect(request.id).pipe(Effect.forkScoped) + + yield* replyEffect({ + requestID: request.id, + answers: [["Option 1"]], + }) + + const result = { status: "answered" as const, answers: [["Option 1"]] } + expect(yield* Fiber.join(fiber)).toEqual(result) + expect(yield* waitEffect(request.id)).toEqual(result) + }), + { git: true }, +) + +it.instance( + "wait - returns rejected when pending ask is interrupted", + () => + Effect.gen(function* () { + const fiber = yield* askEffect({ + sessionID: SessionID.make("ses_test"), + questions: [ + { + question: "What would you like to do?", + header: "Action", + options: [{ label: "Option 1", description: "First option" }], + }, + ], + }).pipe(Effect.forkScoped) + + const pending = yield* waitForPending(1) + const waiter = yield* waitEffect(pending[0].id).pipe(Effect.forkScoped) + + yield* Fiber.interrupt(fiber) + + expect(yield* Fiber.join(waiter)).toEqual({ status: "rejected" }) + expect(yield* waitEffect(pending[0].id)).toEqual({ status: "rejected" }) + }), + { git: true }, +) + it.instance( "reply - fails for unknown requestID", () => @@ -259,6 +331,28 @@ it.instance( { git: true }, ) +it.instance( + "wait - returns rejected for completed rejected requests", + () => + Effect.gen(function* () { + const request = yield* createEffect({ + sessionID: SessionID.make("ses_test"), + questions: [ + { + question: "What would you like to do?", + header: "Action", + options: [{ label: "Option 1", description: "First option" }], + }, + ], + }) + + yield* rejectEffect(request.id) + + expect(yield* waitEffect(request.id)).toEqual({ status: "rejected" }) + }), + { git: true }, +) + it.instance( "reject - fails for unknown requestID", () => diff --git a/packages/opencode/test/server/httpapi-exercise/index.ts b/packages/opencode/test/server/httpapi-exercise/index.ts index f2b132cb7320..c3e188d9d924 100644 --- a/packages/opencode/test/server/httpapi-exercise/index.ts +++ b/packages/opencode/test/server/httpapi-exercise/index.ts @@ -235,6 +235,34 @@ const scenarios: Scenario[] = [ })) .json(404, object, "status"), http.protected.get("/question", "question.list").json(200, array), + http.protected + .post("/question", "question.ask") + .mutating() + .at((ctx) => ({ + path: "/question", + headers: ctx.headers(), + body: { + sessionID: "ses_httpapi_question", + questions: [ + { + question: "Proceed?", + header: "Proceed", + options: [{ label: "Yes", description: "Continue" }], + }, + ], + }, + })) + .json(200, (body) => { + object(body) + check(typeof body.id === "string" && body.id.startsWith("que_"), "question ask should return request id") + }), + http.protected + .get("/question/{requestID}/wait", "question.wait.missing") + .at((ctx) => ({ + path: route("/question/{requestID}/wait", { requestID: "que_httpapi_wait" }), + headers: ctx.headers(), + })) + .json(404, object, "status"), http.protected .post("/question/{requestID}/reply", "question.reply.invalid") .at((ctx) => ({ diff --git a/packages/sdk/js/src/v2/gen/sdk.gen.ts b/packages/sdk/js/src/v2/gen/sdk.gen.ts index cd17e70fdf0e..22fa8c89afed 100644 --- a/packages/sdk/js/src/v2/gen/sdk.gen.ts +++ b/packages/sdk/js/src/v2/gen/sdk.gen.ts @@ -151,12 +151,17 @@ import type { PtyUpdateErrors, PtyUpdateResponses, QuestionAnswer, + QuestionAskErrors, + QuestionAskResponses, + QuestionInfo, QuestionListErrors, QuestionListResponses, QuestionRejectErrors, QuestionRejectResponses, QuestionReplyErrors, QuestionReplyResponses, + QuestionWaitErrors, + QuestionWaitResponses, SessionAbortErrors, SessionAbortResponses, SessionChildrenErrors, @@ -2687,6 +2692,77 @@ export class Question extends HeyApiClient { }) } + /** + * Ask questions + * + * Create a question request and return its id. + */ + public ask( + parameters?: { + directory?: string + workspace?: string + sessionID?: string + questions?: Array + }, + options?: Options, + ) { + const params = buildClientParams( + [parameters], + [ + { + args: [ + { in: "query", key: "directory" }, + { in: "query", key: "workspace" }, + { in: "body", key: "sessionID" }, + { in: "body", key: "questions" }, + ], + }, + ], + ) + return (options?.client ?? this.client).post({ + url: "/question", + ...options, + ...params, + headers: { + "Content-Type": "application/json", + ...options?.headers, + ...params.headers, + }, + }) + } + + /** + * Wait for question result + * + * Wait for a question request to be answered or rejected. + */ + public wait( + parameters: { + requestID: string + directory?: string + workspace?: string + }, + options?: Options, + ) { + const params = buildClientParams( + [parameters], + [ + { + args: [ + { in: "path", key: "requestID" }, + { in: "query", key: "directory" }, + { in: "query", key: "workspace" }, + ], + }, + ], + ) + return (options?.client ?? this.client).get({ + url: "/question/{requestID}/wait", + ...options, + ...params, + }) + } + /** * Reply to question request * diff --git a/packages/sdk/js/src/v2/gen/types.gen.ts b/packages/sdk/js/src/v2/gen/types.gen.ts index aae1b06ad320..e5e366b70599 100644 --- a/packages/sdk/js/src/v2/gen/types.gen.ts +++ b/packages/sdk/js/src/v2/gen/types.gen.ts @@ -1717,6 +1717,18 @@ export type PtyForbiddenError = { message: string } +export type QuestionResult = + | { + status: "answered" + /** + * User answers in order of questions (each answer is an array of selected labels) + */ + answers: Array + } + | { + status: "rejected" + } + export type QuestionNotFoundError = { _tag: "QuestionNotFoundError" requestID: string @@ -5743,6 +5755,76 @@ export type QuestionListResponses = { export type QuestionListResponse = QuestionListResponses[keyof QuestionListResponses] +export type QuestionAskData = { + body?: { + sessionID: string + /** + * Questions to ask + */ + questions: Array + } + path?: never + query?: { + directory?: string + workspace?: string + } + url: "/question" +} + +export type QuestionAskErrors = { + /** + * Bad request + */ + 400: BadRequestError +} + +export type QuestionAskError = QuestionAskErrors[keyof QuestionAskErrors] + +export type QuestionAskResponses = { + /** + * Created question request + */ + 200: { + id: string + } +} + +export type QuestionAskResponse = QuestionAskResponses[keyof QuestionAskResponses] + +export type QuestionWaitData = { + body?: never + path: { + requestID: string + } + query?: { + directory?: string + workspace?: string + } + url: "/question/{requestID}/wait" +} + +export type QuestionWaitErrors = { + /** + * Bad request + */ + 400: BadRequestError + /** + * QuestionNotFoundError + */ + 404: QuestionNotFoundError +} + +export type QuestionWaitError = QuestionWaitErrors[keyof QuestionWaitErrors] + +export type QuestionWaitResponses = { + /** + * Question result + */ + 200: QuestionResult +} + +export type QuestionWaitResponse = QuestionWaitResponses[keyof QuestionWaitResponses] + export type QuestionReplyData = { body?: { /**