Skip to content
Closed
Show file tree
Hide file tree
Changes from 3 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
112 changes: 112 additions & 0 deletions apps/server/src/provider/Layers/OpenCodeAdapter.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@ import * as NodeAssert from "node:assert/strict";
import * as NodeServices from "@effect/platform-node/NodeServices";
import { it } from "@effect/vitest";
import * as Context from "effect/Context";
import * as Deferred from "effect/Deferred";
import * as Effect from "effect/Effect";
import * as Exit from "effect/Exit";
import * as Fiber from "effect/Fiber";
Expand All @@ -19,6 +20,7 @@ import {
OpenCodeSettings,
ProviderDriverKind,
ProviderInstanceId,
type ProviderRuntimeEvent,
ThreadId,
} from "@t3tools/contracts";
import { createModelSelection } from "@t3tools/shared/model";
Expand Down Expand Up @@ -628,6 +630,116 @@ it.layer(OpenCodeAdapterTestLayer)("OpenCodeAdapterLive", (it) => {
}),
);

it.effect("cancels pending permission requests when stopping a session", () =>
Effect.gen(function* () {
const adapter = yield* OpenCodeAdapter;
const threadId = asThreadId("thread-opencode-stop-pending-permission");
runtimeMock.state.subscribedEvents = [
{
type: "permission.asked",
properties: {
id: "perm-stop-1",
sessionID: "http://127.0.0.1:9999/session",
permission: "bash",
patterns: ["rm *"],
metadata: {},
},
},
];
const opened = yield* Deferred.make<ProviderRuntimeEvent>();
const resolved = yield* Deferred.make<ProviderRuntimeEvent>();
const eventsFiber = yield* adapter.streamEvents.pipe(
Stream.runForEach((event) => {
if (event.threadId !== threadId) {
return Effect.void;
}
if (event.type === "request.opened") {
return Deferred.succeed(opened, event).pipe(Effect.asVoid, Effect.ignore);
}
if (event.type === "request.resolved") {
return Deferred.succeed(resolved, event).pipe(Effect.asVoid, Effect.ignore);
}
return Effect.void;
}),
Effect.forkChild,
);

yield* adapter.startSession({
provider: ProviderDriverKind.make("opencode"),
threadId,
runtimeMode: "full-access",
});
const openedEvent = yield* Deferred.await(opened).pipe(Effect.timeout("1 second"));
yield* adapter.stopSession(threadId);
const resolvedEvent = yield* Deferred.await(resolved).pipe(Effect.timeout("1 second"));

NodeAssert.equal(openedEvent.type, "request.opened");
if (openedEvent.type === "request.opened") {
NodeAssert.equal(openedEvent.requestId, "perm-stop-1");
NodeAssert.equal(openedEvent.payload.requestType, "command_execution_approval");
}
NodeAssert.equal(resolvedEvent.type, "request.resolved");
if (resolvedEvent.type === "request.resolved") {
NodeAssert.equal(resolvedEvent.requestId, "perm-stop-1");
NodeAssert.equal(resolvedEvent.payload.decision, "cancel");
}
yield* Fiber.interrupt(eventsFiber);
}),
);

it.effect("cancels pending permission requests on stopAll", () =>
Effect.gen(function* () {
const adapter = yield* OpenCodeAdapter;
const threadId = asThreadId("thread-opencode-stop-all-pending-permission");
runtimeMock.state.subscribedEvents = [
{
type: "permission.asked",
properties: {
id: "perm-stop-all-1",
sessionID: "http://127.0.0.1:9999/session",
permission: "edit",
patterns: [],
metadata: {},
},
},
];
const opened = yield* Deferred.make<ProviderRuntimeEvent>();
const resolved = yield* Deferred.make<ProviderRuntimeEvent>();
const eventsFiber = yield* adapter.streamEvents.pipe(
Stream.runForEach((event) => {
if (event.threadId !== threadId) {
return Effect.void;
}
if (event.type === "request.opened") {
return Deferred.succeed(opened, event).pipe(Effect.asVoid, Effect.ignore);
}
if (event.type === "request.resolved") {
return Deferred.succeed(resolved, event).pipe(Effect.asVoid, Effect.ignore);
}
return Effect.void;
}),
Effect.forkChild,
);

yield* adapter.startSession({
provider: ProviderDriverKind.make("opencode"),
threadId,
runtimeMode: "full-access",
});
yield* Deferred.await(opened).pipe(Effect.timeout("1 second"));
yield* adapter.stopAll();
const resolvedEvent = yield* Deferred.await(resolved).pipe(Effect.timeout("1 second"));

NodeAssert.equal(resolvedEvent.type, "request.resolved");
if (resolvedEvent.type === "request.resolved") {
NodeAssert.equal(resolvedEvent.requestId, "perm-stop-all-1");
NodeAssert.equal(resolvedEvent.payload.requestType, "file_change_approval");
NodeAssert.equal(resolvedEvent.payload.decision, "cancel");
}
yield* Fiber.interrupt(eventsFiber);
}),
);

it.effect("clears session state even when cleanup finalizers throw", () =>
Effect.gen(function* () {
const adapter = yield* OpenCodeAdapter;
Expand Down
109 changes: 98 additions & 11 deletions apps/server/src/provider/Layers/OpenCodeAdapter.ts
Original file line number Diff line number Diff line change
Expand Up @@ -229,6 +229,8 @@ interface OpenCodeSessionContext {
readonly openCodeSessionId: string;
readonly pendingPermissions: Map<string, PermissionRequest>;
readonly pendingQuestions: Map<string, QuestionRequest>;
/** False once teardown starts so the event pump cannot open new requests. */
readonly acceptingRequests: Ref.Ref<boolean>;
readonly messageRoleById: Map<string, "user" | "assistant">;
readonly partById: Map<string, Part>;
readonly emittedTextByPartId: Map<string, string>;
Expand Down Expand Up @@ -638,12 +640,57 @@ export function makeOpenCodeAdapter(
})),
);

// Layer-level finalizer: when the adapter layer shuts down, stop every
// session. Each session's `Scope.close` tears down its spawned OpenCode
// server (via the `ChildProcessSpawner` finalizer installed in
// `startOpenCodeServerProcess`) and interrupts the forked event/exit
// fibers. Consumers that can't reason about Effect scopes therefore
// cannot leak OpenCode child processes by forgetting to call `stopAll`.
const emit = (event: ProviderRuntimeEvent) =>
Queue.offer(runtimeEvents, event).pipe(Effect.asVoid);

Comment thread
cursor[bot] marked this conversation as resolved.
const settleOpenCodePendingAsCancelled = (context: OpenCodeSessionContext) =>
Effect.gen(function* () {
yield* Ref.set(context.acceptingRequests, false);
const permissions = [...context.pendingPermissions.entries()];
Comment thread
macroscopeapp[bot] marked this conversation as resolved.
Outdated
yield* Effect.forEach(
permissions,
([requestId, request]) =>
Effect.gen(function* () {
yield* emit({
...(yield* buildEventBase({
threadId: context.session.threadId,
requestId,
})),
type: "request.resolved",
payload: {
requestType: mapPermissionToRequestType(request.permission),
decision: "cancel",
},
});
context.pendingPermissions.delete(requestId);
}),
{ discard: true },
);
const questions = [...context.pendingQuestions.keys()];
yield* Effect.forEach(
questions,
(requestId) =>
Effect.gen(function* () {
yield* emit({
...(yield* buildEventBase({
threadId: context.session.threadId,
requestId,
})),
type: "user-input.resolved",
payload: { answers: {} },
});
context.pendingQuestions.delete(requestId);
}),
{ discard: true },
);
});

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This changes backend behavior (pending permissions/questions are now settled as cancel / empty answers on stop, resume-restart, unexpected exit, and stopAll, and late permission.asked / question.asked events are auto-resolved once acceptingRequests is false), but no tests accompany it. Consider adding focused cases to OpenCodeAdapter.test.ts — the existing it.layer(OpenCodeAdapterTestLayer) harness can push a permission.asked event and then assert the emitted request.resolved on stop/stopAll — so the new settle path is pinned.

Posted via Macroscope — Effect Service Conventions

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Added focused OpenCodeAdapter tests in cead9c1: a permission.asked event is opened, then stopSession / stopAll emit request.resolved with decision: "cancel".

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Sorry, I'm unable to act on this request because you do not have permissions within this repository.


// Layer-level finalizer: when the adapter layer shuts down, settle every
// pending request then stop the session. Each session's `Scope.close`
// tears down its spawned OpenCode server (via the `ChildProcessSpawner`
// finalizer installed in `startOpenCodeServerProcess`) and interrupts
// the forked event/exit fibers. Provider instance remove/replace closes
// this scope without calling `stopAll`, so settlement has to live here.
yield* Effect.addFinalizer(() =>
Effect.gen(function* () {
const contexts = [...sessions.values()];
Expand All @@ -653,7 +700,12 @@ export function makeOpenCodeAdapter(
// the remaining cleanups.
yield* Effect.forEach(
contexts,
(context) => Effect.ignoreCause(stopOpenCodeContext(context)),
(context) =>
Effect.ignoreCause(
settleOpenCodePendingAsCancelled(context).pipe(
Effect.andThen(stopOpenCodeContext(context)),
),

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Settle failure skips session teardown

Medium Severity

settleOpenCodePendingAsCancelled is chained with Effect.andThen(stopOpenCodeContext) under Effect.ignoreCause in stopAll and the layer finalizer. If settlement fails after flipping acceptingRequests, teardown never runs and the error is swallowed, so the OpenCode session scope and child process can leak. emitUnexpectedExit already ignores settle failures and continues teardown; replace/stopSession share the same settle-then-stop short-circuit.

Additional Locations (2)
Fix in Cursor Fix in Web

Reviewed by Cursor Bugbot for commit debfa76. Configure here.

),
{ concurrency: "unbounded", discard: true },
);
// Close the logger AFTER session teardown so any final lifecycle
Expand All @@ -665,9 +717,6 @@ export function makeOpenCodeAdapter(
}
}).pipe(Effect.ensuring(Queue.shutdown(runtimeEvents))),
);

const emit = (event: ProviderRuntimeEvent) =>
Queue.offer(runtimeEvents, event).pipe(Effect.asVoid);
const writeNativeEvent = (
threadId: ThreadId,
event: {
Expand Down Expand Up @@ -696,6 +745,7 @@ export function makeOpenCodeAdapter(
}
const turnId = context.activeTurnId;
sessions.delete(context.session.threadId);
yield* settleOpenCodePendingAsCancelled(context).pipe(Effect.ignore);
// Emit lifecycle events BEFORE tearing down the scope. Both call sites
// run this inside a fiber forked via `Effect.forkIn(context.sessionScope)`;
// closing that scope triggers the fiber-interrupt finalizer, so any
Expand Down Expand Up @@ -956,6 +1006,22 @@ export function makeOpenCodeAdapter(
}

case "permission.asked": {
if (!(yield* Ref.get(context.acceptingRequests))) {
yield* emit({
...(yield* buildEventBase({
threadId: context.session.threadId,
turnId,
requestId: event.properties.id,
raw: event,
})),
type: "request.resolved",
payload: {
requestType: mapPermissionToRequestType(event.properties.permission),
decision: "cancel",
},
});
break;
}
Comment thread
cursor[bot] marked this conversation as resolved.
Outdated
context.pendingPermissions.set(event.properties.id, event.properties);
yield* emit({
...(yield* buildEventBase({
Expand Down Expand Up @@ -996,6 +1062,19 @@ export function makeOpenCodeAdapter(
}

case "question.asked": {
if (!(yield* Ref.get(context.acceptingRequests))) {
yield* emit({
...(yield* buildEventBase({
threadId: context.session.threadId,
turnId,
requestId: event.properties.id,
raw: event,
})),
type: "user-input.resolved",
payload: { answers: {} },
});
break;
}
context.pendingQuestions.set(event.properties.id, event.properties);
yield* emit({
...(yield* buildEventBase({
Expand Down Expand Up @@ -1210,6 +1289,7 @@ export function makeOpenCodeAdapter(
const resumeSessionId = parseOpenCodeResume(input.resumeCursor)?.sessionId;
const existing = sessions.get(input.threadId);
if (existing) {
yield* settleOpenCodePendingAsCancelled(existing);
yield* stopOpenCodeContext(existing);
sessions.delete(input.threadId);
}
Expand Down Expand Up @@ -1394,6 +1474,7 @@ export function makeOpenCodeAdapter(
openCodeSessionId: started.openCodeSession.id,
pendingPermissions: new Map(),
pendingQuestions: new Map(),
acceptingRequests: yield* Ref.make(true),
partById: new Map(),
emittedTextByPartId: new Map(),
messageRoleById: new Map(),
Expand Down Expand Up @@ -1626,6 +1707,7 @@ export function makeOpenCodeAdapter(
threadId,
});
}
yield* settleOpenCodePendingAsCancelled(context);
Comment thread
cursor[bot] marked this conversation as resolved.
const stopped = yield* stopOpenCodeContext(context);
sessions.delete(threadId);
if (!stopped) {
Expand Down Expand Up @@ -1710,7 +1792,12 @@ export function makeOpenCodeAdapter(
// interrupt the sibling fibers. Same pattern as the layer finalizer.
yield* Effect.forEach(
contexts,
(context) => Effect.ignoreCause(stopOpenCodeContext(context)),
(context) =>
Effect.ignoreCause(
settleOpenCodePendingAsCancelled(context).pipe(
Effect.andThen(stopOpenCodeContext(context)),
),
),
{ concurrency: "unbounded", discard: true },
);
});
Expand Down
Loading