Skip to content

Commit c5442d4

Browse files
kitlangtonkitlangton
authored
refactor(session): effectify SessionPrompt service (#19483)
1 parent fa95a61 commit c5442d4

20 files changed

Lines changed: 4313 additions & 2016 deletions

‎packages/opencode/src/effect/runner.ts‎

Lines changed: 216 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,216 @@
1+
import{Cause,Deferred,Effect,Exit,Fiber,Option,Schema,Scope,SynchronizedRef}from"effect"
2+
3+
exportinterfaceRunner<A,E=never>{
4+
readonlystate: Runner.State<A,E>
5+
readonlybusy: boolean
6+
readonlyensureRunning: (work: Effect.Effect<A,E>)=>Effect.Effect<A,E>
7+
readonlystartShell: (work: (signal: AbortSignal)=>Effect.Effect<A,E>)=>Effect.Effect<A,E>
8+
readonlycancel: Effect.Effect<void>
9+
}
10+
11+
exportnamespaceRunner{
12+
exportclassCancelledextendsSchema.TaggedErrorClass<Cancelled>()("RunnerCancelled",{}){}
13+
14+
interfaceRunHandle<A,E>{
15+
id: number
16+
done: Deferred.Deferred<A,E|Cancelled>
17+
fiber: Fiber.Fiber<A,E>
18+
}
19+
20+
interfaceShellHandle<A,E>{
21+
id: number
22+
fiber: Fiber.Fiber<A,E>
23+
abort: AbortController
24+
}
25+
26+
interfacePendingHandle<A,E>{
27+
id: number
28+
done: Deferred.Deferred<A,E|Cancelled>
29+
work: Effect.Effect<A,E>
30+
}
31+
32+
exporttypeState<A,E>=
33+
|{readonly_tag: "Idle"}
34+
|{readonly_tag: "Running";readonlyrun: RunHandle<A,E>}
35+
|{readonly_tag: "Shell";readonlyshell: ShellHandle<A,E>}
36+
|{readonly_tag: "ShellThenRun";readonlyshell: ShellHandle<A,E>;readonlyrun: PendingHandle<A,E>}
37+
38+
exportconstmake=<A,E=never>(
39+
scope: Scope.Scope,
40+
opts?: {
41+
onIdle?: Effect.Effect<void>
42+
onBusy?: Effect.Effect<void>
43+
onInterrupt?: Effect.Effect<A,E>
44+
busy?: ()=>never
45+
},
46+
): Runner<A,E>=>{
47+
constref=SynchronizedRef.makeUnsafe<State<A,E>>({_tag: "Idle"})
48+
constidle=opts?.onIdle??Effect.void
49+
constbusy=opts?.onBusy??Effect.void
50+
constonInterrupt=opts?.onInterrupt
51+
letids=0
52+
53+
conststate=()=>SynchronizedRef.getUnsafe(ref)
54+
constnext=()=>{
55+
ids+=1
56+
returnids
57+
}
58+
59+
constcomplete=(done: Deferred.Deferred<A,E|Cancelled>,exit: Exit.Exit<A,E>)=>
60+
Exit.isFailure(exit)&&Cause.hasInterruptsOnly(exit.cause)
61+
? Deferred.fail(done,newCancelled()).pipe(Effect.asVoid)
62+
: Deferred.done(done,exit).pipe(Effect.asVoid)
63+
64+
constidleIfCurrent=()=>
65+
SynchronizedRef.modify(ref,(st)=>[st._tag==="Idle" ? idle : Effect.void,st]asconst).pipe(Effect.flatten)
66+
67+
constfinishRun=(id: number,done: Deferred.Deferred<A,E|Cancelled>,exit: Exit.Exit<A,E>)=>
68+
SynchronizedRef.modify(
69+
ref,
70+
(st)=>
71+
[
72+
Effect.gen(function*(){
73+
if(st._tag==="Running"&&st.run.id===id)yield*idle
74+
yield*complete(done,exit)
75+
}),
76+
st._tag==="Running"&&st.run.id===id ? ({_tag: "Idle"}asconst) : st,
77+
]asconst,
78+
).pipe(Effect.flatten)
79+
80+
conststartRun=(work: Effect.Effect<A,E>,done: Deferred.Deferred<A,E|Cancelled>)=>
81+
Effect.gen(function*(){
82+
constid=next()
83+
constfiber=yield*work.pipe(
84+
Effect.onExit((exit)=>finishRun(id,done,exit)),
85+
Effect.forkIn(scope),
86+
)
87+
return{ id, done, fiber }satisfiesRunHandle<A,E>
88+
})
89+
90+
constfinishShell=(id: number)=>
91+
SynchronizedRef.modifyEffect(
92+
ref,
93+
Effect.fnUntraced(function*(st){
94+
if(st._tag==="Shell"&&st.shell.id===id)return[idle,{_tag: "Idle"}]asconst
95+
if(st._tag==="ShellThenRun"&&st.shell.id===id){
96+
construn=yield*startRun(st.run.work,st.run.done)
97+
return[Effect.void,{_tag: "Running", run }]asconst
98+
}
99+
return[Effect.void,st]asconst
100+
}),
101+
).pipe(Effect.flatten)
102+
103+
conststopShell=(shell: ShellHandle<A,E>)=>
104+
Effect.gen(function*(){
105+
shell.abort.abort()
106+
constexit=yield*Fiber.await(shell.fiber).pipe(Effect.timeoutOption("100 millis"))
107+
if(Option.isNone(exit))yield*Fiber.interrupt(shell.fiber)
108+
yield*Fiber.await(shell.fiber).pipe(Effect.exit,Effect.asVoid)
109+
})
110+
111+
constensureRunning=(work: Effect.Effect<A,E>)=>
112+
SynchronizedRef.modifyEffect(
113+
ref,
114+
Effect.fnUntraced(function*(st){
115+
switch(st._tag){
116+
case"Running":
117+
case"ShellThenRun":
118+
return[Deferred.await(st.run.done),st]asconst
119+
case"Shell": {
120+
construn={
121+
id: next(),
122+
done: yield*Deferred.make<A,E|Cancelled>(),
123+
work,
124+
}satisfiesPendingHandle<A,E>
125+
return[Deferred.await(run.done),{_tag: "ShellThenRun",shell: st.shell, run }]asconst
126+
}
127+
case"Idle": {
128+
constdone=yield*Deferred.make<A,E|Cancelled>()
129+
construn=yield*startRun(work,done)
130+
return[Deferred.await(done),{_tag: "Running", run }]asconst
131+
}
132+
}
133+
}),
134+
).pipe(
135+
Effect.flatten,
136+
Effect.catch((e): Effect.Effect<A,E>=>
137+
einstanceofCancelled ? (onInterrupt??Effect.die(e)) : Effect.fail(easE),
138+
),
139+
)
140+
141+
conststartShell=(work: (signal: AbortSignal)=>Effect.Effect<A,E>)=>
142+
SynchronizedRef.modifyEffect(
143+
ref,
144+
Effect.fnUntraced(function*(st){
145+
if(st._tag!=="Idle"){
146+
return[
147+
Effect.sync(()=>{
148+
if(opts?.busy)opts.busy()
149+
thrownewError("Runner is busy")
150+
}),
151+
st,
152+
]asconst
153+
}
154+
yield*busy
155+
constid=next()
156+
constabort=newAbortController()
157+
constfiber=yield*work(abort.signal).pipe(Effect.ensuring(finishShell(id)),Effect.forkChild)
158+
constshell={ id, fiber, abort }satisfiesShellHandle<A,E>
159+
return[
160+
Effect.gen(function*(){
161+
constexit=yield*Fiber.await(fiber)
162+
if(Exit.isSuccess(exit))returnexit.value
163+
if(Cause.hasInterruptsOnly(exit.cause)&&onInterrupt)returnyield*onInterrupt
164+
returnyield*Effect.failCause(exit.cause)
165+
}),
166+
{_tag: "Shell", shell },
167+
]asconst
168+
}),
169+
).pipe(Effect.flatten)
170+
171+
constcancel=SynchronizedRef.modify(ref,(st)=>{
172+
switch(st._tag){
173+
case"Idle":
174+
return[Effect.void,st]asconst
175+
case"Running":
176+
return[
177+
Effect.gen(function*(){
178+
yield*Fiber.interrupt(st.run.fiber)
179+
yield*Deferred.await(st.run.done).pipe(Effect.exit,Effect.asVoid)
180+
yield*idleIfCurrent()
181+
}),
182+
{_tag: "Idle"}asconst,
183+
]asconst
184+
case"Shell":
185+
return[
186+
Effect.gen(function*(){
187+
yield*stopShell(st.shell)
188+
yield*idleIfCurrent()
189+
}),
190+
{_tag: "Idle"}asconst,
191+
]asconst
192+
case"ShellThenRun":
193+
return[
194+
Effect.gen(function*(){
195+
yield*Deferred.fail(st.run.done,newCancelled()).pipe(Effect.asVoid)
196+
yield*stopShell(st.shell)
197+
yield*idleIfCurrent()
198+
}),
199+
{_tag: "Idle"}asconst,
200+
]asconst
201+
}
202+
}).pipe(Effect.flatten)
203+
204+
return{
205+
getstate(){
206+
returnstate()
207+
},
208+
getbusy(){
209+
returnstate()._tag!=="Idle"
210+
},
211+
ensureRunning,
212+
startShell,
213+
cancel,
214+
}
215+
}
216+
}

‎packages/opencode/src/server/routes/session.ts‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -381,7 +381,7 @@ export const SessionRoutes = lazy(() =>
381381
}),
382382
),
383383
async(c)=>{
384-
SessionPrompt.cancel(c.req.valid("param").sessionID)
384+
awaitSessionPrompt.cancel(c.req.valid("param").sessionID)
385385
returnc.json(true)
386386
},
387387
)
@@ -699,7 +699,7 @@ export const SessionRoutes = lazy(() =>
699699
),
700700
async(c)=>{
701701
constparams=c.req.valid("param")
702-
SessionPrompt.assertNotBusy(params.sessionID)
702+
awaitSessionPrompt.assertNotBusy(params.sessionID)
703703
awaitSession.removeMessage({
704704
sessionID: params.sessionID,
705705
messageID: params.messageID,

‎packages/opencode/src/session/compaction.ts‎

Lines changed: 29 additions & 35 deletions
Original file line numberDiff line numberDiff line change
@@ -15,7 +15,7 @@ import { Plugin } from "@/plugin"
1515
import{Config}from"@/config/config"
1616
import{NotFoundError}from"@/storage/db"
1717
import{ModelID,ProviderID}from"@/provider/schema"
18-
import{Cause,Effect,Exit,Layer,ServiceMap}from"effect"
18+
import{Effect,Layer,ServiceMap}from"effect"
1919
import{makeRuntime}from"@/effect/run-service"
2020
import{isOverflowasoverflow}from"./overflow"
2121

@@ -45,7 +45,6 @@ export namespace SessionCompaction {
4545
parentID: MessageID
4646
messages: MessageV2.WithParts[]
4747
sessionID: SessionID
48-
abort: AbortSignal
4948
auto: boolean
5049
overflow?: boolean
5150
})=>Effect.Effect<"continue"|"stop">
@@ -135,20 +134,28 @@ export namespace SessionCompaction {
135134
parentID: MessageID
136135
messages: MessageV2.WithParts[]
137136
sessionID: SessionID
138-
abort: AbortSignal
139137
auto: boolean
140138
overflow?: boolean
141139
}){
142-
constuserMessage=input.messages.findLast((m)=>m.info.id===input.parentID)!.infoasMessageV2.User
140+
constparent=input.messages.findLast((m)=>m.info.id===input.parentID)
141+
if(!parent||parent.info.role!=="user"){
142+
thrownewError(`Compaction parent must be a user message: ${input.parentID}`)
143+
}
144+
constuserMessage=parent.info
143145

144146
letmessages=input.messages
145-
letreplay: MessageV2.WithParts|undefined
147+
letreplay:
148+
|{
149+
info: MessageV2.User
150+
parts: MessageV2.Part[]
151+
}
152+
|undefined
146153
if(input.overflow){
147154
constidx=input.messages.findIndex((m)=>m.info.id===input.parentID)
148155
for(leti=idx-1;i>=0;i--){
149156
constmsg=input.messages[i]
150157
if(msg.info.role==="user"&&!msg.parts.some((p)=>p.type==="compaction")){
151-
replay=msg
158+
replay={info: msg.info,parts: msg.parts}
152159
messages=input.messages.slice(0,i)
153160
break
154161
}
@@ -206,7 +213,7 @@ When constructing the summary, try to stick to this template:
206213
constmsgs=structuredClone(messages)
207214
yield*plugin.trigger("experimental.chat.messages.transform",{},{messages: msgs})
208215
constmodelMessages=yield*Effect.promise(()=>MessageV2.toModelMessages(msgs,model,{stripMedia: true}))
209-
constmsg=(yield*session.updateMessage({
216+
constmsg: MessageV2.Assistant={
210217
id: MessageID.ascending(),
211218
role: "assistant",
212219
parentID: input.parentID,
@@ -231,25 +238,17 @@ When constructing the summary, try to stick to this template:
231238
time: {
232239
created: Date.now(),
233240
},
234-
}))asMessageV2.Assistant
241+
}
242+
yield*session.updateMessage(msg)
235243
constprocessor=yield*processors.create({
236244
assistantMessage: msg,
237245
sessionID: input.sessionID,
238246
model,
239-
abort: input.abort,
240-
})
241-
constcancel=Effect.fn("SessionCompaction.cancel")(function*(){
242-
if(!input.abort.aborted||msg.time.completed)return
243-
msg.error=msg.error??newMessageV2.AbortedError({message: "Aborted"}).toObject()
244-
msg.finish=msg.finish??"error"
245-
msg.time.completed=Date.now()
246-
yield*session.updateMessage(msg)
247247
})
248248
constresult=yield*processor
249249
.process({
250250
user: userMessage,
251251
agent,
252-
abort: input.abort,
253252
sessionID: input.sessionID,
254253
tools: {},
255254
system: [],
@@ -262,7 +261,7 @@ When constructing the summary, try to stick to this template:
262261
],
263262
model,
264263
})
265-
.pipe(Effect.ensuring(cancel()))
264+
.pipe(Effect.onInterrupt(()=>processor.abort()))
266265

267266
if(result==="compact"){
268267
processor.message.error=newMessageV2.ContextOverflowError({
@@ -277,7 +276,7 @@ When constructing the summary, try to stick to this template:
277276

278277
if(result==="continue"&&input.auto){
279278
if(replay){
280-
constoriginal=replay.infoasMessageV2.User
279+
constoriginal=replay.info
281280
constreplayMsg=yield*session.updateMessage({
282281
id: MessageID.ascending(),
283282
role: "user",
@@ -386,7 +385,7 @@ When constructing the summary, try to stick to this template:
386385
),
387386
)
388387

389-
const{ runPromise, runPromiseExit}=makeRuntime(Service,defaultLayer)
388+
const{ runPromise }=makeRuntime(Service,defaultLayer)
390389

391390
exportasyncfunctionisOverflow(input: {tokens: MessageV2.Assistant["tokens"];model: Provider.Model}){
392391
returnrunPromise((svc)=>svc.isOverflow(input))
@@ -396,21 +395,16 @@ When constructing the summary, try to stick to this template:
396395
returnrunPromise((svc)=>svc.prune(input))
397396
}
398397

399-
exportasyncfunctionprocess(input: {
400-
parentID: MessageID
401-
messages: MessageV2.WithParts[]
402-
sessionID: SessionID
403-
abort: AbortSignal
404-
auto: boolean
405-
overflow?: boolean
406-
}){
407-
constexit=awaitrunPromiseExit((svc)=>svc.process(input),{signal: input.abort})
408-
if(Exit.isFailure(exit)){
409-
if(Cause.hasInterrupts(exit.cause)&&input.abort.aborted)return"stop"
410-
throwCause.squash(exit.cause)
411-
}
412-
returnexit.value
413-
}
398+
exportconstprocess=fn(
399+
z.object({
400+
parentID: MessageID.zod,
401+
messages: z.custom<MessageV2.WithParts[]>(),
402+
sessionID: SessionID.zod,
403+
auto: z.boolean(),
404+
overflow: z.boolean().optional(),
405+
}),
406+
(input)=>runPromise((svc)=>svc.process(input)),
407+
)
414408

415409
exportconstcreate=fn(
416410
z.object({

0 commit comments

Comments
 (0)