Skip to content

Commit 10441ef

Browse files
kitlangtonkitlangton
authored
refactor(effect): extract session run state service (#21744)
1 parent 3199383 commit 10441ef

9 files changed

Lines changed: 186 additions & 113 deletions

File tree

‎packages/opencode/src/project/vcs.ts‎

Lines changed: 30 additions & 32 deletions
Original file line numberDiff line numberDiff line change
@@ -161,39 +161,37 @@ export namespace Vcs {
161161
const bus = yield* Bus.Service
162162

163163
const state = yield* InstanceState.make<State>(
164-
Effect.fn("Vcs.state")((ctx) =>
165-
Effect.gen(function* () {
166-
if (ctx.project.vcs !== "git") {
167-
return { current: undefined, root: undefined }
168-
}
169-
170-
const get = Effect.fnUntraced(function* () {
171-
return yield* git.branch(ctx.directory)
172-
})
173-
const [current, root] = yield* Effect.all([git.branch(ctx.directory), git.defaultBranch(ctx.directory)], {
174-
concurrency: 2,
175-
})
176-
const value = { current, root }
177-
log.info("initialized", { branch: value.current, default_branch: value.root?.name })
178-
179-
yield* bus.subscribe(FileWatcher.Event.Updated).pipe(
180-
Stream.filter((evt) => evt.properties.file.endsWith("HEAD")),
181-
Stream.runForEach((_evt) =>
182-
Effect.gen(function* () {
183-
const next = yield* get()
184-
if (next !== value.current) {
185-
log.info("branch changed", { from: value.current, to: next })
186-
value.current = next
187-
yield* bus.publish(Event.BranchUpdated, { branch: next })
188-
}
189-
}),
190-
),
191-
Effect.forkScoped,
192-
)
164+
Effect.fn("Vcs.state")(function* (ctx) {
165+
if (ctx.project.vcs !== "git") {
166+
return { current: undefined, root: undefined }
167+
}
193168

194-
return value
195-
}),
196-
),
169+
const get = Effect.fnUntraced(function* () {
170+
return yield* git.branch(ctx.directory)
171+
})
172+
const [current, root] = yield* Effect.all([git.branch(ctx.directory), git.defaultBranch(ctx.directory)], {
173+
concurrency: 2,
174+
})
175+
const value = { current, root }
176+
log.info("initialized", { branch: value.current, default_branch: value.root?.name })
177+
178+
yield* bus.subscribe(FileWatcher.Event.Updated).pipe(
179+
Stream.filter((evt) => evt.properties.file.endsWith("HEAD")),
180+
Stream.runForEach((_evt) =>
181+
Effect.gen(function* () {
182+
const next = yield* get()
183+
if (next !== value.current) {
184+
log.info("branch changed", { from: value.current, to: next })
185+
value.current = next
186+
yield* bus.publish(Event.BranchUpdated, { branch: next })
187+
}
188+
}),
189+
),
190+
Effect.forkScoped,
191+
)
192+
193+
return value
194+
}),
197195
)
198196

199197
return Service.of({

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

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,7 @@ import z from "zod"
66
import { Session } from "../../session"
77
import { MessageV2 } from "../../session/message-v2"
88
import { SessionPrompt } from "../../session/prompt"
9+
import { SessionRunState } from "@/session/run-state"
910
import { SessionCompaction } from "../../session/compaction"
1011
import { SessionRevert } from "../../session/revert"
1112
import { SessionStatus } from "@/session/status"
@@ -698,7 +699,7 @@ export const SessionRoutes = lazy(() =>
698699
),
699700
async (c) => {
700701
const params = c.req.valid("param")
701-
await SessionPrompt.assertNotBusy(params.sessionID)
702+
await SessionRunState.assertNotBusy(params.sessionID)
702703
await Session.removeMessage({
703704
sessionID: params.sessionID,
704705
messageID: params.messageID,

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

Lines changed: 6 additions & 59 deletions
Original file line numberDiff line numberDiff line change
@@ -20,7 +20,6 @@ import PROMPT_PLAN from "../session/prompt/plan.txt"
2020
import BUILD_SWITCH from "../session/prompt/build-switch.txt"
2121
import MAX_STEPS from "../session/prompt/max-steps.txt"
2222
import { ToolRegistry } from "../tool/registry"
23-
import { Runner } from "@/effect/runner"
2423
import { MCP } from "../mcp"
2524
import { LSP } from "../lsp"
2625
import { FileTime } from "../file/time"
@@ -48,6 +47,7 @@ import { Cause, Effect, Exit, Layer, Option, Scope, ServiceMap } from "effect"
4847
import { InstanceState } from "@/effect/instance-state"
4948
import { makeRuntime } from "@/effect/run-service"
5049
import { TaskTool } from "@/tool/task"
50+
import { SessionRunState } from "./run-state"
5151

5252
// @ts-ignore
5353
globalThis.AI_SDK_LOG_WARNINGS = false
@@ -66,7 +66,6 @@ export namespace SessionPrompt {
6666
const log = Log.create({ service: "session.prompt" })
6767

6868
export interface Interface {
69-
readonly assertNotBusy: (sessionID: SessionID) => Effect.Effect<void, Session.BusyError>
7069
readonly cancel: (sessionID: SessionID) => Effect.Effect<void>
7170
readonly prompt: (input: PromptInput) => Effect.Effect<MessageV2.WithParts>
7271
readonly loop: (input: z.infer<typeof LoopInput>) => Effect.Effect<MessageV2.WithParts>
@@ -99,55 +98,11 @@ export namespace SessionPrompt {
9998
const spawner = yield* ChildProcessSpawner.ChildProcessSpawner
10099
const scope = yield* Scope.Scope
101100
const instruction = yield* Instruction.Service
102-
103-
const state = yield* InstanceState.make(
104-
Effect.fn("SessionPrompt.state")(function* () {
105-
const runners = new Map<string, Runner<MessageV2.WithParts>>()
106-
yield* Effect.addFinalizer(
107-
Effect.fnUntraced(function* () {
108-
yield* Effect.forEach(runners.values(), (r) => r.cancel, { concurrency: "unbounded", discard: true })
109-
runners.clear()
110-
}),
111-
)
112-
return { runners }
113-
}),
114-
)
115-
116-
const getRunner = (runners: Map<string, Runner<MessageV2.WithParts>>, sessionID: SessionID) => {
117-
const existing = runners.get(sessionID)
118-
if (existing) return existing
119-
const runner = Runner.make<MessageV2.WithParts>(scope, {
120-
onIdle: Effect.gen(function* () {
121-
runners.delete(sessionID)
122-
yield* status.set(sessionID, { type: "idle" })
123-
}),
124-
onBusy: status.set(sessionID, { type: "busy" }),
125-
onInterrupt: lastAssistant(sessionID),
126-
busy: () => {
127-
throw new Session.BusyError(sessionID)
128-
},
129-
})
130-
runners.set(sessionID, runner)
131-
return runner
132-
}
133-
134-
const assertNotBusy: (sessionID: SessionID) => Effect.Effect<void, Session.BusyError> = Effect.fn(
135-
"SessionPrompt.assertNotBusy",
136-
)(function* (sessionID: SessionID) {
137-
const s = yield* InstanceState.get(state)
138-
const runner = s.runners.get(sessionID)
139-
if (runner?.busy) throw new Session.BusyError(sessionID)
140-
})
101+
const state = yield* SessionRunState.Service
141102

142103
const cancel = Effect.fn("SessionPrompt.cancel")(function* (sessionID: SessionID) {
143104
log.info("cancel", { sessionID })
144-
const s = yield* InstanceState.get(state)
145-
const runner = s.runners.get(sessionID)
146-
if (!runner || !runner.busy) {
147-
yield* status.set(sessionID, { type: "idle" })
148-
return
149-
}
150-
yield* runner.cancel
105+
yield* state.cancel(sessionID)
151106
})
152107

153108
const resolvePromptParts = Effect.fn("SessionPrompt.resolvePromptParts")(function* (template: string) {
@@ -1574,16 +1529,12 @@ NOTE: At any point in time through this workflow you should feel free to ask the
15741529
const loop: (input: z.infer<typeof LoopInput>) => Effect.Effect<MessageV2.WithParts> = Effect.fn(
15751530
"SessionPrompt.loop",
15761531
)(function* (input: z.infer<typeof LoopInput>) {
1577-
const s = yield* InstanceState.get(state)
1578-
const runner = getRunner(s.runners, input.sessionID)
1579-
return yield* runner.ensureRunning(runLoop(input.sessionID))
1532+
return yield* state.ensureRunning(input.sessionID, lastAssistant(input.sessionID), runLoop(input.sessionID))
15801533
})
15811534

15821535
const shell: (input: ShellInput) => Effect.Effect<MessageV2.WithParts> = Effect.fn("SessionPrompt.shell")(
15831536
function* (input: ShellInput) {
1584-
const s = yield* InstanceState.get(state)
1585-
const runner = getRunner(s.runners, input.sessionID)
1586-
return yield* runner.startShell(shellImpl(input))
1537+
return yield* state.startShell(input.sessionID, lastAssistant(input.sessionID), shellImpl(input))
15871538
},
15881539
)
15891540

@@ -1704,7 +1655,6 @@ NOTE: At any point in time through this workflow you should feel free to ask the
17041655
})
17051656

17061657
return Service.of({
1707-
assertNotBusy,
17081658
cancel,
17091659
prompt,
17101660
loop,
@@ -1718,6 +1668,7 @@ NOTE: At any point in time through this workflow you should feel free to ask the
17181668
const defaultLayer = Layer.unwrap(
17191669
Effect.sync(() =>
17201670
layer.pipe(
1671+
Layer.provide(SessionRunState.layer),
17211672
Layer.provide(SessionStatus.layer),
17221673
Layer.provide(SessionCompaction.defaultLayer),
17231674
Layer.provide(SessionProcessor.defaultLayer),
@@ -1741,10 +1692,6 @@ NOTE: At any point in time through this workflow you should feel free to ask the
17411692
)
17421693
const { runPromise } = makeRuntime(Service, defaultLayer)
17431694

1744-
export async function assertNotBusy(sessionID: SessionID) {
1745-
return runPromise((svc) => svc.assertNotBusy(SessionID.zod.parse(sessionID)))
1746-
}
1747-
17481695
export const PromptInput = z.object({
17491696
sessionID: SessionID.zod,
17501697
messageID: MessageID.zod.optional(),

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

Lines changed: 7 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -9,8 +9,9 @@ import { Log } from "../util/log"
99
import { Session } from "."
1010
import { MessageV2 } from "./message-v2"
1111
import { SessionID, MessageID, PartID } from "./schema"
12-
import { SessionPrompt } from "./prompt"
12+
import { SessionRunState } from "./run-state"
1313
import { SessionSummary } from "./summary"
14+
import { SessionStatus } from "./status"
1415

1516
export namespace SessionRevert {
1617
const log = Log.create({ service: "session.revert" })
@@ -38,9 +39,10 @@ export namespace SessionRevert {
3839
const storage = yield* Storage.Service
3940
const bus = yield* Bus.Service
4041
const summary = yield* SessionSummary.Service
42+
const state = yield* SessionRunState.Service
4143

4244
const revert = Effect.fn("SessionRevert.revert")(function* (input: RevertInput) {
43-
yield* Effect.promise(() => SessionPrompt.assertNotBusy(input.sessionID))
45+
yield* state.assertNotBusy(input.sessionID)
4446
const all = yield* sessions.messages({ sessionID: input.sessionID })
4547
let lastUser: MessageV2.User | undefined
4648
const session = yield* sessions.get(input.sessionID)
@@ -93,7 +95,7 @@ export namespace SessionRevert {
9395

9496
const unrevert = Effect.fn("SessionRevert.unrevert")(function* (input: { sessionID: SessionID }) {
9597
log.info("unreverting", input)
96-
yield* Effect.promise(() => SessionPrompt.assertNotBusy(input.sessionID))
98+
yield* state.assertNotBusy(input.sessionID)
9799
const session = yield* sessions.get(input.sessionID)
98100
if (!session.revert) return session
99101
if (session.revert.snapshot) yield* snap.restore(session.revert!.snapshot!)
@@ -151,6 +153,8 @@ export namespace SessionRevert {
151153
export const defaultLayer = Layer.unwrap(
152154
Effect.sync(() =>
153155
layer.pipe(
156+
Layer.provide(SessionRunState.layer),
157+
Layer.provide(SessionStatus.layer),
154158
Layer.provide(Session.defaultLayer),
155159
Layer.provide(Snapshot.defaultLayer),
156160
Layer.provide(Storage.defaultLayer),

‎packages/opencode/src/session/run-state.ts‎

Lines changed: 114 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,114 @@
1+
import { InstanceState } from "@/effect/instance-state"
2+
import { Runner } from "@/effect/runner"
3+
import { makeRuntime } from "@/effect/run-service"
4+
import { Effect, Layer, Scope, ServiceMap } from "effect"
5+
import { Session } from "."
6+
import { MessageV2 } from "./message-v2"
7+
import { SessionID } from "./schema"
8+
import { SessionStatus } from "./status"
9+
10+
export namespace SessionRunState {
11+
export interface Interface {
12+
readonly assertNotBusy: (sessionID: SessionID) => Effect.Effect<void>
13+
readonly cancel: (sessionID: SessionID) => Effect.Effect<void>
14+
readonly ensureRunning: (
15+
sessionID: SessionID,
16+
onInterrupt: Effect.Effect<MessageV2.WithParts>,
17+
work: Effect.Effect<MessageV2.WithParts>,
18+
) => Effect.Effect<MessageV2.WithParts>
19+
readonly startShell: (
20+
sessionID: SessionID,
21+
onInterrupt: Effect.Effect<MessageV2.WithParts>,
22+
work: Effect.Effect<MessageV2.WithParts>,
23+
) => Effect.Effect<MessageV2.WithParts>
24+
}
25+
26+
export class Service extends ServiceMap.Service<Service, Interface>()("@opencode/SessionRunState") {}
27+
28+
export const layer = Layer.effect(
29+
Service,
30+
Effect.gen(function* () {
31+
const status = yield* SessionStatus.Service
32+
33+
const state = yield* InstanceState.make(
34+
Effect.fn("SessionRunState.state")(function* () {
35+
const scope = yield* Scope.Scope
36+
const runners = new Map<SessionID, Runner<MessageV2.WithParts>>()
37+
yield* Effect.addFinalizer(
38+
Effect.fnUntraced(function* () {
39+
yield* Effect.forEach(runners.values(), (runner) => runner.cancel, {
40+
concurrency: "unbounded",
41+
discard: true,
42+
})
43+
runners.clear()
44+
}),
45+
)
46+
return { runners, scope }
47+
}),
48+
)
49+
50+
const runner = Effect.fn("SessionRunState.runner")(function* (
51+
sessionID: SessionID,
52+
onInterrupt: Effect.Effect<MessageV2.WithParts>,
53+
) {
54+
const data = yield* InstanceState.get(state)
55+
const existing = data.runners.get(sessionID)
56+
if (existing) return existing
57+
const next = Runner.make<MessageV2.WithParts>(data.scope, {
58+
onIdle: Effect.gen(function* () {
59+
data.runners.delete(sessionID)
60+
yield* status.set(sessionID, { type: "idle" })
61+
}),
62+
onBusy: status.set(sessionID, { type: "busy" }),
63+
onInterrupt,
64+
busy: () => {
65+
throw new Session.BusyError(sessionID)
66+
},
67+
})
68+
data.runners.set(sessionID, next)
69+
return next
70+
})
71+
72+
const assertNotBusy = Effect.fn("SessionRunState.assertNotBusy")(function* (sessionID: SessionID) {
73+
const data = yield* InstanceState.get(state)
74+
const existing = data.runners.get(sessionID)
75+
if (existing?.busy) throw new Session.BusyError(sessionID)
76+
})
77+
78+
const cancel = Effect.fn("SessionRunState.cancel")(function* (sessionID: SessionID) {
79+
const data = yield* InstanceState.get(state)
80+
const existing = data.runners.get(sessionID)
81+
if (!existing || !existing.busy) {
82+
yield* status.set(sessionID, { type: "idle" })
83+
return
84+
}
85+
yield* existing.cancel
86+
})
87+
88+
const ensureRunning = Effect.fn("SessionRunState.ensureRunning")(function* (
89+
sessionID: SessionID,
90+
onInterrupt: Effect.Effect<MessageV2.WithParts>,
91+
work: Effect.Effect<MessageV2.WithParts>,
92+
) {
93+
return yield* (yield* runner(sessionID, onInterrupt)).ensureRunning(work)
94+
})
95+
96+
const startShell = Effect.fn("SessionRunState.startShell")(function* (
97+
sessionID: SessionID,
98+
onInterrupt: Effect.Effect<MessageV2.WithParts>,
99+
work: Effect.Effect<MessageV2.WithParts>,
100+
) {
101+
return yield* (yield* runner(sessionID, onInterrupt)).startShell(work)
102+
})
103+
104+
return Service.of({ assertNotBusy, cancel, ensureRunning, startShell })
105+
}),
106+
)
107+
108+
export const defaultLayer = layer.pipe(Layer.provide(SessionStatus.defaultLayer))
109+
const { runPromise } = makeRuntime(Service, defaultLayer)
110+
111+
export async function assertNotBusy(sessionID: SessionID) {
112+
return runPromise((svc) => svc.assertNotBusy(sessionID))
113+
}
114+
}

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

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -85,7 +85,7 @@ export namespace SessionStatus {
8585
}),
8686
)
8787

88-
const defaultLayer = layer.pipe(Layer.provide(Bus.layer))
88+
export const defaultLayer = layer.pipe(Layer.provide(Bus.layer))
8989
const { runPromise } = makeRuntime(Service, defaultLayer)
9090

9191
export async function get(sessionID: SessionID) {

‎packages/opencode/test/server/session-actions.test.ts‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -5,6 +5,7 @@ import { Session } from "../../src/session"
55
import { ModelID, ProviderID } from "../../src/provider/schema"
66
import { MessageID, PartID, type SessionID } from "../../src/session/schema"
77
import { SessionPrompt } from "../../src/session/prompt"
8+
import { SessionRunState } from "../../src/session/run-state"
89
import { Log } from "../../src/util/log"
910
import { tmpdir } from "../fixture/fixture"
1011

@@ -64,7 +65,7 @@ describe("session action routes", () => {
6465
fn: async () => {
6566
const session = await Session.create({})
6667
const msg = await user(session.id, "hello")
67-
const busy = spyOn(SessionPrompt, "assertNotBusy").mockRejectedValue(new Session.BusyError(session.id))
68+
const busy = spyOn(SessionRunState, "assertNotBusy").mockRejectedValue(new Session.BusyError(session.id))
6869
const remove = spyOn(Session, "removeMessage").mockResolvedValue(msg.id)
6970
const app = Server.Default().app
7071

0 commit comments

Comments
 (0)