revert(session): remove threaded status context
This commit is contained in:
parent
522322980e
commit
fd14c00481
6 changed files with 23 additions and 98 deletions
|
|
@ -57,7 +57,6 @@ type Input = {
|
||||||
assistantMessage: SessionLegacy.Assistant
|
assistantMessage: SessionLegacy.Assistant
|
||||||
sessionID: SessionID
|
sessionID: SessionID
|
||||||
model: Provider.Model
|
model: Provider.Model
|
||||||
statusContext?: SessionStatus.SetContext
|
|
||||||
}
|
}
|
||||||
|
|
||||||
export interface Interface {
|
export interface Interface {
|
||||||
|
|
@ -778,7 +777,7 @@ export const layer = Layer.effect(
|
||||||
sessionID: ctx.assistantMessage.sessionID,
|
sessionID: ctx.assistantMessage.sessionID,
|
||||||
error: ctx.assistantMessage.error,
|
error: ctx.assistantMessage.error,
|
||||||
})
|
})
|
||||||
yield* status.set(ctx.sessionID, { type: "idle" }, ctx.statusContext)
|
yield* status.set(ctx.sessionID, { type: "idle" })
|
||||||
})
|
})
|
||||||
|
|
||||||
const process = Effect.fn("SessionProcessor.process")(function* (streamInput: LLM.StreamInput) {
|
const process = Effect.fn("SessionProcessor.process")(function* (streamInput: LLM.StreamInput) {
|
||||||
|
|
@ -790,7 +789,7 @@ export const layer = Layer.effect(
|
||||||
yield* Effect.gen(function* () {
|
yield* Effect.gen(function* () {
|
||||||
ctx.currentText = undefined
|
ctx.currentText = undefined
|
||||||
ctx.reasoningMap = {}
|
ctx.reasoningMap = {}
|
||||||
yield* status.set(ctx.sessionID, { type: "busy" }, ctx.statusContext)
|
yield* status.set(ctx.sessionID, { type: "busy" })
|
||||||
const stream = llm.stream(streamInput)
|
const stream = llm.stream(streamInput)
|
||||||
|
|
||||||
yield* stream.pipe(
|
yield* stream.pipe(
|
||||||
|
|
@ -830,17 +829,13 @@ export const layer = Layer.effect(
|
||||||
: Effect.void
|
: Effect.void
|
||||||
return event.pipe(
|
return event.pipe(
|
||||||
Effect.andThen(
|
Effect.andThen(
|
||||||
status.set(
|
status.set(ctx.sessionID, {
|
||||||
ctx.sessionID,
|
type: "retry",
|
||||||
{
|
attempt: info.attempt,
|
||||||
type: "retry",
|
message: info.message,
|
||||||
attempt: info.attempt,
|
action: info.action,
|
||||||
message: info.message,
|
next: info.next,
|
||||||
action: info.action,
|
}),
|
||||||
next: info.next,
|
|
||||||
},
|
|
||||||
ctx.statusContext,
|
|
||||||
),
|
|
||||||
),
|
),
|
||||||
)
|
)
|
||||||
},
|
},
|
||||||
|
|
|
||||||
|
|
@ -140,8 +140,7 @@ export const layer = Layer.effect(
|
||||||
|
|
||||||
const cancel = Effect.fn("SessionPrompt.cancel")(function* (sessionID: SessionID) {
|
const cancel = Effect.fn("SessionPrompt.cancel")(function* (sessionID: SessionID) {
|
||||||
yield* elog.info("cancel", { sessionID })
|
yield* elog.info("cancel", { sessionID })
|
||||||
const session = yield* sessions.get(sessionID).pipe(Effect.catchCause(() => Effect.succeed(undefined)))
|
yield* state.cancel(sessionID)
|
||||||
yield* state.cancel(sessionID, session?.parentID ? { parentID: session.parentID } : undefined)
|
|
||||||
})
|
})
|
||||||
|
|
||||||
const resolvePromptParts = Effect.fn("SessionPrompt.resolvePromptParts")(function* (template: string) {
|
const resolvePromptParts = Effect.fn("SessionPrompt.resolvePromptParts")(function* (template: string) {
|
||||||
|
|
@ -1252,10 +1251,9 @@ export const layer = Layer.effect(
|
||||||
let structured: unknown
|
let structured: unknown
|
||||||
let step = 0
|
let step = 0
|
||||||
const session = yield* sessions.get(sessionID).pipe(Effect.orDie)
|
const session = yield* sessions.get(sessionID).pipe(Effect.orDie)
|
||||||
const statusContext = session.parentID ? { parentID: session.parentID } : {}
|
|
||||||
|
|
||||||
while (true) {
|
while (true) {
|
||||||
yield* status.set(sessionID, { type: "busy" }, statusContext)
|
yield* status.set(sessionID, { type: "busy" })
|
||||||
yield* slog.info("loop", { step })
|
yield* slog.info("loop", { step })
|
||||||
|
|
||||||
let msgs = yield* MessageV2.filterCompactedEffect(sessionID).pipe(
|
let msgs = yield* MessageV2.filterCompactedEffect(sessionID).pipe(
|
||||||
|
|
@ -1383,7 +1381,6 @@ export const layer = Layer.effect(
|
||||||
assistantMessage: msg,
|
assistantMessage: msg,
|
||||||
sessionID,
|
sessionID,
|
||||||
model,
|
model,
|
||||||
statusContext,
|
|
||||||
})
|
})
|
||||||
.pipe(Effect.onInterrupt(() => finalizeInterruptedAssistant))
|
.pipe(Effect.onInterrupt(() => finalizeInterruptedAssistant))
|
||||||
|
|
||||||
|
|
@ -1507,13 +1504,7 @@ export const layer = Layer.effect(
|
||||||
|
|
||||||
const loop: (input: LoopInput) => Effect.Effect<SessionLegacy.WithParts> = Effect.fn("SessionPrompt.loop")(
|
const loop: (input: LoopInput) => Effect.Effect<SessionLegacy.WithParts> = Effect.fn("SessionPrompt.loop")(
|
||||||
function* (input: LoopInput) {
|
function* (input: LoopInput) {
|
||||||
const session = yield* sessions.get(input.sessionID).pipe(Effect.orDie)
|
return yield* state.ensureRunning(input.sessionID, lastAssistant(input.sessionID), runLoop(input.sessionID))
|
||||||
return yield* state.ensureRunning(
|
|
||||||
input.sessionID,
|
|
||||||
session.parentID ? { parentID: session.parentID } : {},
|
|
||||||
lastAssistant(input.sessionID),
|
|
||||||
runLoop(input.sessionID),
|
|
||||||
)
|
|
||||||
},
|
},
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
@ -1521,14 +1512,7 @@ export const layer = Layer.effect(
|
||||||
"SessionPrompt.shell",
|
"SessionPrompt.shell",
|
||||||
)(function* (input: ShellInput) {
|
)(function* (input: ShellInput) {
|
||||||
const ready = yield* Latch.make()
|
const ready = yield* Latch.make()
|
||||||
const session = yield* sessions.get(input.sessionID).pipe(Effect.orDie)
|
return yield* state.startShell(input.sessionID, lastAssistant(input.sessionID), shellImpl(input, ready), ready)
|
||||||
return yield* state.startShell(
|
|
||||||
input.sessionID,
|
|
||||||
session.parentID ? { parentID: session.parentID } : {},
|
|
||||||
lastAssistant(input.sessionID),
|
|
||||||
shellImpl(input, ready),
|
|
||||||
ready,
|
|
||||||
)
|
|
||||||
})
|
})
|
||||||
|
|
||||||
const command = Effect.fn("SessionPrompt.command")(function* (input: CommandInput) {
|
const command = Effect.fn("SessionPrompt.command")(function* (input: CommandInput) {
|
||||||
|
|
|
||||||
|
|
@ -10,16 +10,14 @@ import { SessionStatus } from "./status"
|
||||||
|
|
||||||
export interface Interface {
|
export interface Interface {
|
||||||
readonly assertNotBusy: (sessionID: SessionID) => Effect.Effect<void, Session.BusyError>
|
readonly assertNotBusy: (sessionID: SessionID) => Effect.Effect<void, Session.BusyError>
|
||||||
readonly cancel: (sessionID: SessionID, statusContext?: SessionStatus.SetContext) => Effect.Effect<void>
|
readonly cancel: (sessionID: SessionID) => Effect.Effect<void>
|
||||||
readonly ensureRunning: (
|
readonly ensureRunning: (
|
||||||
sessionID: SessionID,
|
sessionID: SessionID,
|
||||||
statusContext: SessionStatus.SetContext,
|
|
||||||
onInterrupt: Effect.Effect<SessionLegacy.WithParts>,
|
onInterrupt: Effect.Effect<SessionLegacy.WithParts>,
|
||||||
work: Effect.Effect<SessionLegacy.WithParts>,
|
work: Effect.Effect<SessionLegacy.WithParts>,
|
||||||
) => Effect.Effect<SessionLegacy.WithParts>
|
) => Effect.Effect<SessionLegacy.WithParts>
|
||||||
readonly startShell: (
|
readonly startShell: (
|
||||||
sessionID: SessionID,
|
sessionID: SessionID,
|
||||||
statusContext: SessionStatus.SetContext,
|
|
||||||
onInterrupt: Effect.Effect<SessionLegacy.WithParts>,
|
onInterrupt: Effect.Effect<SessionLegacy.WithParts>,
|
||||||
work: Effect.Effect<SessionLegacy.WithParts>,
|
work: Effect.Effect<SessionLegacy.WithParts>,
|
||||||
ready?: Latch.Latch,
|
ready?: Latch.Latch,
|
||||||
|
|
@ -53,7 +51,6 @@ export const layer = Layer.effect(
|
||||||
|
|
||||||
const runner = Effect.fn("SessionRunState.runner")(function* (
|
const runner = Effect.fn("SessionRunState.runner")(function* (
|
||||||
sessionID: SessionID,
|
sessionID: SessionID,
|
||||||
statusContext: SessionStatus.SetContext,
|
|
||||||
onInterrupt: Effect.Effect<SessionLegacy.WithParts>,
|
onInterrupt: Effect.Effect<SessionLegacy.WithParts>,
|
||||||
) {
|
) {
|
||||||
const data = yield* InstanceState.get(state)
|
const data = yield* InstanceState.get(state)
|
||||||
|
|
@ -62,9 +59,9 @@ export const layer = Layer.effect(
|
||||||
const next = Runner.make<SessionLegacy.WithParts>(data.scope, {
|
const next = Runner.make<SessionLegacy.WithParts>(data.scope, {
|
||||||
onIdle: Effect.gen(function* () {
|
onIdle: Effect.gen(function* () {
|
||||||
data.runners.delete(sessionID)
|
data.runners.delete(sessionID)
|
||||||
yield* status.set(sessionID, { type: "idle" }, statusContext)
|
yield* status.set(sessionID, { type: "idle" })
|
||||||
}),
|
}),
|
||||||
onBusy: status.set(sessionID, { type: "busy" }, statusContext),
|
onBusy: status.set(sessionID, { type: "busy" }),
|
||||||
onInterrupt,
|
onInterrupt,
|
||||||
})
|
})
|
||||||
data.runners.set(sessionID, next)
|
data.runners.set(sessionID, next)
|
||||||
|
|
@ -77,15 +74,12 @@ export const layer = Layer.effect(
|
||||||
if (existing?.busy) yield* busyError(sessionID)
|
if (existing?.busy) yield* busyError(sessionID)
|
||||||
})
|
})
|
||||||
|
|
||||||
const cancel = Effect.fn("SessionRunState.cancel")(function* (
|
const cancel = Effect.fn("SessionRunState.cancel")(function* (sessionID: SessionID) {
|
||||||
sessionID: SessionID,
|
|
||||||
statusContext?: SessionStatus.SetContext,
|
|
||||||
) {
|
|
||||||
yield* cancelBackgroundJobs(background, sessionID)
|
yield* cancelBackgroundJobs(background, sessionID)
|
||||||
const data = yield* InstanceState.get(state)
|
const data = yield* InstanceState.get(state)
|
||||||
const existing = data.runners.get(sessionID)
|
const existing = data.runners.get(sessionID)
|
||||||
if (!existing || !existing.busy) {
|
if (!existing || !existing.busy) {
|
||||||
yield* status.set(sessionID, { type: "idle" }, statusContext)
|
yield* status.set(sessionID, { type: "idle" })
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
yield* existing.cancel
|
yield* existing.cancel
|
||||||
|
|
@ -93,21 +87,19 @@ export const layer = Layer.effect(
|
||||||
|
|
||||||
const ensureRunning = Effect.fn("SessionRunState.ensureRunning")(function* (
|
const ensureRunning = Effect.fn("SessionRunState.ensureRunning")(function* (
|
||||||
sessionID: SessionID,
|
sessionID: SessionID,
|
||||||
statusContext: SessionStatus.SetContext,
|
|
||||||
onInterrupt: Effect.Effect<SessionLegacy.WithParts>,
|
onInterrupt: Effect.Effect<SessionLegacy.WithParts>,
|
||||||
work: Effect.Effect<SessionLegacy.WithParts>,
|
work: Effect.Effect<SessionLegacy.WithParts>,
|
||||||
) {
|
) {
|
||||||
return yield* (yield* runner(sessionID, statusContext, onInterrupt)).ensureRunning(work)
|
return yield* (yield* runner(sessionID, onInterrupt)).ensureRunning(work)
|
||||||
})
|
})
|
||||||
|
|
||||||
const startShell = Effect.fn("SessionRunState.startShell")(function* (
|
const startShell = Effect.fn("SessionRunState.startShell")(function* (
|
||||||
sessionID: SessionID,
|
sessionID: SessionID,
|
||||||
statusContext: SessionStatus.SetContext,
|
|
||||||
onInterrupt: Effect.Effect<SessionLegacy.WithParts>,
|
onInterrupt: Effect.Effect<SessionLegacy.WithParts>,
|
||||||
work: Effect.Effect<SessionLegacy.WithParts>,
|
work: Effect.Effect<SessionLegacy.WithParts>,
|
||||||
ready?: Latch.Latch,
|
ready?: Latch.Latch,
|
||||||
) {
|
) {
|
||||||
return yield* (yield* runner(sessionID, statusContext, onInterrupt))
|
return yield* (yield* runner(sessionID, onInterrupt))
|
||||||
.startShell(work, ready)
|
.startShell(work, ready)
|
||||||
.pipe(Effect.catchTag("RunnerBusy", () => Effect.fail(busyError(sessionID))))
|
.pipe(Effect.catchTag("RunnerBusy", () => Effect.fail(busyError(sessionID))))
|
||||||
})
|
})
|
||||||
|
|
|
||||||
|
|
@ -31,18 +31,12 @@ export const Info = Schema.Union([
|
||||||
]).annotate({ identifier: "SessionStatus" })
|
]).annotate({ identifier: "SessionStatus" })
|
||||||
export type Info = Schema.Schema.Type<typeof Info>
|
export type Info = Schema.Schema.Type<typeof Info>
|
||||||
|
|
||||||
export const SetContext = Schema.Struct({
|
|
||||||
parentID: Schema.optional(SessionID),
|
|
||||||
})
|
|
||||||
export type SetContext = Schema.Schema.Type<typeof SetContext>
|
|
||||||
|
|
||||||
export const Event = {
|
export const Event = {
|
||||||
Status: EventV2.define({
|
Status: EventV2.define({
|
||||||
type: "session.status",
|
type: "session.status",
|
||||||
schema: {
|
schema: {
|
||||||
sessionID: SessionID,
|
sessionID: SessionID,
|
||||||
status: Info,
|
status: Info,
|
||||||
parentID: Schema.optional(SessionID),
|
|
||||||
},
|
},
|
||||||
}),
|
}),
|
||||||
// deprecated
|
// deprecated
|
||||||
|
|
@ -57,7 +51,7 @@ export const Event = {
|
||||||
export interface Interface {
|
export interface Interface {
|
||||||
readonly get: (sessionID: SessionID) => Effect.Effect<Info>
|
readonly get: (sessionID: SessionID) => Effect.Effect<Info>
|
||||||
readonly list: () => Effect.Effect<Map<SessionID, Info>>
|
readonly list: () => Effect.Effect<Map<SessionID, Info>>
|
||||||
readonly set: (sessionID: SessionID, status: Info, context?: SetContext) => Effect.Effect<void>
|
readonly set: (sessionID: SessionID, status: Info) => Effect.Effect<void>
|
||||||
}
|
}
|
||||||
|
|
||||||
export class Service extends Context.Service<Service, Interface>()("@opencode/SessionStatus") {}
|
export class Service extends Context.Service<Service, Interface>()("@opencode/SessionStatus") {}
|
||||||
|
|
@ -80,13 +74,9 @@ export const layer = Layer.effect(
|
||||||
return new Map(yield* InstanceState.get(state))
|
return new Map(yield* InstanceState.get(state))
|
||||||
})
|
})
|
||||||
|
|
||||||
const set = Effect.fn("SessionStatus.set")(function* (sessionID: SessionID, status: Info, context?: SetContext) {
|
const set = Effect.fn("SessionStatus.set")(function* (sessionID: SessionID, status: Info) {
|
||||||
const data = yield* InstanceState.get(state)
|
const data = yield* InstanceState.get(state)
|
||||||
yield* events.publish(Event.Status, {
|
yield* events.publish(Event.Status, { sessionID, status })
|
||||||
sessionID,
|
|
||||||
status,
|
|
||||||
...(context?.parentID ? { parentID: context.parentID } : {}),
|
|
||||||
})
|
|
||||||
if (status.type === "idle") {
|
if (status.type === "idle") {
|
||||||
yield* events.publish(Event.Idle, { sessionID })
|
yield* events.publish(Event.Idle, { sessionID })
|
||||||
data.delete(sessionID)
|
data.delete(sessionID)
|
||||||
|
|
|
||||||
|
|
@ -867,40 +867,6 @@ it.instance(
|
||||||
3_000,
|
3_000,
|
||||||
)
|
)
|
||||||
|
|
||||||
it.instance(
|
|
||||||
"loop includes parentID on child session status events",
|
|
||||||
() =>
|
|
||||||
Effect.gen(function* () {
|
|
||||||
const { llm } = yield* useServerConfig(providerCfg)
|
|
||||||
const prompt = yield* SessionPrompt.Service
|
|
||||||
const sessions = yield* Session.Service
|
|
||||||
const events = yield* EventV2Bridge.Service
|
|
||||||
const ready = yield* Deferred.make<void>()
|
|
||||||
|
|
||||||
yield* llm.hang
|
|
||||||
|
|
||||||
const parent = yield* sessions.create({})
|
|
||||||
const child = yield* sessions.create({ parentID: parent.id })
|
|
||||||
yield* user(child.id, "hi")
|
|
||||||
|
|
||||||
const off = yield* events.listen((evt) => {
|
|
||||||
if (evt.type !== SessionStatus.Event.Status.type) return Effect.void
|
|
||||||
const data = evt.data as typeof SessionStatus.Event.Status.data.Type
|
|
||||||
if (data.sessionID !== child.id || data.status.type !== "busy") return Effect.void
|
|
||||||
if (data.parentID !== parent.id) return Effect.void
|
|
||||||
Deferred.doneUnsafe(ready, Effect.void)
|
|
||||||
return Effect.void
|
|
||||||
})
|
|
||||||
yield* Effect.addFinalizer(() => off)
|
|
||||||
|
|
||||||
const fiber = yield* prompt.loop({ sessionID: child.id }).pipe(Effect.forkChild)
|
|
||||||
yield* Deferred.await(ready).pipe(Effect.timeout("2 seconds"))
|
|
||||||
yield* prompt.cancel(child.id)
|
|
||||||
yield* Fiber.await(fiber)
|
|
||||||
}),
|
|
||||||
3_000,
|
|
||||||
)
|
|
||||||
|
|
||||||
// Cancel semantics
|
// Cancel semantics
|
||||||
|
|
||||||
it.instance(
|
it.instance(
|
||||||
|
|
|
||||||
|
|
@ -1188,7 +1188,6 @@ export type GlobalEvent = {
|
||||||
properties: {
|
properties: {
|
||||||
sessionID: string
|
sessionID: string
|
||||||
status: SessionStatus
|
status: SessionStatus
|
||||||
parentID?: string
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
| {
|
| {
|
||||||
|
|
@ -4276,7 +4275,6 @@ export type EventSessionStatus = {
|
||||||
properties: {
|
properties: {
|
||||||
sessionID: string
|
sessionID: string
|
||||||
status: SessionStatus
|
status: SessionStatus
|
||||||
parentID?: string
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue