Skip to content
Merged
128 changes: 122 additions & 6 deletions apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -71,16 +71,18 @@ describe("ProviderCommandReactor", () => {
async function createHarness() {
const now = new Date().toISOString();
const runtimeEventPubSub = Effect.runSync(PubSub.unbounded<ProviderRuntimeEvent>());
const startSession = vi.fn((_: unknown, __: unknown) =>
Effect.succeed({
sessionId: asSessionId("sess-1"),
let nextSessionIndex = 1;
const startSession = vi.fn((_: unknown, __: unknown) => {
const sessionIndex = nextSessionIndex++;
return Effect.succeed({
sessionId: asSessionId(`sess-${sessionIndex}`),
provider: "codex" as const,
status: "ready" as const,
threadId: ProviderThreadId.makeUnsafe("provider-thread-1"),
threadId: ProviderThreadId.makeUnsafe(`provider-thread-${sessionIndex}`),
createdAt: now,
updatedAt: now,
}),
);
});
});
const sendTurn = vi.fn((_: unknown) =>
Effect.succeed({
threadId: ProviderThreadId.makeUnsafe("provider-thread-1"),
Expand Down Expand Up @@ -171,6 +173,8 @@ describe("ProviderCommandReactor", () => {
text: "hello reactor",
attachments: [],
},
approvalPolicy: "on-request",
sandboxMode: "workspace-write",
createdAt: now,
}),
);
Expand All @@ -181,11 +185,123 @@ describe("ProviderCommandReactor", () => {
expect(harness.startSession.mock.calls[0]?.[1]).toMatchObject({
cwd: "/tmp/provider-project",
model: "gpt-5-codex",
approvalPolicy: "on-request",
sandboxMode: "workspace-write",
});

const readModel = await Effect.runPromise(harness.engine.getReadModel());
const thread = readModel.threads.find((entry) => entry.id === ThreadId.makeUnsafe("thread-1"));
expect(thread?.session?.providerSessionId).toBe("sess-1");
expect(thread?.session?.approvalPolicy).toBe("on-request");
expect(thread?.session?.sandboxMode).toBe("workspace-write");
});

it("reuses the same provider session when runtime mode is unchanged", async () => {
const harness = await createHarness();
const now = new Date().toISOString();

await Effect.runPromise(
harness.engine.dispatch({
type: "thread.turn.start",
commandId: CommandId.makeUnsafe("cmd-turn-start-unchanged-1"),
threadId: ThreadId.makeUnsafe("thread-1"),
message: {
messageId: asMessageId("user-message-unchanged-1"),
role: "user",
text: "first",
attachments: [],
},
approvalPolicy: "on-request",
sandboxMode: "workspace-write",
createdAt: now,
}),
);

await waitFor(() => harness.startSession.mock.calls.length === 1);
await waitFor(() => harness.sendTurn.mock.calls.length === 1);

await Effect.runPromise(
harness.engine.dispatch({
type: "thread.turn.start",
commandId: CommandId.makeUnsafe("cmd-turn-start-unchanged-2"),
threadId: ThreadId.makeUnsafe("thread-1"),
message: {
messageId: asMessageId("user-message-unchanged-2"),
role: "user",
text: "second",
attachments: [],
},
approvalPolicy: "on-request",
sandboxMode: "workspace-write",
createdAt: now,
}),
);

await waitFor(() => harness.sendTurn.mock.calls.length === 2);
expect(harness.startSession.mock.calls.length).toBe(1);
expect(harness.stopSession.mock.calls.length).toBe(0);
});

it("restarts the provider session when runtime mode changes", async () => {
const harness = await createHarness();
const now = new Date().toISOString();

await Effect.runPromise(
harness.engine.dispatch({
type: "thread.turn.start",
commandId: CommandId.makeUnsafe("cmd-turn-start-runtime-mode-1"),
threadId: ThreadId.makeUnsafe("thread-1"),
message: {
messageId: asMessageId("user-message-runtime-mode-1"),
role: "user",
text: "first",
attachments: [],
},
approvalPolicy: "never",
sandboxMode: "danger-full-access",
createdAt: now,
}),
);

await waitFor(() => harness.startSession.mock.calls.length === 1);
await waitFor(() => harness.sendTurn.mock.calls.length === 1);

await Effect.runPromise(
harness.engine.dispatch({
type: "thread.turn.start",
commandId: CommandId.makeUnsafe("cmd-turn-start-runtime-mode-2"),
threadId: ThreadId.makeUnsafe("thread-1"),
message: {
messageId: asMessageId("user-message-runtime-mode-2"),
role: "user",
text: "second",
attachments: [],
},
approvalPolicy: "on-request",
sandboxMode: "workspace-write",
createdAt: now,
}),
);

await waitFor(() => harness.stopSession.mock.calls.length === 1);
await waitFor(() => harness.startSession.mock.calls.length === 2);
await waitFor(() => harness.sendTurn.mock.calls.length === 2);

expect(harness.stopSession.mock.calls[0]?.[0]).toEqual({ sessionId: asSessionId("sess-1") });
expect(harness.startSession.mock.calls[1]?.[1]).toMatchObject({
resumeThreadId: ProviderThreadId.makeUnsafe("provider-thread-1"),
approvalPolicy: "on-request",
sandboxMode: "workspace-write",
});
expect(harness.sendTurn.mock.calls[1]?.[0]).toMatchObject({
sessionId: asSessionId("sess-2"),
});

const readModel = await Effect.runPromise(harness.engine.getReadModel());
const thread = readModel.threads.find((entry) => entry.id === ThreadId.makeUnsafe("thread-1"));
expect(thread?.session?.providerSessionId).toBe("sess-2");
expect(thread?.session?.approvalPolicy).toBe("on-request");
expect(thread?.session?.sandboxMode).toBe("workspace-write");
});

it("reacts to thread.turn.interrupt-requested by calling provider interrupt", async () => {
Expand Down
90 changes: 65 additions & 25 deletions apps/server/src/orchestration/Layers/ProviderCommandReactor.ts
Original file line number Diff line number Diff line change
Expand Up @@ -3,9 +3,13 @@ import {
CommandId,
EventId,
type OrchestrationEvent,
type ProviderApprovalPolicy,
type ProviderKind,
type ProviderSandboxMode,
type OrchestrationSession,
type ThreadId,
type ProviderSession,
type ProviderThreadId,
type TurnId,
} from "@t3tools/contracts";
import { Cache, Cause, Duration, Effect, Layer, Option, Queue, Stream } from "effect";
Expand Down Expand Up @@ -128,18 +132,19 @@ const make = Effect.gen(function* () {
const ensureSessionForThread = Effect.fnUntraced(function* (
threadId: ThreadId,
createdAt: string,
options?: {
readonly approvalPolicy?: ProviderApprovalPolicy;
readonly sandboxMode?: ProviderSandboxMode;
},
Comment thread
cursor[bot] marked this conversation as resolved.
) {
const readModel = yield* orchestrationEngine.getReadModel();
const thread = readModel.threads.find((entry) => entry.id === threadId);
if (!thread) {
return yield* Effect.die(new Error(`Thread '${threadId}' was not found in read model.`));
}

const existingSessionId = thread.session?.providerSessionId;
if (existingSessionId) {
return existingSessionId;
}

const desiredApprovalPolicy = options?.approvalPolicy ?? thread.session?.approvalPolicy;
const desiredSandboxMode = options?.sandboxMode ?? thread.session?.sandboxMode;

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.

鈿狅笍 Potential issue | 馃煛 Minor

Fallback policy/mode continuity breaks after explicit session stops.

Line 146 and Line 147 fall back to values on thread.session, but the stop path writes a stopped session without approvalPolicy/sandboxMode, so those fallbacks become unavailable after a stop.

馃挕 Suggested fix
     yield* setThreadSession({
       threadId: thread.id,
       session: {
         threadId: thread.id,
         status: "stopped",
         providerName: thread.session?.providerName ?? null,
         providerSessionId: null,
         providerThreadId: null,
+        ...(thread.session?.approvalPolicy !== undefined
+          ? { approvalPolicy: thread.session.approvalPolicy }
+          : {}),
+        ...(thread.session?.sandboxMode !== undefined
+          ? { sandboxMode: thread.session.sandboxMode }
+          : {}),
         activeTurnId: null,
         lastError: thread.session?.lastError ?? null,
         updatedAt: now,
       },
       createdAt: now,
     });
馃 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.

In `@apps/server/src/orchestration/Layers/ProviderCommandReactor.ts` around lines
146 - 147, desiredApprovalPolicy and desiredSandboxMode currently fallback to
thread.session values, but after an explicit stop those properties are missing;
update the fallback logic so after options you check for
preserved/session-history values before using thread.session directly (e.g.,
options?.approvalPolicy ?? thread.session?.approvalPolicy ??
thread.previousSession?.approvalPolicy ?? defaultApprovalPolicy and similarly
for sandboxMode), and if no such history field exists add and use a stable store
on Thread (e.g., previousSession or sessionSnapshot set before stop) so
ProviderCommandReactor's desiredApprovalPolicy and desiredSandboxMode always
resolve to the last known policy/mode.

const preferredProvider: ProviderKind | undefined =
thread.session?.providerName === "codex" || thread.session?.providerName === "claudeCode"
? thread.session.providerName
Expand All @@ -149,28 +154,54 @@ const make = Effect.gen(function* () {
projects: readModel.projects,
});

const startedSession = yield* providerService.startSession(threadId, {
...(preferredProvider ? { provider: preferredProvider } : {}),
...(effectiveCwd ? { cwd: effectiveCwd } : {}),
...(thread.model ? { model: thread.model } : {}),
});
const startProviderSession = (resumeThreadId?: ProviderThreadId | null) =>
providerService.startSession(threadId, {
...(preferredProvider ? { provider: preferredProvider } : {}),
...(effectiveCwd ? { cwd: effectiveCwd } : {}),
...(thread.model ? { model: thread.model } : {}),
...(resumeThreadId ? { resumeThreadId } : {}),
...(desiredApprovalPolicy !== undefined ? { approvalPolicy: desiredApprovalPolicy } : {}),
...(desiredSandboxMode !== undefined ? { sandboxMode: desiredSandboxMode } : {}),
});

yield* setThreadSession({
threadId,
session: {
const bindSessionToThread = (session: ProviderSession) =>
setThreadSession({
threadId,
status: mapProviderSessionStatusToOrchestrationStatus(startedSession.status),
providerName: startedSession.provider,
providerSessionId: startedSession.sessionId,
providerThreadId: startedSession.threadId ?? null,
// Provider turn ids are not orchestration turn ids.
activeTurnId: null,
lastError: startedSession.lastError ?? null,
updatedAt: startedSession.updatedAt,
},
createdAt,
});
session: {
threadId,
status: mapProviderSessionStatusToOrchestrationStatus(session.status),
providerName: session.provider,
providerSessionId: session.sessionId,
providerThreadId: session.threadId ?? null,
...(desiredApprovalPolicy !== undefined ? { approvalPolicy: desiredApprovalPolicy } : {}),
...(desiredSandboxMode !== undefined ? { sandboxMode: desiredSandboxMode } : {}),
// Provider turn ids are not orchestration turn ids.
activeTurnId: null,
lastError: session.lastError ?? null,
updatedAt: session.updatedAt,
},
createdAt,
});

const existingSessionId = thread.session?.providerSessionId;
if (existingSessionId) {
const approvalPolicyChanged =
options?.approvalPolicy !== undefined && options.approvalPolicy !== thread.session?.approvalPolicy;
const sandboxModeChanged =
options?.sandboxMode !== undefined && options.sandboxMode !== thread.session?.sandboxMode;

if (!approvalPolicyChanged && !sandboxModeChanged) {
return existingSessionId;
}

yield* providerService.stopSession({ sessionId: existingSessionId });
Comment thread
macroscopeapp[bot] marked this conversation as resolved.
Outdated
const restartedSession = yield* startProviderSession(thread.session?.providerThreadId ?? null);
yield* bindSessionToThread(restartedSession);
return restartedSession.sessionId;
Comment thread
cursor[bot] marked this conversation as resolved.
}
Comment thread
juliusmarminge marked this conversation as resolved.
Outdated

const startedSession = yield* startProviderSession();
yield* bindSessionToThread(startedSession);
return startedSession.sessionId;
});

Expand All @@ -180,13 +211,18 @@ const make = Effect.gen(function* () {
readonly attachments?: ReadonlyArray<ChatAttachment>;
readonly model?: string;
readonly effort?: string;
readonly approvalPolicy?: ProviderApprovalPolicy;
readonly sandboxMode?: ProviderSandboxMode;
readonly createdAt: string;
}) {
const thread = yield* resolveThread(input.threadId);
if (!thread) {
return;
}
const sessionId = yield* ensureSessionForThread(input.threadId, input.createdAt);
const sessionId = yield* ensureSessionForThread(input.threadId, input.createdAt, {
...(input.approvalPolicy !== undefined ? { approvalPolicy: input.approvalPolicy } : {}),
...(input.sandboxMode !== undefined ? { sandboxMode: input.sandboxMode } : {}),
});
const normalizedInput = toNonEmptyProviderInput(input.messageText);
const normalizedAttachments = input.attachments ?? [];

Expand Down Expand Up @@ -231,6 +267,10 @@ const make = Effect.gen(function* () {
...(message.attachments !== undefined ? { attachments: message.attachments } : {}),
...(event.payload.model !== undefined ? { model: event.payload.model } : {}),
...(event.payload.effort !== undefined ? { effort: event.payload.effort } : {}),
...(event.payload.approvalPolicy !== undefined
? { approvalPolicy: event.payload.approvalPolicy }
: {}),
...(event.payload.sandboxMode !== undefined ? { sandboxMode: event.payload.sandboxMode } : {}),
createdAt: event.payload.createdAt,
});
});
Expand Down
12 changes: 12 additions & 0 deletions apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts
Original file line number Diff line number Diff line change
Expand Up @@ -371,6 +371,12 @@ const make = Effect.gen(function* () {
providerName: event.provider,
providerSessionId: event.sessionId,
providerThreadId,
...(thread.session?.approvalPolicy !== undefined
? { approvalPolicy: thread.session.approvalPolicy }
: {}),
...(thread.session?.sandboxMode !== undefined
? { sandboxMode: thread.session.sandboxMode }
: {}),
activeTurnId,
lastError,
updatedAt: now,
Expand Down Expand Up @@ -479,6 +485,12 @@ const make = Effect.gen(function* () {
providerName: event.provider,
providerSessionId: event.sessionId,
providerThreadId,
...(thread.session?.approvalPolicy !== undefined
? { approvalPolicy: thread.session.approvalPolicy }
: {}),
...(thread.session?.sandboxMode !== undefined
? { sandboxMode: thread.session.sandboxMode }
: {}),
activeTurnId: toTurnId(event.turnId) ?? null,
lastError: event.message,
updatedAt: now,
Expand Down
10 changes: 10 additions & 0 deletions apps/server/src/orchestration/decider.projectScripts.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -152,6 +152,8 @@ describe("decider project scripts", () => {
},
model: "gpt-5",
effort: "high",
approvalPolicy: "on-request",
sandboxMode: "workspace-write",
createdAt: now,
},
readModel,
Expand All @@ -169,5 +171,13 @@ describe("decider project scripts", () => {
return;
}
expect(turnStartEvent.payload.assistantDeliveryMode).toBe("buffered");
expect(turnStartEvent.payload).toMatchObject({
threadId: ThreadId.makeUnsafe("thread-1"),
messageId: asMessageId("message-user-1"),
model: "gpt-5",
effort: "high",
approvalPolicy: "on-request",
sandboxMode: "workspace-write",
});
});
});
4 changes: 4 additions & 0 deletions apps/server/src/orchestration/decider.ts
Original file line number Diff line number Diff line change
Expand Up @@ -254,6 +254,10 @@ export const decideOrchestrationCommand = Effect.fn("decideOrchestrationCommand"
...(command.model !== undefined ? { model: command.model } : {}),
...(command.effort !== undefined ? { effort: command.effort } : {}),
assistantDeliveryMode: command.assistantDeliveryMode ?? DEFAULT_ASSISTANT_DELIVERY_MODE,
...(command.approvalPolicy !== undefined
? { approvalPolicy: command.approvalPolicy }
: {}),
...(command.sandboxMode !== undefined ? { sandboxMode: command.sandboxMode } : {}),
createdAt: command.createdAt,
},
};
Expand Down
5 changes: 5 additions & 0 deletions apps/web/src/components/ChatView.tsx
Original file line number Diff line number Diff line change
Expand Up @@ -1496,6 +1496,9 @@ export default function ChatView({ threadId }: ChatViewProps) {
),
);
attemptedTurnStart = true;
const approvalPolicy = state.runtimeMode === "full-access" ? "never" : "on-request";
const sandboxMode =
state.runtimeMode === "full-access" ? "danger-full-access" : "workspace-write";
await api.orchestration.dispatchCommand({
type: "thread.turn.start",
commandId: newCommandId(),
Expand All @@ -1509,6 +1512,8 @@ export default function ChatView({ threadId }: ChatViewProps) {
model: selectedModel || undefined,
effort: selectedEffort || undefined,
assistantDeliveryMode: settings.enableAssistantStreaming ? "streaming" : "buffered",
approvalPolicy,
sandboxMode,
createdAt: messageCreatedAt,
});
} catch (err) {
Expand Down
Loading