1import { assert, it } from "@effect/vitest";
2import {
3 CommandId,
4 EventId,
5 MessageId,
6 ProjectId,
7 ProviderDriverKind,
8 ProviderInstanceId,
9 ThreadId,
10} from "@t3tools/contracts";
11import * as DateTime from "effect/DateTime";
12import * as Effect from "effect/Effect";
13import * as Layer from "effect/Layer";
14import * as SqlitePersistence from "../persistence/Sqlite.ts";
15import { CodexProviderCapabilitiesV2 } from "./Adapters/CodexAdapterV2.ts";
16import { DispatchModeLimit } from "./DispatchModeLimit.ts";
17import * as EventSink from "./EventSink.ts";
18import * as Orchestrator from "./Orchestrator.ts";
19import * as ProjectionStore from "./ProjectionStore.ts";
20import * as ProviderAdapterRegistry from "./ProviderAdapterRegistry.ts";
21import * as ProviderReplayHarness from "./testkit/ProviderReplayHarness.ts";
22import type * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter";
23
24const instanceId = ProviderInstanceId.make("codex");
25const adapter = {
26 instanceId,
27 driver: ProviderDriverKind.make("codex"),
28 getCapabilities: () => Effect.succeed(CodexProviderCapabilitiesV2),
29 planSelectionTransition: () => Effect.succeed({ type: "apply_on_next_turn" as const }),
30 openSession: () => Effect.die("No provider process needed for metadata commands"),
31} as ProviderAdapter.ProviderAdapterV2["Service"];
32const layerDatabase = SqlitePersistence.layerMemory;
33const layerTest = Layer.mergeAll(
34 layerDatabase,
35 ProjectionStore.layer.pipe(Layer.provide(layerDatabase)),
36 ProviderReplayHarness.layerWithRegistry(
37 { name: "dispatch-mode-limit" },
38 ProviderAdapterRegistry.layerFromAdapters([adapter]),
39 { databaseLayer: layerDatabase, runEffectWorker: false },
40 ),
41);
42
43const supervised = { runtimeMode: "approval-required", interactionMode: "default" } as const;
44
45const createThread = (threadId: ThreadId) =>
46 Effect.gen(function* () {
47 const orchestrator = yield* Orchestrator.OrchestratorV2;
48 yield* orchestrator.dispatch({
49 type: "thread.create",
50 commandId: CommandId.make(`create:${threadId}`),
51 threadId,
52 projectId: ProjectId.make("project:dispatch-mode-limit"),
53 title: "Before",
54 modelSelection: { instanceId, model: "gpt-5" },
55 ...supervised,
56 branch: null,
57 worktreePath: null,
58 createdBy: "user",
59 creationSource: "web",
60 });
61 });
62
63it.layer(layerTest)("DispatchModeLimit", (it) => {
64 it.effect("refuses a limited command on a thread its user raised, and records nothing", () =>
65 Effect.gen(function* () {
66 const orchestrator = yield* Orchestrator.OrchestratorV2;
67 const projections = yield* ProjectionStore.ProjectionStoreV2;
68 const threadId = ThreadId.make("thread:limit-raised");
69 yield* createThread(threadId);
70 yield* orchestrator.dispatch({
71 type: "thread.runtime-mode.set",
72 commandId: CommandId.make("raise"),
73 threadId,
74 runtimeMode: "full-access",
75 });
76 const rename = {
77 type: "thread.metadata.update",
78 commandId: CommandId.make("rename"),
79 threadId,
80 title: "After",
81 } as const;
82 const refused = yield* orchestrator
83 .dispatch(rename)
84 .pipe(Effect.provideService(DispatchModeLimit, supervised), Effect.flip);
85 assert.equal(refused._tag, "OrchestratorThreadAboveModeLimitError");
86 assert.equal((yield* projections.getThreadShell(threadId))?.title, "Before");
87
88 yield* orchestrator.dispatch({
89 type: "thread.runtime-mode.set",
90 commandId: CommandId.make("lower"),
91 threadId,
92 runtimeMode: "approval-required",
93 });
94 yield* orchestrator
95 .dispatch(rename)
96 .pipe(Effect.provideService(DispatchModeLimit, supervised));
97 assert.equal((yield* projections.getThreadShell(threadId))?.title, "After");
98 }),
99 );
100
101 it.effect("refuses to fork a source thread its user raised", () =>
102 Effect.gen(function* () {
103 const orchestrator = yield* Orchestrator.OrchestratorV2;
104 const projections = yield* ProjectionStore.ProjectionStoreV2;
105 const sourceThreadId = ThreadId.make("thread:limit-fork-source");
106 const targetThreadId = ThreadId.make("thread:limit-fork-target");
107 yield* createThread(sourceThreadId);
108 yield* orchestrator.dispatch({
109 type: "thread.runtime-mode.set",
110 commandId: CommandId.make("raise-source"),
111 threadId: sourceThreadId,
112 runtimeMode: "full-access",
113 });
114 const fork = {
115 type: "thread.fork",
116 commandId: CommandId.make("fork"),
117 sourceThreadId,
118 targetThreadId,
119 sourcePoint: { type: "latest_stable" },
120 createdBy: "agent",
121 creationSource: "mcp",
122 } as const;
123 const refused = yield* orchestrator
124 .dispatch(fork)
125 .pipe(Effect.provideService(DispatchModeLimit, supervised), Effect.flip);
126 assert.ok(refused._tag === "OrchestratorThreadAboveModeLimitError");
127 assert.equal(refused.threadId, sourceThreadId);
128 assert.isNull(yield* projections.getThreadShell(targetThreadId));
129
130 yield* orchestrator.dispatch({
131 type: "thread.runtime-mode.set",
132 commandId: CommandId.make("lower-source"),
133 threadId: sourceThreadId,
134 runtimeMode: "approval-required",
135 });
136 const retried = yield* orchestrator
137 .dispatch(fork)
138 .pipe(Effect.provideService(DispatchModeLimit, supervised), Effect.flip);
139 assert.equal(retried._tag, "OrchestratorDispatchError");
140 }),
141 );
142});
143
144const layerPolicyDatabase = SqlitePersistence.layerMemory;
145const layerWithCeiling = Layer.mergeAll(
146 layerPolicyDatabase,
147 ProjectionStore.layer.pipe(Layer.provide(layerPolicyDatabase)),
148 ProviderReplayHarness.layerWithRegistry(
149 { name: "runtime-mode-ceiling" },
150 ProviderAdapterRegistry.layerFromAdapters([adapter]),
151 {
152 databaseLayer: layerPolicyDatabase,
153 runEffectWorker: false,
154 managedPolicy: {
155 schemaVersion: 1,
156 organization: "Example Corp",
157 limits: { maxRuntimeMode: "auto-accept-edits" },
158 },
159 },
160 ),
161);
162
163const sendMessage = (threadId: ThreadId, id: string) => ({
164 type: "message.dispatch" as const,
165 commandId: CommandId.make(`send:${id}`),
166 threadId,
167 messageId: MessageId.make(`message:${id}`),
168 text: id,
169 attachments: [],
170 dispatchMode: { type: "start_immediately" as const },
171 createdBy: "user" as const,
172 creationSource: "web" as const,
173});
174
175/** A thread that ran in full access before the policy arrived, as an import or an older build leaves it. */
176const raiseBehindPolicy = (threadId: ThreadId) =>
177 Effect.gen(function* () {
178 const orchestrator = yield* Orchestrator.OrchestratorV2;
179 const sink = yield* EventSink.EventSinkV2;
180 yield* createThread(threadId);
181 const { thread } = yield* orchestrator.getThreadRecords(threadId, []);
182 yield* sink.write({
183 events: [
184 {
185 id: EventId.make(`event:raise-behind-policy:${threadId}`),
186 type: "thread.runtime-mode-updated",
187 threadId,
188 occurredAt: yield* DateTime.now,
189 payload: { ...thread, runtimeMode: "full-access" },
190 },
191 ],
192 });
193 });
194
195it.layer(layerWithCeiling)("runtime mode ceiling", (it) => {
196 it.effect("refuses the user's own thread and mode changes above the ceiling", () =>
197 Effect.gen(function* () {
198 const orchestrator = yield* Orchestrator.OrchestratorV2;
199 const projections = yield* ProjectionStore.ProjectionStoreV2;
200 const threadId = ThreadId.make("thread:ceiling-user");
201 const create = {
202 type: "thread.create",
203 commandId: CommandId.make("create:ceiling-user:full"),
204 threadId,
205 projectId: ProjectId.make("project:dispatch-mode-limit"),
206 title: "Full",
207 modelSelection: { instanceId, model: "gpt-5" },
208 runtimeMode: "full-access",
209 interactionMode: "default",
210 branch: null,
211 worktreePath: null,
212 createdBy: "user",
213 creationSource: "web",
214 } as const;
215 const refusedCreate = yield* orchestrator.dispatch(create).pipe(Effect.flip);
216 assert.ok(refusedCreate._tag === "OrchestratorRuntimeModeAboveCeilingError");
217 assert.equal(refusedCreate.refused, "mode");
218 assert.equal(refusedCreate.message, "Example Corp limits threads to Auto-accept edits.");
219 assert.isNull(yield* projections.getThreadShell(threadId));
220
221 yield* createThread(threadId);
222 const raise = yield* orchestrator
223 .dispatch({
224 type: "thread.runtime-mode.set",
225 commandId: CommandId.make("raise:ceiling-user"),
226 threadId,
227 runtimeMode: "auto",
228 })
229 .pipe(Effect.flip);
230 assert.equal(raise._tag, "OrchestratorRuntimeModeAboveCeilingError");
231 yield* orchestrator.dispatch({
232 type: "thread.runtime-mode.set",
233 commandId: CommandId.make("raise-to-ceiling:ceiling-user"),
234 threadId,
235 runtimeMode: "auto-accept-edits",
236 });
237 assert.equal((yield* projections.getThreadShell(threadId))?.runtimeMode, "auto-accept-edits");
238 }),
239 );
240
241 it.effect("narrows an MCP caller's limit to the ceiling inside the lock", () =>
242 Effect.gen(function* () {
243 const orchestrator = yield* Orchestrator.OrchestratorV2;
244 const projections = yield* ProjectionStore.ProjectionStoreV2;
245 const threadId = ThreadId.make("thread:ceiling-mcp");
246 yield* createThread(threadId);
247 const fullAccessCaller = { runtimeMode: "full-access", interactionMode: "default" } as const;
248 const raise = yield* orchestrator
249 .dispatch({
250 type: "thread.runtime-mode.set",
251 commandId: CommandId.make("raise:ceiling-mcp"),
252 threadId,
253 runtimeMode: "full-access",
254 })
255 .pipe(Effect.provideService(DispatchModeLimit, fullAccessCaller), Effect.flip);
256 assert.equal(raise._tag, "OrchestratorRuntimeModeAboveCeilingError");
257
258 const raisedThreadId = ThreadId.make("thread:ceiling-mcp-raised");
259 yield* raiseBehindPolicy(raisedThreadId);
260 const rename = yield* orchestrator
261 .dispatch({
262 type: "thread.metadata.update",
263 commandId: CommandId.make("rename:ceiling-mcp"),
264 threadId: raisedThreadId,
265 title: "After",
266 })
267 .pipe(Effect.provideService(DispatchModeLimit, fullAccessCaller), Effect.flip);
268 assert.equal(rename._tag, "OrchestratorThreadAboveModeLimitError");
269 assert.equal((yield* projections.getThreadShell(raisedThreadId))?.title, "Before");
270 }),
271 );
272
273 it.effect("refuses turns on a thread above the ceiling until its mode is lowered", () =>
274 Effect.gen(function* () {
275 const orchestrator = yield* Orchestrator.OrchestratorV2;
276 const projections = yield* ProjectionStore.ProjectionStoreV2;
277 const threadId = ThreadId.make("thread:ceiling-raised");
278 yield* raiseBehindPolicy(threadId);
279
280 const refused = yield* orchestrator
281 .dispatch(sendMessage(threadId, "first"))
282 .pipe(Effect.flip);
283 assert.ok(refused._tag === "OrchestratorRuntimeModeAboveCeilingError");
284 assert.equal(refused.refused, "turn");
285 assert.equal(
286 refused.message,
287 "Example Corp limits threads to Auto-accept edits. Switch this thread's mode to continue.",
288 );
289 assert.lengthOf((yield* orchestrator.getThreadRecords(threadId, ["runs"])).runs, 0);
290
291 // The user's own housekeeping and the way back down stay open.
292 yield* orchestrator.dispatch({
293 type: "thread.metadata.update",
294 commandId: CommandId.make("rename:ceiling-raised"),
295 threadId,
296 title: "After",
297 });
298 yield* orchestrator.dispatch({
299 type: "thread.runtime-mode.set",
300 commandId: CommandId.make("lower:ceiling-raised"),
301 threadId,
302 runtimeMode: "approval-required",
303 });
304 // Refusals record no receipt, so the same command goes through once the mode is lowered.
305 yield* orchestrator.dispatch(sendMessage(threadId, "first"));
306 assert.lengthOf((yield* orchestrator.getThreadRecords(threadId, ["runs"])).runs, 1);
307 assert.equal((yield* projections.getThreadShell(threadId))?.title, "After");
308 }),
309 );
310
311 it.effect("starts a queue that waited above the ceiling once the mode is lowered", () =>
312 Effect.gen(function* () {
313 const orchestrator = yield* Orchestrator.OrchestratorV2;
314 const sink = yield* EventSink.EventSinkV2;
315 const threadId = ThreadId.make("thread:ceiling-queue");
316 yield* createThread(threadId);
317 yield* orchestrator.dispatch(sendMessage(threadId, "running"));
318 yield* orchestrator.dispatch({
319 ...sendMessage(threadId, "waiting"),
320 dispatchMode: { type: "queue_after_active" },
321 });
322 // The policy arrives with the thread above it and its first run ending.
323 const { thread, runs } = yield* orchestrator.getThreadRecords(threadId, ["runs"]);
324 const [running, waiting] = runs;
325 const occurredAt = yield* DateTime.now;
326 yield* sink.write({
327 events: [
328 {
329 id: EventId.make("event:ceiling-queue:raise"),
330 type: "thread.runtime-mode-updated",
331 threadId,
332 occurredAt,
333 payload: { ...thread, runtimeMode: "full-access" },
334 },
335 {
336 id: EventId.make("event:ceiling-queue:complete"),
337 type: "run.updated",
338 threadId,
339 runId: running!.id,
340 providerInstanceId: running!.providerInstanceId,
341 occurredAt,
342 payload: { ...running!, status: "completed" },
343 },
344 ],
345 });
346
347 yield* orchestrator.dispatch({
348 type: "thread.runtime-mode.set",
349 commandId: CommandId.make("lower:ceiling-queue"),
350 threadId,
351 runtimeMode: "approval-required",
352 });
353 const started = (yield* orchestrator.getThreadRecords(threadId, ["runs"])).runs.find(
354 (run) => run.id === waiting!.id,
355 );
356 assert.notEqual(started?.status, "queued");
357 }),
358 );
359});