refactor: split prompt async gate modules
Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-openagent) Co-authored-by: Sisyphus <clio-agent@sisyphuslabs.ai>
This commit is contained in:
YeonGyu-Kim
+51
-724
@@ -1,179 +1,50 @@
|
|||||||
import { log } from "./logger"
|
import { log } from "./logger"
|
||||||
|
import { DEFAULT_SESSION_IDLE_SETTLE_MS } from "./session-idle-settle"
|
||||||
import {
|
import {
|
||||||
isSyntheticOrInternalUserMessage,
|
clearPromptQueueStateForTesting,
|
||||||
type InternalInitiatorMessageLike,
|
enqueueInternalPrompt,
|
||||||
type InternalInitiatorTextPartLike,
|
getQueuedPromptBlocker,
|
||||||
} from "./internal-initiator-marker"
|
isPromptQueueDraining,
|
||||||
|
nextPromptQueueID,
|
||||||
|
releaseInFlightPromptMatchingDedupe,
|
||||||
|
schedulePromptQueueDrain,
|
||||||
|
} from "./prompt-async-gate/queue"
|
||||||
import {
|
import {
|
||||||
DEFAULT_SESSION_IDLE_SETTLE_MS,
|
clearPromptReservationsForTesting,
|
||||||
isSessionActive,
|
deletePromptReservation,
|
||||||
settleAfterSessionIdle,
|
getActiveReservation,
|
||||||
} from "./session-idle-settle"
|
getPromptReservation,
|
||||||
|
reservationSourceMatches,
|
||||||
|
} from "./prompt-async-gate/reservations"
|
||||||
|
import { dispatchAfterSessionIdle } from "./prompt-async-gate/session-idle-dispatch"
|
||||||
|
import {
|
||||||
|
DEFAULT_PROMPT_ASYNC_POST_DISPATCH_HOLD_MS,
|
||||||
|
DEFAULT_PROMPT_DISPATCH_TIMEOUT_MS,
|
||||||
|
DEFAULT_PROMPT_QUEUE_RETRY_MS,
|
||||||
|
resetPromptGateTimingForTesting,
|
||||||
|
} from "./prompt-async-gate/timing"
|
||||||
|
import type {
|
||||||
|
InternalPromptDispatchArgs,
|
||||||
|
InternalPromptDispatchResult,
|
||||||
|
PromptAsyncInput,
|
||||||
|
PromptAsyncReservationReleaseOptions,
|
||||||
|
} from "./prompt-async-gate/types"
|
||||||
|
|
||||||
export const DEFAULT_PROMPT_ASYNC_POST_DISPATCH_HOLD_MS = 2_000
|
export {
|
||||||
export const DEFAULT_PROMPT_DISPATCH_TIMEOUT_MS = 30_000
|
DEFAULT_PROMPT_ASYNC_POST_DISPATCH_HOLD_MS,
|
||||||
export const DEFAULT_PROMPT_GATE_MESSAGES_FETCH_TIMEOUT_MS = 5_000
|
DEFAULT_PROMPT_DISPATCH_TIMEOUT_MS,
|
||||||
export const DEFAULT_PROMPT_QUEUE_RETRY_MS = 250
|
DEFAULT_PROMPT_GATE_MESSAGES_FETCH_TIMEOUT_MS,
|
||||||
|
DEFAULT_PROMPT_QUEUE_RETRY_MS,
|
||||||
|
_setPromptGateMessagesFetchTimeoutMsForTesting,
|
||||||
|
} from "./prompt-async-gate/timing"
|
||||||
|
|
||||||
type PromptAsyncInput = {
|
export type {
|
||||||
path?: { id?: string }
|
InternalPromptDispatchArgs,
|
||||||
body?: unknown
|
InternalPromptDispatchMode,
|
||||||
query?: unknown
|
InternalPromptDispatchResult,
|
||||||
signal?: unknown
|
InternalPromptQueueBehavior,
|
||||||
[key: string]: unknown
|
PromptAsyncGateResult,
|
||||||
}
|
} from "./prompt-async-gate/types"
|
||||||
|
|
||||||
type PromptMessagesQuery = {
|
|
||||||
directory: string
|
|
||||||
limit?: number
|
|
||||||
}
|
|
||||||
|
|
||||||
type PromptAsyncClient<TInput> = {
|
|
||||||
session?: {
|
|
||||||
status?: () => Promise<unknown>
|
|
||||||
messages?: (input: { path: { id: string }; query: PromptMessagesQuery }) => Promise<unknown>
|
|
||||||
promptAsync?: (input: TInput) => Promise<unknown>
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
type PromptClient<TInput> = {
|
|
||||||
session?: {
|
|
||||||
status?: () => Promise<unknown>
|
|
||||||
messages?: (input: { path: { id: string }; query: PromptMessagesQuery }) => Promise<unknown>
|
|
||||||
prompt?: (input: TInput) => Promise<unknown>
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
export type InternalPromptDispatchMode = "async" | "sync"
|
|
||||||
export type InternalPromptQueueBehavior = "enqueue" | "defer"
|
|
||||||
|
|
||||||
type InternalPromptDispatchCommonArgs<TInput> = {
|
|
||||||
sessionID: string
|
|
||||||
input: TInput
|
|
||||||
source: string
|
|
||||||
dedupeKey?: string
|
|
||||||
queueBehavior?: InternalPromptQueueBehavior
|
|
||||||
queue?: boolean
|
|
||||||
queueRetryMs?: number
|
|
||||||
settleMs?: number
|
|
||||||
postDispatchHoldMs?: number
|
|
||||||
dispatchTimeoutMs?: number
|
|
||||||
checkStatus?: boolean
|
|
||||||
checkToolState?: boolean
|
|
||||||
}
|
|
||||||
|
|
||||||
export type InternalPromptDispatchArgs<TInput = PromptAsyncInput> = InternalPromptDispatchCommonArgs<TInput> & (
|
|
||||||
| { mode: "async"; client: PromptAsyncClient<TInput> }
|
|
||||||
| { mode: "sync"; client: PromptClient<TInput> }
|
|
||||||
)
|
|
||||||
|
|
||||||
type PromptAsyncReservation = {
|
|
||||||
source: string
|
|
||||||
dedupeKey: string
|
|
||||||
reservedAt: number
|
|
||||||
token: symbol
|
|
||||||
expiresAt?: number
|
|
||||||
}
|
|
||||||
|
|
||||||
declare function setTimeout(callback: () => void, delay?: number): unknown
|
|
||||||
declare function clearTimeout(timeout: unknown): void
|
|
||||||
|
|
||||||
let promptGateMessagesFetchTimeoutMsForTesting: number | undefined
|
|
||||||
|
|
||||||
export type InternalPromptDispatchResult =
|
|
||||||
| { status: "dispatched"; response: unknown }
|
|
||||||
| { status: "queued"; queuedBy: string; position: number }
|
|
||||||
| { status: "active" }
|
|
||||||
| { status: "reserved"; reservedBy: string }
|
|
||||||
| { status: "unavailable" }
|
|
||||||
| { status: "failed"; error: unknown; dispatchAttempted: boolean }
|
|
||||||
|
|
||||||
export type PromptAsyncGateResult = InternalPromptDispatchResult
|
|
||||||
|
|
||||||
type PromptAsyncReservationReleaseOptions = {
|
|
||||||
reservedBy?: string | readonly string[]
|
|
||||||
reservedByPrefix?: string | readonly string[]
|
|
||||||
}
|
|
||||||
|
|
||||||
const promptAsyncReservations = new Map<string, PromptAsyncReservation>()
|
|
||||||
const promptQueues = new Map<string, QueuedInternalPrompt[]>()
|
|
||||||
const promptQueueDraining = new Set<string>()
|
|
||||||
const promptQueueInFlight = new Map<string, QueuedInternalPrompt>()
|
|
||||||
const promptQueueTimers = new Map<string, unknown>()
|
|
||||||
let promptQueueSequence = 0
|
|
||||||
|
|
||||||
type PromptDispatchClient = {
|
|
||||||
session?: {
|
|
||||||
status?: () => Promise<unknown>
|
|
||||||
messages?: (input: { path: { id: string }; query: PromptMessagesQuery }) => Promise<unknown>
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
type QueuedInternalPrompt = {
|
|
||||||
id: number
|
|
||||||
sessionID: string
|
|
||||||
sessionName: "promptAsync" | "prompt"
|
|
||||||
client: PromptDispatchClient
|
|
||||||
input: unknown
|
|
||||||
source: string
|
|
||||||
dedupeKey: string
|
|
||||||
settleMs: number
|
|
||||||
postDispatchHoldMs: number
|
|
||||||
dispatchTimeoutMs: number
|
|
||||||
queueRetryMs: number
|
|
||||||
checkStatus: boolean
|
|
||||||
checkToolState: boolean
|
|
||||||
dispatch: (input: unknown) => Promise<unknown>
|
|
||||||
}
|
|
||||||
|
|
||||||
export function _setPromptGateMessagesFetchTimeoutMsForTesting(value: number | undefined): void {
|
|
||||||
promptGateMessagesFetchTimeoutMsForTesting = value
|
|
||||||
}
|
|
||||||
|
|
||||||
function getPromptGateMessagesFetchTimeoutMs(): number {
|
|
||||||
return promptGateMessagesFetchTimeoutMsForTesting ?? DEFAULT_PROMPT_GATE_MESSAGES_FETCH_TIMEOUT_MS
|
|
||||||
}
|
|
||||||
|
|
||||||
function pruneExpiredReservations(now = Date.now()): void {
|
|
||||||
const expiredSessionIDs: string[] = []
|
|
||||||
for (const [sessionID, reservation] of promptAsyncReservations) {
|
|
||||||
if (typeof reservation.expiresAt === "number" && reservation.expiresAt <= now) {
|
|
||||||
promptAsyncReservations.delete(sessionID)
|
|
||||||
expiredSessionIDs.push(sessionID)
|
|
||||||
log("[prompt-async-gate] expired reservation released", {
|
|
||||||
sessionID,
|
|
||||||
source: reservation.source,
|
|
||||||
})
|
|
||||||
}
|
|
||||||
}
|
|
||||||
for (const sessionID of expiredSessionIDs) {
|
|
||||||
schedulePromptQueueDrain(sessionID, 0)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
function getActiveReservation(sessionID: string): PromptAsyncReservation | undefined {
|
|
||||||
pruneExpiredReservations()
|
|
||||||
return promptAsyncReservations.get(sessionID)
|
|
||||||
}
|
|
||||||
|
|
||||||
function getPromptQueue(sessionID: string): QueuedInternalPrompt[] {
|
|
||||||
const existing = promptQueues.get(sessionID)
|
|
||||||
if (existing) {
|
|
||||||
return existing
|
|
||||||
}
|
|
||||||
|
|
||||||
const queue: QueuedInternalPrompt[] = []
|
|
||||||
promptQueues.set(sessionID, queue)
|
|
||||||
return queue
|
|
||||||
}
|
|
||||||
|
|
||||||
function setPromptQueue(sessionID: string, queue: QueuedInternalPrompt[]): void {
|
|
||||||
if (queue.length === 0) {
|
|
||||||
promptQueues.delete(sessionID)
|
|
||||||
return
|
|
||||||
}
|
|
||||||
promptQueues.set(sessionID, queue)
|
|
||||||
}
|
|
||||||
|
|
||||||
function stringifyPromptInputForDedupe(input: unknown): string {
|
function stringifyPromptInputForDedupe(input: unknown): string {
|
||||||
try {
|
try {
|
||||||
@@ -197,539 +68,6 @@ function createDefaultDedupeKey(source: string, input: unknown): string {
|
|||||||
return `${source}:${fingerprint.length}:${fingerprint.slice(0, 8192)}`
|
return `${source}:${fingerprint.length}:${fingerprint.slice(0, 8192)}`
|
||||||
}
|
}
|
||||||
|
|
||||||
function queuedResult(entry: QueuedInternalPrompt, position: number, queuedBy = entry.source): InternalPromptDispatchResult {
|
|
||||||
return {
|
|
||||||
status: "queued",
|
|
||||||
queuedBy,
|
|
||||||
position,
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
function clearPromptQueueTimer(sessionID: string): void {
|
|
||||||
const timer = promptQueueTimers.get(sessionID)
|
|
||||||
if (timer !== undefined) {
|
|
||||||
clearTimeout(timer)
|
|
||||||
promptQueueTimers.delete(sessionID)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
function schedulePromptQueueDrain(sessionID: string, delayMs: number): void {
|
|
||||||
const queue = promptQueues.get(sessionID)
|
|
||||||
if (!queue || queue.length === 0) {
|
|
||||||
clearPromptQueueTimer(sessionID)
|
|
||||||
return
|
|
||||||
}
|
|
||||||
|
|
||||||
clearPromptQueueTimer(sessionID)
|
|
||||||
const timer = setTimeout(() => {
|
|
||||||
promptQueueTimers.delete(sessionID)
|
|
||||||
void drainPromptQueue(sessionID).catch((error: unknown) => {
|
|
||||||
log("[prompt-async-gate] queued prompt drain failed", {
|
|
||||||
sessionID,
|
|
||||||
error: String(error),
|
|
||||||
})
|
|
||||||
})
|
|
||||||
}, Math.max(0, delayMs))
|
|
||||||
promptQueueTimers.set(sessionID, timer)
|
|
||||||
}
|
|
||||||
|
|
||||||
function removePromptQueueEntry(sessionID: string, entry: QueuedInternalPrompt): void {
|
|
||||||
const queue = promptQueues.get(sessionID)
|
|
||||||
if (!queue) {
|
|
||||||
return
|
|
||||||
}
|
|
||||||
const nextQueue = queue.filter((queued) => queued.id !== entry.id)
|
|
||||||
setPromptQueue(sessionID, nextQueue)
|
|
||||||
}
|
|
||||||
|
|
||||||
function getQueuedPromptBlocker(sessionID: string): string | undefined {
|
|
||||||
const inFlight = promptQueueInFlight.get(sessionID)
|
|
||||||
if (inFlight) {
|
|
||||||
return inFlight.source
|
|
||||||
}
|
|
||||||
|
|
||||||
const queue = promptQueues.get(sessionID)
|
|
||||||
return queue?.[0]?.source
|
|
||||||
}
|
|
||||||
|
|
||||||
function reservationSourceMatches(
|
|
||||||
reservationSource: string,
|
|
||||||
expectedSource: string | readonly string[],
|
|
||||||
expectedPrefix?: string | readonly string[],
|
|
||||||
): boolean {
|
|
||||||
if (typeof expectedSource === "string") {
|
|
||||||
if (reservationSource === expectedSource) {
|
|
||||||
return true
|
|
||||||
}
|
|
||||||
} else if (expectedSource.includes(reservationSource)) {
|
|
||||||
return true
|
|
||||||
}
|
|
||||||
|
|
||||||
if (expectedPrefix === undefined) {
|
|
||||||
return false
|
|
||||||
}
|
|
||||||
|
|
||||||
const prefixes = typeof expectedPrefix === "string" ? [expectedPrefix] : expectedPrefix
|
|
||||||
return prefixes
|
|
||||||
.filter((prefix) => prefix.length > 0 && prefix.endsWith(":"))
|
|
||||||
.some((prefix) => reservationSource.startsWith(prefix))
|
|
||||||
}
|
|
||||||
|
|
||||||
async function withDispatchTimeout<T>(
|
|
||||||
operation: Promise<T>,
|
|
||||||
dispatchTimeoutMs: number,
|
|
||||||
operationName: string,
|
|
||||||
): Promise<T> {
|
|
||||||
if (dispatchTimeoutMs <= 0) {
|
|
||||||
return operation
|
|
||||||
}
|
|
||||||
|
|
||||||
let timeoutID: unknown
|
|
||||||
const timeoutPromise = new Promise<never>((_, reject) => {
|
|
||||||
timeoutID = setTimeout(() => {
|
|
||||||
reject(new Error(`${operationName} timed out after ${dispatchTimeoutMs}ms`))
|
|
||||||
}, dispatchTimeoutMs)
|
|
||||||
})
|
|
||||||
|
|
||||||
try {
|
|
||||||
return await Promise.race([operation, timeoutPromise])
|
|
||||||
} finally {
|
|
||||||
if (timeoutID !== undefined) {
|
|
||||||
clearTimeout(timeoutID)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
function isRecord(value: unknown): value is Record<string, unknown> {
|
|
||||||
return typeof value === "object" && value !== null
|
|
||||||
}
|
|
||||||
|
|
||||||
function getPromptQuery(input: unknown): PromptMessagesQuery {
|
|
||||||
if (!isRecord(input)) {
|
|
||||||
return { directory: "" }
|
|
||||||
}
|
|
||||||
const query = input.query
|
|
||||||
if (!isRecord(query)) {
|
|
||||||
return { directory: "" }
|
|
||||||
}
|
|
||||||
|
|
||||||
const promptQuery: PromptMessagesQuery = { directory: "" }
|
|
||||||
if (typeof query.directory === "string") {
|
|
||||||
promptQuery.directory = query.directory
|
|
||||||
}
|
|
||||||
if (typeof query.limit === "number") {
|
|
||||||
promptQuery.limit = query.limit
|
|
||||||
}
|
|
||||||
return promptQuery
|
|
||||||
}
|
|
||||||
|
|
||||||
function getMessagesData(response: unknown): unknown[] {
|
|
||||||
if (isRecord(response) && Array.isArray(response.data)) {
|
|
||||||
return response.data
|
|
||||||
}
|
|
||||||
return Array.isArray(response) ? response : []
|
|
||||||
}
|
|
||||||
|
|
||||||
function messageRole(message: unknown): string | undefined {
|
|
||||||
if (!isRecord(message)) {
|
|
||||||
return undefined
|
|
||||||
}
|
|
||||||
const info = message.info
|
|
||||||
if (isRecord(info) && typeof info.role === "string") {
|
|
||||||
return info.role
|
|
||||||
}
|
|
||||||
return typeof message.role === "string" ? message.role : undefined
|
|
||||||
}
|
|
||||||
|
|
||||||
function messageFinish(message: unknown): string | true | undefined {
|
|
||||||
if (!isRecord(message)) {
|
|
||||||
return undefined
|
|
||||||
}
|
|
||||||
const info = message.info
|
|
||||||
if (isRecord(info)) {
|
|
||||||
if (info.finish === true) {
|
|
||||||
return true
|
|
||||||
}
|
|
||||||
if (typeof info.finish === "string" && info.finish.length > 0) {
|
|
||||||
return info.finish
|
|
||||||
}
|
|
||||||
}
|
|
||||||
if (message.finish === true) {
|
|
||||||
return true
|
|
||||||
}
|
|
||||||
return typeof message.finish === "string" && message.finish.length > 0 ? message.finish : undefined
|
|
||||||
}
|
|
||||||
|
|
||||||
function messageCompleted(message: unknown): boolean {
|
|
||||||
if (!isRecord(message)) {
|
|
||||||
return false
|
|
||||||
}
|
|
||||||
const info = message.info
|
|
||||||
const time = isRecord(info) && isRecord(info.time) ? info.time : undefined
|
|
||||||
const completed = time?.completed
|
|
||||||
if (typeof completed === "number" && Number.isFinite(completed)) {
|
|
||||||
return true
|
|
||||||
}
|
|
||||||
return typeof completed === "string" && completed.length > 0
|
|
||||||
}
|
|
||||||
|
|
||||||
function toInternalInitiatorTextPartLike(part: unknown): InternalInitiatorTextPartLike {
|
|
||||||
const result: InternalInitiatorTextPartLike = {}
|
|
||||||
if (!isRecord(part)) {
|
|
||||||
return result
|
|
||||||
}
|
|
||||||
|
|
||||||
if (typeof part.type === "string") {
|
|
||||||
result.type = part.type
|
|
||||||
}
|
|
||||||
if (typeof part.text === "string") {
|
|
||||||
result.text = part.text
|
|
||||||
}
|
|
||||||
if (typeof part.synthetic === "boolean") {
|
|
||||||
result.synthetic = part.synthetic
|
|
||||||
}
|
|
||||||
return result
|
|
||||||
}
|
|
||||||
|
|
||||||
function toInternalInitiatorMessageLike(message: unknown): InternalInitiatorMessageLike | undefined {
|
|
||||||
if (!isRecord(message)) {
|
|
||||||
return undefined
|
|
||||||
}
|
|
||||||
|
|
||||||
const result: InternalInitiatorMessageLike = {}
|
|
||||||
const info = message.info
|
|
||||||
if (isRecord(info) && typeof info.role === "string") {
|
|
||||||
result.info = { role: info.role }
|
|
||||||
}
|
|
||||||
if (typeof message.role === "string") {
|
|
||||||
result.role = message.role
|
|
||||||
}
|
|
||||||
if (Array.isArray(message.parts)) {
|
|
||||||
result.parts = message.parts.map(toInternalInitiatorTextPartLike)
|
|
||||||
}
|
|
||||||
return result
|
|
||||||
}
|
|
||||||
|
|
||||||
function messageIsSyntheticOrInternalUser(message: unknown): boolean {
|
|
||||||
const initiatorMessage = toInternalInitiatorMessageLike(message)
|
|
||||||
return initiatorMessage !== undefined && isSyntheticOrInternalUserMessage(initiatorMessage)
|
|
||||||
}
|
|
||||||
|
|
||||||
function partIsWaitingOnTool(part: unknown): boolean {
|
|
||||||
if (!isRecord(part)) {
|
|
||||||
return false
|
|
||||||
}
|
|
||||||
if (
|
|
||||||
part.type !== "tool"
|
|
||||||
&& part.type !== "tool_use"
|
|
||||||
&& part.type !== "tool-call"
|
|
||||||
&& part.type !== "tool-invocation"
|
|
||||||
) {
|
|
||||||
return false
|
|
||||||
}
|
|
||||||
|
|
||||||
const state = part.state
|
|
||||||
if (!isRecord(state)) {
|
|
||||||
return false
|
|
||||||
}
|
|
||||||
return state.status === "pending" || state.status === "running"
|
|
||||||
}
|
|
||||||
|
|
||||||
function latestAssistantTurnBlocksInternalPrompt(messages: unknown[]): boolean {
|
|
||||||
for (let index = messages.length - 1; index >= 0; index--) {
|
|
||||||
const message = messages[index]
|
|
||||||
const role = messageRole(message)
|
|
||||||
if (role === "assistant") {
|
|
||||||
if (messageCompleted(message)) {
|
|
||||||
return false
|
|
||||||
}
|
|
||||||
const finish = messageFinish(message)
|
|
||||||
if (finish === true) {
|
|
||||||
return false
|
|
||||||
}
|
|
||||||
if (finish === undefined || finish === "unknown") {
|
|
||||||
return true
|
|
||||||
}
|
|
||||||
if (!isRecord(message) || !Array.isArray(message.parts)) {
|
|
||||||
return finish === "tool-calls"
|
|
||||||
}
|
|
||||||
return finish === "tool-calls" || message.parts.some(partIsWaitingOnTool)
|
|
||||||
}
|
|
||||||
if (role === "user") {
|
|
||||||
if (messageIsSyntheticOrInternalUser(message)) {
|
|
||||||
continue
|
|
||||||
}
|
|
||||||
return false
|
|
||||||
}
|
|
||||||
}
|
|
||||||
return false
|
|
||||||
}
|
|
||||||
|
|
||||||
async function sessionLatestAssistantBlocksInternalPrompt<TInput>(args: {
|
|
||||||
client: { session?: { messages?: (input: { path: { id: string }; query: PromptMessagesQuery }) => Promise<unknown> } }
|
|
||||||
sessionID: string
|
|
||||||
input: TInput
|
|
||||||
sessionName: "promptAsync" | "prompt"
|
|
||||||
source: string
|
|
||||||
timeoutMs: number
|
|
||||||
}): Promise<boolean> {
|
|
||||||
const session = args.client.session
|
|
||||||
if (typeof session?.messages !== "function") {
|
|
||||||
return false
|
|
||||||
}
|
|
||||||
const messages = session.messages.bind(session)
|
|
||||||
|
|
||||||
try {
|
|
||||||
const response = await withDispatchTimeout(
|
|
||||||
messages({
|
|
||||||
path: { id: args.sessionID },
|
|
||||||
query: getPromptQuery(args.input),
|
|
||||||
}),
|
|
||||||
args.timeoutMs,
|
|
||||||
`[prompt-async-gate] ${args.sessionName} session.messages`,
|
|
||||||
)
|
|
||||||
return latestAssistantTurnBlocksInternalPrompt(getMessagesData(response))
|
|
||||||
} catch (error) {
|
|
||||||
log("[prompt-async-gate] latest assistant prompt-block check failed", {
|
|
||||||
sessionID: args.sessionID,
|
|
||||||
source: args.source,
|
|
||||||
error: String(error),
|
|
||||||
})
|
|
||||||
return false
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
async function dispatchAfterSessionIdle<TInput>(args: {
|
|
||||||
sessionName: "promptAsync" | "prompt"
|
|
||||||
client: {
|
|
||||||
session?: {
|
|
||||||
status?: () => Promise<unknown>
|
|
||||||
messages?: (input: { path: { id: string }; query: PromptMessagesQuery }) => Promise<unknown>
|
|
||||||
}
|
|
||||||
}
|
|
||||||
sessionID: string
|
|
||||||
input: TInput
|
|
||||||
source: string
|
|
||||||
dedupeKey: string
|
|
||||||
settleMs: number
|
|
||||||
postDispatchHoldMs: number
|
|
||||||
dispatchTimeoutMs: number
|
|
||||||
checkStatus: boolean
|
|
||||||
checkToolState: boolean
|
|
||||||
dispatch: (input: TInput) => Promise<unknown>
|
|
||||||
}): Promise<InternalPromptDispatchResult> {
|
|
||||||
const {
|
|
||||||
sessionName,
|
|
||||||
client,
|
|
||||||
sessionID,
|
|
||||||
input,
|
|
||||||
source,
|
|
||||||
dedupeKey,
|
|
||||||
settleMs,
|
|
||||||
postDispatchHoldMs,
|
|
||||||
dispatchTimeoutMs,
|
|
||||||
checkStatus,
|
|
||||||
checkToolState,
|
|
||||||
dispatch,
|
|
||||||
} = args
|
|
||||||
|
|
||||||
const existing = getActiveReservation(sessionID)
|
|
||||||
if (existing) {
|
|
||||||
log(`[prompt-async-gate] ${sessionName} skipped because session is reserved`, {
|
|
||||||
sessionID,
|
|
||||||
source,
|
|
||||||
reservedBy: existing.source,
|
|
||||||
reservedAgeMs: Date.now() - existing.reservedAt,
|
|
||||||
})
|
|
||||||
return { status: "reserved", reservedBy: existing.source }
|
|
||||||
}
|
|
||||||
|
|
||||||
const reservation: PromptAsyncReservation = {
|
|
||||||
source,
|
|
||||||
dedupeKey,
|
|
||||||
reservedAt: Date.now(),
|
|
||||||
token: Symbol(source),
|
|
||||||
}
|
|
||||||
promptAsyncReservations.set(sessionID, reservation)
|
|
||||||
let dispatchAttempted = false
|
|
||||||
|
|
||||||
try {
|
|
||||||
const canReadStatus = checkStatus && typeof client.session?.status === "function"
|
|
||||||
if (settleMs > 0) {
|
|
||||||
await settleAfterSessionIdle(settleMs)
|
|
||||||
}
|
|
||||||
|
|
||||||
let sessionActive = false
|
|
||||||
if (canReadStatus) {
|
|
||||||
try {
|
|
||||||
sessionActive = await withDispatchTimeout(
|
|
||||||
isSessionActive(client, sessionID),
|
|
||||||
Math.min(dispatchTimeoutMs, 5000),
|
|
||||||
`[prompt-async-gate] ${sessionName} isSessionActive`,
|
|
||||||
)
|
|
||||||
} catch {
|
|
||||||
sessionActive = false
|
|
||||||
}
|
|
||||||
}
|
|
||||||
if (sessionActive) {
|
|
||||||
log(`[prompt-async-gate] ${sessionName} skipped because session is active`, { sessionID, source })
|
|
||||||
return { status: "active" }
|
|
||||||
}
|
|
||||||
|
|
||||||
if (
|
|
||||||
checkToolState
|
|
||||||
&& typeof client.session?.messages === "function"
|
|
||||||
&& await sessionLatestAssistantBlocksInternalPrompt({
|
|
||||||
client,
|
|
||||||
sessionID,
|
|
||||||
input,
|
|
||||||
sessionName,
|
|
||||||
source,
|
|
||||||
timeoutMs: Math.min(dispatchTimeoutMs, getPromptGateMessagesFetchTimeoutMs()),
|
|
||||||
})
|
|
||||||
) {
|
|
||||||
log(`[prompt-async-gate] ${sessionName} skipped because latest assistant is still active`, {
|
|
||||||
sessionID,
|
|
||||||
source,
|
|
||||||
})
|
|
||||||
return { status: "active" }
|
|
||||||
}
|
|
||||||
|
|
||||||
log(`[prompt-async-gate] ${sessionName} dispatching`, { sessionID, source })
|
|
||||||
dispatchAttempted = true
|
|
||||||
const response = await withDispatchTimeout(
|
|
||||||
dispatch(input),
|
|
||||||
dispatchTimeoutMs,
|
|
||||||
`[prompt-async-gate] ${sessionName} dispatch`,
|
|
||||||
)
|
|
||||||
log(`[prompt-async-gate] ${sessionName} dispatched`, { sessionID, source })
|
|
||||||
return { status: "dispatched", response }
|
|
||||||
} catch (error) {
|
|
||||||
log(`[prompt-async-gate] ${sessionName} failed`, { sessionID, source, error: String(error) })
|
|
||||||
return { status: "failed", error, dispatchAttempted }
|
|
||||||
} finally {
|
|
||||||
const current = promptAsyncReservations.get(sessionID)
|
|
||||||
if (current?.token === reservation.token) {
|
|
||||||
if (dispatchAttempted && postDispatchHoldMs > 0) {
|
|
||||||
reservation.expiresAt = Date.now() + postDispatchHoldMs
|
|
||||||
} else {
|
|
||||||
promptAsyncReservations.delete(sessionID)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
async function drainPromptQueue(sessionID: string, awaitedEntry?: QueuedInternalPrompt): Promise<InternalPromptDispatchResult | undefined> {
|
|
||||||
if (promptQueueDraining.has(sessionID)) {
|
|
||||||
return awaitedEntry ? queuedResult(awaitedEntry, 1) : undefined
|
|
||||||
}
|
|
||||||
|
|
||||||
promptQueueDraining.add(sessionID)
|
|
||||||
clearPromptQueueTimer(sessionID)
|
|
||||||
|
|
||||||
let awaitedResult: InternalPromptDispatchResult | undefined
|
|
||||||
try {
|
|
||||||
while (true) {
|
|
||||||
const queue = promptQueues.get(sessionID)
|
|
||||||
const entry = queue?.[0]
|
|
||||||
if (!entry) {
|
|
||||||
break
|
|
||||||
}
|
|
||||||
|
|
||||||
promptQueueInFlight.set(sessionID, entry)
|
|
||||||
const result = await dispatchAfterSessionIdle({
|
|
||||||
sessionName: entry.sessionName,
|
|
||||||
client: entry.client,
|
|
||||||
sessionID: entry.sessionID,
|
|
||||||
input: entry.input,
|
|
||||||
source: entry.source,
|
|
||||||
dedupeKey: entry.dedupeKey,
|
|
||||||
settleMs: entry.settleMs,
|
|
||||||
postDispatchHoldMs: entry.postDispatchHoldMs,
|
|
||||||
dispatchTimeoutMs: entry.dispatchTimeoutMs,
|
|
||||||
checkStatus: entry.checkStatus,
|
|
||||||
checkToolState: entry.checkToolState,
|
|
||||||
dispatch: entry.dispatch,
|
|
||||||
})
|
|
||||||
if (promptQueueInFlight.get(sessionID)?.id === entry.id) {
|
|
||||||
promptQueueInFlight.delete(sessionID)
|
|
||||||
}
|
|
||||||
|
|
||||||
if (result.status === "active" || result.status === "reserved") {
|
|
||||||
const queued = queuedResult(
|
|
||||||
entry,
|
|
||||||
1,
|
|
||||||
result.status === "reserved" ? result.reservedBy : entry.source,
|
|
||||||
)
|
|
||||||
if (awaitedEntry?.id === entry.id) {
|
|
||||||
awaitedResult = queued
|
|
||||||
}
|
|
||||||
schedulePromptQueueDrain(sessionID, entry.queueRetryMs)
|
|
||||||
break
|
|
||||||
}
|
|
||||||
|
|
||||||
removePromptQueueEntry(sessionID, entry)
|
|
||||||
if (awaitedEntry?.id === entry.id) {
|
|
||||||
awaitedResult = result
|
|
||||||
}
|
|
||||||
|
|
||||||
const remainingQueue = promptQueues.get(sessionID)
|
|
||||||
if (!remainingQueue || remainingQueue.length === 0) {
|
|
||||||
break
|
|
||||||
}
|
|
||||||
|
|
||||||
schedulePromptQueueDrain(sessionID, entry.postDispatchHoldMs)
|
|
||||||
break
|
|
||||||
}
|
|
||||||
} finally {
|
|
||||||
promptQueueDraining.delete(sessionID)
|
|
||||||
}
|
|
||||||
|
|
||||||
return awaitedResult
|
|
||||||
}
|
|
||||||
|
|
||||||
async function enqueueInternalPrompt(entry: QueuedInternalPrompt): Promise<InternalPromptDispatchResult> {
|
|
||||||
const activeReservation = getActiveReservation(entry.sessionID)
|
|
||||||
if (activeReservation?.dedupeKey === entry.dedupeKey) {
|
|
||||||
log("[prompt-async-gate] queued prompt coalesced with recent dispatch", {
|
|
||||||
sessionID: entry.sessionID,
|
|
||||||
source: entry.source,
|
|
||||||
queuedBy: activeReservation.source,
|
|
||||||
})
|
|
||||||
return queuedResult(entry, 0, activeReservation.source)
|
|
||||||
}
|
|
||||||
|
|
||||||
const queue = getPromptQueue(entry.sessionID)
|
|
||||||
const existingIndex = queue.findIndex((queued) => queued.dedupeKey === entry.dedupeKey)
|
|
||||||
if (existingIndex >= 0) {
|
|
||||||
const existing = queue[existingIndex]
|
|
||||||
if (existing) {
|
|
||||||
log("[prompt-async-gate] queued prompt coalesced with pending dispatch", {
|
|
||||||
sessionID: entry.sessionID,
|
|
||||||
source: entry.source,
|
|
||||||
queuedBy: existing.source,
|
|
||||||
position: existingIndex + 1,
|
|
||||||
})
|
|
||||||
return queuedResult(existing, existingIndex + 1)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
queue.push(entry)
|
|
||||||
log("[prompt-async-gate] queued prompt accepted", {
|
|
||||||
sessionID: entry.sessionID,
|
|
||||||
source: entry.source,
|
|
||||||
position: queue.length,
|
|
||||||
})
|
|
||||||
|
|
||||||
if (queue.length > 1 || promptQueueDraining.has(entry.sessionID)) {
|
|
||||||
schedulePromptQueueDrain(entry.sessionID, 0)
|
|
||||||
return queuedResult(entry, queue.length)
|
|
||||||
}
|
|
||||||
|
|
||||||
const result = await drainPromptQueue(entry.sessionID, entry)
|
|
||||||
return result ?? queuedResult(entry, 1)
|
|
||||||
}
|
|
||||||
|
|
||||||
export async function dispatchInternalPrompt<TInput = PromptAsyncInput>(
|
export async function dispatchInternalPrompt<TInput = PromptAsyncInput>(
|
||||||
args: InternalPromptDispatchArgs<TInput>,
|
args: InternalPromptDispatchArgs<TInput>,
|
||||||
): Promise<InternalPromptDispatchResult> {
|
): Promise<InternalPromptDispatchResult> {
|
||||||
@@ -777,7 +115,7 @@ export async function dispatchInternalPrompt<TInput = PromptAsyncInput>(
|
|||||||
}
|
}
|
||||||
|
|
||||||
const queuedBy = getQueuedPromptBlocker(sessionID)
|
const queuedBy = getQueuedPromptBlocker(sessionID)
|
||||||
if (queuedBy !== undefined || promptQueueDraining.has(sessionID)) {
|
if (queuedBy !== undefined || isPromptQueueDraining(sessionID)) {
|
||||||
return { status: "reserved", reservedBy: queuedBy ?? source }
|
return { status: "reserved", reservedBy: queuedBy ?? source }
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -799,7 +137,7 @@ export async function dispatchInternalPrompt<TInput = PromptAsyncInput>(
|
|||||||
|
|
||||||
if (args.queue !== false) {
|
if (args.queue !== false) {
|
||||||
return enqueueInternalPrompt({
|
return enqueueInternalPrompt({
|
||||||
id: promptQueueSequence += 1,
|
id: nextPromptQueueID(),
|
||||||
sessionID,
|
sessionID,
|
||||||
sessionName,
|
sessionName,
|
||||||
client,
|
client,
|
||||||
@@ -812,7 +150,7 @@ export async function dispatchInternalPrompt<TInput = PromptAsyncInput>(
|
|||||||
queueRetryMs,
|
queueRetryMs,
|
||||||
checkStatus: args.checkStatus !== false,
|
checkStatus: args.checkStatus !== false,
|
||||||
checkToolState: args.checkToolState !== false,
|
checkToolState: args.checkToolState !== false,
|
||||||
dispatch: dispatch as (dispatchInput: unknown) => Promise<unknown>,
|
dispatch: async (_dispatchInput: unknown) => dispatch(input),
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -833,15 +171,9 @@ export async function dispatchInternalPrompt<TInput = PromptAsyncInput>(
|
|||||||
}
|
}
|
||||||
|
|
||||||
export function releaseAllPromptAsyncReservationsForTesting(): void {
|
export function releaseAllPromptAsyncReservationsForTesting(): void {
|
||||||
promptAsyncReservations.clear()
|
clearPromptReservationsForTesting()
|
||||||
promptQueues.clear()
|
clearPromptQueueStateForTesting()
|
||||||
promptQueueDraining.clear()
|
resetPromptGateTimingForTesting()
|
||||||
promptQueueInFlight.clear()
|
|
||||||
for (const timer of promptQueueTimers.values()) {
|
|
||||||
clearTimeout(timer)
|
|
||||||
}
|
|
||||||
promptQueueTimers.clear()
|
|
||||||
promptGateMessagesFetchTimeoutMsForTesting = undefined
|
|
||||||
}
|
}
|
||||||
|
|
||||||
export function isInternalPromptDispatchAccepted(result: InternalPromptDispatchResult): boolean {
|
export function isInternalPromptDispatchAccepted(result: InternalPromptDispatchResult): boolean {
|
||||||
@@ -853,7 +185,7 @@ export function releasePromptAsyncReservation(
|
|||||||
source: string,
|
source: string,
|
||||||
options?: PromptAsyncReservationReleaseOptions,
|
options?: PromptAsyncReservationReleaseOptions,
|
||||||
): boolean {
|
): boolean {
|
||||||
const existing = promptAsyncReservations.get(sessionID)
|
const existing = getPromptReservation(sessionID)
|
||||||
if (!existing) {
|
if (!existing) {
|
||||||
return false
|
return false
|
||||||
}
|
}
|
||||||
@@ -868,13 +200,8 @@ export function releasePromptAsyncReservation(
|
|||||||
return false
|
return false
|
||||||
}
|
}
|
||||||
|
|
||||||
promptAsyncReservations.delete(sessionID)
|
deletePromptReservation(sessionID)
|
||||||
const inFlight = promptQueueInFlight.get(sessionID)
|
releaseInFlightPromptMatchingDedupe(sessionID, existing.dedupeKey)
|
||||||
if (inFlight?.dedupeKey === existing.dedupeKey) {
|
|
||||||
removePromptQueueEntry(sessionID, inFlight)
|
|
||||||
promptQueueInFlight.delete(sessionID)
|
|
||||||
promptQueueDraining.delete(sessionID)
|
|
||||||
}
|
|
||||||
schedulePromptQueueDrain(sessionID, 0)
|
schedulePromptQueueDrain(sessionID, 0)
|
||||||
log("[prompt-async-gate] promptAsync reservation released", {
|
log("[prompt-async-gate] promptAsync reservation released", {
|
||||||
sessionID,
|
sessionID,
|
||||||
|
|||||||
@@ -0,0 +1,206 @@
|
|||||||
|
import { log } from "../logger"
|
||||||
|
import {
|
||||||
|
isSyntheticOrInternalUserMessage,
|
||||||
|
type InternalInitiatorMessageLike,
|
||||||
|
type InternalInitiatorTextPartLike,
|
||||||
|
} from "../internal-initiator-marker"
|
||||||
|
import { isRecord } from "../record-type-guard"
|
||||||
|
import { withDispatchTimeout } from "./timing"
|
||||||
|
import type { PromptDispatchClient, PromptMessagesQuery, PromptSessionName } from "./types"
|
||||||
|
|
||||||
|
function getPromptQuery(input: unknown): PromptMessagesQuery {
|
||||||
|
if (!isRecord(input)) {
|
||||||
|
return { directory: "" }
|
||||||
|
}
|
||||||
|
const query = input.query
|
||||||
|
if (!isRecord(query)) {
|
||||||
|
return { directory: "" }
|
||||||
|
}
|
||||||
|
|
||||||
|
const promptQuery: PromptMessagesQuery = { directory: "" }
|
||||||
|
if (typeof query.directory === "string") {
|
||||||
|
return typeof query.limit === "number"
|
||||||
|
? { directory: query.directory, limit: query.limit }
|
||||||
|
: { directory: query.directory }
|
||||||
|
}
|
||||||
|
if (typeof query.limit === "number") {
|
||||||
|
return { ...promptQuery, limit: query.limit }
|
||||||
|
}
|
||||||
|
return promptQuery
|
||||||
|
}
|
||||||
|
|
||||||
|
function getMessagesData(response: unknown): unknown[] {
|
||||||
|
if (isRecord(response) && Array.isArray(response.data)) {
|
||||||
|
return response.data
|
||||||
|
}
|
||||||
|
return Array.isArray(response) ? response : []
|
||||||
|
}
|
||||||
|
|
||||||
|
function messageRole(message: unknown): string | undefined {
|
||||||
|
if (!isRecord(message)) {
|
||||||
|
return undefined
|
||||||
|
}
|
||||||
|
const info = message.info
|
||||||
|
if (isRecord(info) && typeof info.role === "string") {
|
||||||
|
return info.role
|
||||||
|
}
|
||||||
|
return typeof message.role === "string" ? message.role : undefined
|
||||||
|
}
|
||||||
|
|
||||||
|
function messageFinish(message: unknown): string | true | undefined {
|
||||||
|
if (!isRecord(message)) {
|
||||||
|
return undefined
|
||||||
|
}
|
||||||
|
const info = message.info
|
||||||
|
if (isRecord(info)) {
|
||||||
|
if (info.finish === true) {
|
||||||
|
return true
|
||||||
|
}
|
||||||
|
if (typeof info.finish === "string" && info.finish.length > 0) {
|
||||||
|
return info.finish
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if (message.finish === true) {
|
||||||
|
return true
|
||||||
|
}
|
||||||
|
return typeof message.finish === "string" && message.finish.length > 0 ? message.finish : undefined
|
||||||
|
}
|
||||||
|
|
||||||
|
function messageCompleted(message: unknown): boolean {
|
||||||
|
if (!isRecord(message)) {
|
||||||
|
return false
|
||||||
|
}
|
||||||
|
const info = message.info
|
||||||
|
const time = isRecord(info) && isRecord(info.time) ? info.time : undefined
|
||||||
|
const completed = time?.completed
|
||||||
|
if (typeof completed === "number" && Number.isFinite(completed)) {
|
||||||
|
return true
|
||||||
|
}
|
||||||
|
return typeof completed === "string" && completed.length > 0
|
||||||
|
}
|
||||||
|
|
||||||
|
function toInternalInitiatorTextPartLike(part: unknown): InternalInitiatorTextPartLike {
|
||||||
|
const result: InternalInitiatorTextPartLike = {}
|
||||||
|
if (!isRecord(part)) {
|
||||||
|
return result
|
||||||
|
}
|
||||||
|
|
||||||
|
if (typeof part.type === "string") {
|
||||||
|
result.type = part.type
|
||||||
|
}
|
||||||
|
if (typeof part.text === "string") {
|
||||||
|
result.text = part.text
|
||||||
|
}
|
||||||
|
if (typeof part.synthetic === "boolean") {
|
||||||
|
result.synthetic = part.synthetic
|
||||||
|
}
|
||||||
|
return result
|
||||||
|
}
|
||||||
|
|
||||||
|
function toInternalInitiatorMessageLike(message: unknown): InternalInitiatorMessageLike | undefined {
|
||||||
|
if (!isRecord(message)) {
|
||||||
|
return undefined
|
||||||
|
}
|
||||||
|
|
||||||
|
const result: InternalInitiatorMessageLike = {}
|
||||||
|
const info = message.info
|
||||||
|
if (isRecord(info) && typeof info.role === "string") {
|
||||||
|
result.info = { role: info.role }
|
||||||
|
}
|
||||||
|
if (typeof message.role === "string") {
|
||||||
|
result.role = message.role
|
||||||
|
}
|
||||||
|
if (Array.isArray(message.parts)) {
|
||||||
|
result.parts = message.parts.map(toInternalInitiatorTextPartLike)
|
||||||
|
}
|
||||||
|
return result
|
||||||
|
}
|
||||||
|
|
||||||
|
function messageIsSyntheticOrInternalUser(message: unknown): boolean {
|
||||||
|
const initiatorMessage = toInternalInitiatorMessageLike(message)
|
||||||
|
return initiatorMessage !== undefined && isSyntheticOrInternalUserMessage(initiatorMessage)
|
||||||
|
}
|
||||||
|
|
||||||
|
function partIsWaitingOnTool(part: unknown): boolean {
|
||||||
|
if (!isRecord(part)) {
|
||||||
|
return false
|
||||||
|
}
|
||||||
|
if (
|
||||||
|
part.type !== "tool"
|
||||||
|
&& part.type !== "tool_use"
|
||||||
|
&& part.type !== "tool-call"
|
||||||
|
&& part.type !== "tool-invocation"
|
||||||
|
) {
|
||||||
|
return false
|
||||||
|
}
|
||||||
|
|
||||||
|
const state = part.state
|
||||||
|
if (!isRecord(state)) {
|
||||||
|
return false
|
||||||
|
}
|
||||||
|
return state.status === "pending" || state.status === "running"
|
||||||
|
}
|
||||||
|
|
||||||
|
function latestAssistantTurnBlocksInternalPrompt(messages: unknown[]): boolean {
|
||||||
|
for (let index = messages.length - 1; index >= 0; index--) {
|
||||||
|
const message = messages[index]
|
||||||
|
const role = messageRole(message)
|
||||||
|
if (role === "assistant") {
|
||||||
|
if (messageCompleted(message)) {
|
||||||
|
return false
|
||||||
|
}
|
||||||
|
const finish = messageFinish(message)
|
||||||
|
if (finish === true) {
|
||||||
|
return false
|
||||||
|
}
|
||||||
|
if (finish === undefined || finish === "unknown") {
|
||||||
|
return true
|
||||||
|
}
|
||||||
|
if (!isRecord(message) || !Array.isArray(message.parts)) {
|
||||||
|
return finish === "tool-calls"
|
||||||
|
}
|
||||||
|
return finish === "tool-calls" || message.parts.some(partIsWaitingOnTool)
|
||||||
|
}
|
||||||
|
if (role === "user") {
|
||||||
|
if (messageIsSyntheticOrInternalUser(message)) {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
return false
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return false
|
||||||
|
}
|
||||||
|
|
||||||
|
export async function sessionLatestAssistantBlocksInternalPrompt<TInput>(args: {
|
||||||
|
readonly client: PromptDispatchClient
|
||||||
|
readonly sessionID: string
|
||||||
|
readonly input: TInput
|
||||||
|
readonly sessionName: PromptSessionName
|
||||||
|
readonly source: string
|
||||||
|
readonly timeoutMs: number
|
||||||
|
}): Promise<boolean> {
|
||||||
|
const session = args.client.session
|
||||||
|
if (typeof session?.messages !== "function") {
|
||||||
|
return false
|
||||||
|
}
|
||||||
|
const messages = session.messages.bind(session)
|
||||||
|
|
||||||
|
try {
|
||||||
|
const response = await withDispatchTimeout(
|
||||||
|
messages({
|
||||||
|
path: { id: args.sessionID },
|
||||||
|
query: getPromptQuery(args.input),
|
||||||
|
}),
|
||||||
|
args.timeoutMs,
|
||||||
|
`[prompt-async-gate] ${args.sessionName} session.messages`,
|
||||||
|
)
|
||||||
|
return latestAssistantTurnBlocksInternalPrompt(getMessagesData(response))
|
||||||
|
} catch (error) {
|
||||||
|
log("[prompt-async-gate] latest assistant prompt-block check failed", {
|
||||||
|
sessionID: args.sessionID,
|
||||||
|
source: args.source,
|
||||||
|
error: String(error),
|
||||||
|
})
|
||||||
|
return false
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -0,0 +1,233 @@
|
|||||||
|
import { log } from "../logger"
|
||||||
|
import { dispatchAfterSessionIdle } from "./session-idle-dispatch"
|
||||||
|
import {
|
||||||
|
getActiveReservation,
|
||||||
|
setExpiredReservationHandler,
|
||||||
|
} from "./reservations"
|
||||||
|
import type { InternalPromptDispatchResult, QueuedInternalPrompt } from "./types"
|
||||||
|
|
||||||
|
declare function setTimeout(callback: () => void, delay?: number): unknown
|
||||||
|
declare function clearTimeout(timeout: unknown): void
|
||||||
|
|
||||||
|
const promptQueues = new Map<string, QueuedInternalPrompt[]>()
|
||||||
|
const promptQueueDraining = new Set<string>()
|
||||||
|
const promptQueueInFlight = new Map<string, QueuedInternalPrompt>()
|
||||||
|
const promptQueueTimers = new Map<string, unknown>()
|
||||||
|
let promptQueueSequence = 0
|
||||||
|
|
||||||
|
setExpiredReservationHandler((sessionID) => {
|
||||||
|
schedulePromptQueueDrain(sessionID, 0)
|
||||||
|
})
|
||||||
|
|
||||||
|
function getPromptQueue(sessionID: string): QueuedInternalPrompt[] {
|
||||||
|
const existing = promptQueues.get(sessionID)
|
||||||
|
if (existing) {
|
||||||
|
return existing
|
||||||
|
}
|
||||||
|
|
||||||
|
const queue: QueuedInternalPrompt[] = []
|
||||||
|
promptQueues.set(sessionID, queue)
|
||||||
|
return queue
|
||||||
|
}
|
||||||
|
|
||||||
|
function setPromptQueue(sessionID: string, queue: QueuedInternalPrompt[]): void {
|
||||||
|
if (queue.length === 0) {
|
||||||
|
promptQueues.delete(sessionID)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
promptQueues.set(sessionID, queue)
|
||||||
|
}
|
||||||
|
|
||||||
|
function queuedResult(entry: QueuedInternalPrompt, position: number, queuedBy = entry.source): InternalPromptDispatchResult {
|
||||||
|
return {
|
||||||
|
status: "queued",
|
||||||
|
queuedBy,
|
||||||
|
position,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
function clearPromptQueueTimer(sessionID: string): void {
|
||||||
|
const timer = promptQueueTimers.get(sessionID)
|
||||||
|
if (timer !== undefined) {
|
||||||
|
clearTimeout(timer)
|
||||||
|
promptQueueTimers.delete(sessionID)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
export function schedulePromptQueueDrain(sessionID: string, delayMs: number): void {
|
||||||
|
const queue = promptQueues.get(sessionID)
|
||||||
|
if (!queue || queue.length === 0) {
|
||||||
|
clearPromptQueueTimer(sessionID)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
clearPromptQueueTimer(sessionID)
|
||||||
|
const timer = setTimeout(() => {
|
||||||
|
promptQueueTimers.delete(sessionID)
|
||||||
|
void drainPromptQueue(sessionID).catch((error: unknown) => {
|
||||||
|
log("[prompt-async-gate] queued prompt drain failed", {
|
||||||
|
sessionID,
|
||||||
|
error: String(error),
|
||||||
|
})
|
||||||
|
})
|
||||||
|
}, Math.max(0, delayMs))
|
||||||
|
promptQueueTimers.set(sessionID, timer)
|
||||||
|
}
|
||||||
|
|
||||||
|
function removePromptQueueEntry(sessionID: string, entry: QueuedInternalPrompt): void {
|
||||||
|
const queue = promptQueues.get(sessionID)
|
||||||
|
if (!queue) {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
const nextQueue = queue.filter((queued) => queued.id !== entry.id)
|
||||||
|
setPromptQueue(sessionID, nextQueue)
|
||||||
|
}
|
||||||
|
|
||||||
|
export function getQueuedPromptBlocker(sessionID: string): string | undefined {
|
||||||
|
const inFlight = promptQueueInFlight.get(sessionID)
|
||||||
|
if (inFlight) {
|
||||||
|
return inFlight.source
|
||||||
|
}
|
||||||
|
|
||||||
|
const queue = promptQueues.get(sessionID)
|
||||||
|
return queue?.[0]?.source
|
||||||
|
}
|
||||||
|
|
||||||
|
export function isPromptQueueDraining(sessionID: string): boolean {
|
||||||
|
return promptQueueDraining.has(sessionID)
|
||||||
|
}
|
||||||
|
|
||||||
|
export function nextPromptQueueID(): number {
|
||||||
|
promptQueueSequence += 1
|
||||||
|
return promptQueueSequence
|
||||||
|
}
|
||||||
|
|
||||||
|
export function releaseInFlightPromptMatchingDedupe(sessionID: string, dedupeKey: string): void {
|
||||||
|
const inFlight = promptQueueInFlight.get(sessionID)
|
||||||
|
if (inFlight?.dedupeKey === dedupeKey) {
|
||||||
|
removePromptQueueEntry(sessionID, inFlight)
|
||||||
|
promptQueueInFlight.delete(sessionID)
|
||||||
|
promptQueueDraining.delete(sessionID)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
export function clearPromptQueueStateForTesting(): void {
|
||||||
|
promptQueues.clear()
|
||||||
|
promptQueueDraining.clear()
|
||||||
|
promptQueueInFlight.clear()
|
||||||
|
for (const timer of promptQueueTimers.values()) {
|
||||||
|
clearTimeout(timer)
|
||||||
|
}
|
||||||
|
promptQueueTimers.clear()
|
||||||
|
}
|
||||||
|
|
||||||
|
async function drainPromptQueue(sessionID: string, awaitedEntry?: QueuedInternalPrompt): Promise<InternalPromptDispatchResult | undefined> {
|
||||||
|
if (promptQueueDraining.has(sessionID)) {
|
||||||
|
return awaitedEntry ? queuedResult(awaitedEntry, 1) : undefined
|
||||||
|
}
|
||||||
|
|
||||||
|
promptQueueDraining.add(sessionID)
|
||||||
|
clearPromptQueueTimer(sessionID)
|
||||||
|
|
||||||
|
let awaitedResult: InternalPromptDispatchResult | undefined
|
||||||
|
try {
|
||||||
|
while (true) {
|
||||||
|
const queue = promptQueues.get(sessionID)
|
||||||
|
const entry = queue?.[0]
|
||||||
|
if (!entry) {
|
||||||
|
break
|
||||||
|
}
|
||||||
|
|
||||||
|
promptQueueInFlight.set(sessionID, entry)
|
||||||
|
const result = await dispatchAfterSessionIdle({
|
||||||
|
sessionName: entry.sessionName,
|
||||||
|
client: entry.client,
|
||||||
|
sessionID: entry.sessionID,
|
||||||
|
input: entry.input,
|
||||||
|
source: entry.source,
|
||||||
|
dedupeKey: entry.dedupeKey,
|
||||||
|
settleMs: entry.settleMs,
|
||||||
|
postDispatchHoldMs: entry.postDispatchHoldMs,
|
||||||
|
dispatchTimeoutMs: entry.dispatchTimeoutMs,
|
||||||
|
checkStatus: entry.checkStatus,
|
||||||
|
checkToolState: entry.checkToolState,
|
||||||
|
dispatch: entry.dispatch,
|
||||||
|
})
|
||||||
|
if (promptQueueInFlight.get(sessionID)?.id === entry.id) {
|
||||||
|
promptQueueInFlight.delete(sessionID)
|
||||||
|
}
|
||||||
|
|
||||||
|
if (result.status === "active" || result.status === "reserved") {
|
||||||
|
const queued = queuedResult(
|
||||||
|
entry,
|
||||||
|
1,
|
||||||
|
result.status === "reserved" ? result.reservedBy : entry.source,
|
||||||
|
)
|
||||||
|
if (awaitedEntry?.id === entry.id) {
|
||||||
|
awaitedResult = queued
|
||||||
|
}
|
||||||
|
schedulePromptQueueDrain(sessionID, entry.queueRetryMs)
|
||||||
|
break
|
||||||
|
}
|
||||||
|
|
||||||
|
removePromptQueueEntry(sessionID, entry)
|
||||||
|
if (awaitedEntry?.id === entry.id) {
|
||||||
|
awaitedResult = result
|
||||||
|
}
|
||||||
|
|
||||||
|
const remainingQueue = promptQueues.get(sessionID)
|
||||||
|
if (!remainingQueue || remainingQueue.length === 0) {
|
||||||
|
break
|
||||||
|
}
|
||||||
|
|
||||||
|
schedulePromptQueueDrain(sessionID, entry.postDispatchHoldMs)
|
||||||
|
break
|
||||||
|
}
|
||||||
|
} finally {
|
||||||
|
promptQueueDraining.delete(sessionID)
|
||||||
|
}
|
||||||
|
|
||||||
|
return awaitedResult
|
||||||
|
}
|
||||||
|
|
||||||
|
export async function enqueueInternalPrompt(entry: QueuedInternalPrompt): Promise<InternalPromptDispatchResult> {
|
||||||
|
const activeReservation = getActiveReservation(entry.sessionID)
|
||||||
|
if (activeReservation?.dedupeKey === entry.dedupeKey) {
|
||||||
|
log("[prompt-async-gate] queued prompt coalesced with recent dispatch", {
|
||||||
|
sessionID: entry.sessionID,
|
||||||
|
source: entry.source,
|
||||||
|
queuedBy: activeReservation.source,
|
||||||
|
})
|
||||||
|
return queuedResult(entry, 0, activeReservation.source)
|
||||||
|
}
|
||||||
|
|
||||||
|
const queue = getPromptQueue(entry.sessionID)
|
||||||
|
const existingIndex = queue.findIndex((queued) => queued.dedupeKey === entry.dedupeKey)
|
||||||
|
if (existingIndex >= 0) {
|
||||||
|
const existing = queue[existingIndex]
|
||||||
|
if (existing) {
|
||||||
|
log("[prompt-async-gate] queued prompt coalesced with pending dispatch", {
|
||||||
|
sessionID: entry.sessionID,
|
||||||
|
source: entry.source,
|
||||||
|
queuedBy: existing.source,
|
||||||
|
position: existingIndex + 1,
|
||||||
|
})
|
||||||
|
return queuedResult(existing, existingIndex + 1)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
queue.push(entry)
|
||||||
|
log("[prompt-async-gate] queued prompt accepted", {
|
||||||
|
sessionID: entry.sessionID,
|
||||||
|
source: entry.source,
|
||||||
|
position: queue.length,
|
||||||
|
})
|
||||||
|
|
||||||
|
if (queue.length > 1 || promptQueueDraining.has(entry.sessionID)) {
|
||||||
|
schedulePromptQueueDrain(entry.sessionID, 0)
|
||||||
|
return queuedResult(entry, queue.length)
|
||||||
|
}
|
||||||
|
|
||||||
|
const result = await drainPromptQueue(entry.sessionID, entry)
|
||||||
|
return result ?? queuedResult(entry, 1)
|
||||||
|
}
|
||||||
@@ -0,0 +1,95 @@
|
|||||||
|
import { log } from "../logger"
|
||||||
|
import type { PromptAsyncReservation, PromptAsyncReservationReleaseOptions } from "./types"
|
||||||
|
|
||||||
|
const promptAsyncReservations = new Map<string, PromptAsyncReservation>()
|
||||||
|
let expiredReservationHandler: ((sessionID: string) => void) | undefined
|
||||||
|
|
||||||
|
export function setExpiredReservationHandler(handler: (sessionID: string) => void): void {
|
||||||
|
expiredReservationHandler = handler
|
||||||
|
}
|
||||||
|
|
||||||
|
function notifyExpiredReservation(sessionID: string): void {
|
||||||
|
expiredReservationHandler?.(sessionID)
|
||||||
|
}
|
||||||
|
|
||||||
|
function pruneExpiredReservations(now = Date.now()): void {
|
||||||
|
const expiredSessionIDs: string[] = []
|
||||||
|
for (const [sessionID, reservation] of promptAsyncReservations) {
|
||||||
|
if (typeof reservation.expiresAt === "number" && reservation.expiresAt <= now) {
|
||||||
|
promptAsyncReservations.delete(sessionID)
|
||||||
|
expiredSessionIDs.push(sessionID)
|
||||||
|
log("[prompt-async-gate] expired reservation released", {
|
||||||
|
sessionID,
|
||||||
|
source: reservation.source,
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
for (const sessionID of expiredSessionIDs) {
|
||||||
|
notifyExpiredReservation(sessionID)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
export function getActiveReservation(sessionID: string): PromptAsyncReservation | undefined {
|
||||||
|
pruneExpiredReservations()
|
||||||
|
return promptAsyncReservations.get(sessionID)
|
||||||
|
}
|
||||||
|
|
||||||
|
export function getPromptReservation(sessionID: string): PromptAsyncReservation | undefined {
|
||||||
|
return promptAsyncReservations.get(sessionID)
|
||||||
|
}
|
||||||
|
|
||||||
|
export function setPromptReservation(sessionID: string, reservation: PromptAsyncReservation): void {
|
||||||
|
promptAsyncReservations.set(sessionID, reservation)
|
||||||
|
}
|
||||||
|
|
||||||
|
export function finishPromptReservation(
|
||||||
|
sessionID: string,
|
||||||
|
reservation: PromptAsyncReservation,
|
||||||
|
dispatchAttempted: boolean,
|
||||||
|
postDispatchHoldMs: number,
|
||||||
|
): void {
|
||||||
|
const current = promptAsyncReservations.get(sessionID)
|
||||||
|
if (current?.token !== reservation.token) {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
if (dispatchAttempted && postDispatchHoldMs > 0) {
|
||||||
|
promptAsyncReservations.set(sessionID, {
|
||||||
|
...reservation,
|
||||||
|
expiresAt: Date.now() + postDispatchHoldMs,
|
||||||
|
})
|
||||||
|
return
|
||||||
|
}
|
||||||
|
promptAsyncReservations.delete(sessionID)
|
||||||
|
}
|
||||||
|
|
||||||
|
export function deletePromptReservation(sessionID: string): void {
|
||||||
|
promptAsyncReservations.delete(sessionID)
|
||||||
|
}
|
||||||
|
|
||||||
|
export function clearPromptReservationsForTesting(): void {
|
||||||
|
promptAsyncReservations.clear()
|
||||||
|
}
|
||||||
|
|
||||||
|
export function reservationSourceMatches(
|
||||||
|
reservationSource: string,
|
||||||
|
expectedSource: string | readonly string[],
|
||||||
|
expectedPrefix?: PromptAsyncReservationReleaseOptions["reservedByPrefix"],
|
||||||
|
): boolean {
|
||||||
|
if (typeof expectedSource === "string") {
|
||||||
|
if (reservationSource === expectedSource) {
|
||||||
|
return true
|
||||||
|
}
|
||||||
|
} else if (expectedSource.includes(reservationSource)) {
|
||||||
|
return true
|
||||||
|
}
|
||||||
|
|
||||||
|
if (expectedPrefix === undefined) {
|
||||||
|
return false
|
||||||
|
}
|
||||||
|
|
||||||
|
const prefixes = typeof expectedPrefix === "string" ? [expectedPrefix] : expectedPrefix
|
||||||
|
return prefixes
|
||||||
|
.filter((prefix) => prefix.length > 0 && prefix.endsWith(":"))
|
||||||
|
.some((prefix) => reservationSource.startsWith(prefix))
|
||||||
|
}
|
||||||
@@ -0,0 +1,118 @@
|
|||||||
|
import { log } from "../logger"
|
||||||
|
import { isSessionActive, settleAfterSessionIdle } from "../session-idle-settle"
|
||||||
|
import { sessionLatestAssistantBlocksInternalPrompt } from "./pending-tool-turn"
|
||||||
|
import {
|
||||||
|
finishPromptReservation,
|
||||||
|
getActiveReservation,
|
||||||
|
setPromptReservation,
|
||||||
|
} from "./reservations"
|
||||||
|
import { getPromptGateMessagesFetchTimeoutMs, withDispatchTimeout } from "./timing"
|
||||||
|
import type { InternalPromptDispatchResult, PromptAsyncReservation, PromptDispatchClient, PromptSessionName } from "./types"
|
||||||
|
|
||||||
|
export async function dispatchAfterSessionIdle<TInput>(args: {
|
||||||
|
readonly sessionName: PromptSessionName
|
||||||
|
readonly client: PromptDispatchClient
|
||||||
|
readonly sessionID: string
|
||||||
|
readonly input: TInput
|
||||||
|
readonly source: string
|
||||||
|
readonly dedupeKey: string
|
||||||
|
readonly settleMs: number
|
||||||
|
readonly postDispatchHoldMs: number
|
||||||
|
readonly dispatchTimeoutMs: number
|
||||||
|
readonly checkStatus: boolean
|
||||||
|
readonly checkToolState: boolean
|
||||||
|
readonly dispatch: (input: TInput) => Promise<unknown>
|
||||||
|
}): Promise<InternalPromptDispatchResult> {
|
||||||
|
const {
|
||||||
|
sessionName,
|
||||||
|
client,
|
||||||
|
sessionID,
|
||||||
|
input,
|
||||||
|
source,
|
||||||
|
dedupeKey,
|
||||||
|
settleMs,
|
||||||
|
postDispatchHoldMs,
|
||||||
|
dispatchTimeoutMs,
|
||||||
|
checkStatus,
|
||||||
|
checkToolState,
|
||||||
|
dispatch,
|
||||||
|
} = args
|
||||||
|
|
||||||
|
const existing = getActiveReservation(sessionID)
|
||||||
|
if (existing) {
|
||||||
|
log(`[prompt-async-gate] ${sessionName} skipped because session is reserved`, {
|
||||||
|
sessionID,
|
||||||
|
source,
|
||||||
|
reservedBy: existing.source,
|
||||||
|
reservedAgeMs: Date.now() - existing.reservedAt,
|
||||||
|
})
|
||||||
|
return { status: "reserved", reservedBy: existing.source }
|
||||||
|
}
|
||||||
|
|
||||||
|
const reservation: PromptAsyncReservation = {
|
||||||
|
source,
|
||||||
|
dedupeKey,
|
||||||
|
reservedAt: Date.now(),
|
||||||
|
token: Symbol(source),
|
||||||
|
}
|
||||||
|
setPromptReservation(sessionID, reservation)
|
||||||
|
let dispatchAttempted = false
|
||||||
|
|
||||||
|
try {
|
||||||
|
const canReadStatus = checkStatus && typeof client.session?.status === "function"
|
||||||
|
if (settleMs > 0) {
|
||||||
|
await settleAfterSessionIdle(settleMs)
|
||||||
|
}
|
||||||
|
|
||||||
|
let sessionActive = false
|
||||||
|
if (canReadStatus) {
|
||||||
|
try {
|
||||||
|
sessionActive = await withDispatchTimeout(
|
||||||
|
isSessionActive(client, sessionID),
|
||||||
|
Math.min(dispatchTimeoutMs, 5000),
|
||||||
|
`[prompt-async-gate] ${sessionName} isSessionActive`,
|
||||||
|
)
|
||||||
|
} catch {
|
||||||
|
sessionActive = false
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if (sessionActive) {
|
||||||
|
log(`[prompt-async-gate] ${sessionName} skipped because session is active`, { sessionID, source })
|
||||||
|
return { status: "active" }
|
||||||
|
}
|
||||||
|
|
||||||
|
if (
|
||||||
|
checkToolState
|
||||||
|
&& typeof client.session?.messages === "function"
|
||||||
|
&& await sessionLatestAssistantBlocksInternalPrompt({
|
||||||
|
client,
|
||||||
|
sessionID,
|
||||||
|
input,
|
||||||
|
sessionName,
|
||||||
|
source,
|
||||||
|
timeoutMs: Math.min(dispatchTimeoutMs, getPromptGateMessagesFetchTimeoutMs()),
|
||||||
|
})
|
||||||
|
) {
|
||||||
|
log(`[prompt-async-gate] ${sessionName} skipped because latest assistant is still active`, {
|
||||||
|
sessionID,
|
||||||
|
source,
|
||||||
|
})
|
||||||
|
return { status: "active" }
|
||||||
|
}
|
||||||
|
|
||||||
|
log(`[prompt-async-gate] ${sessionName} dispatching`, { sessionID, source })
|
||||||
|
dispatchAttempted = true
|
||||||
|
const response = await withDispatchTimeout(
|
||||||
|
dispatch(input),
|
||||||
|
dispatchTimeoutMs,
|
||||||
|
`[prompt-async-gate] ${sessionName} dispatch`,
|
||||||
|
)
|
||||||
|
log(`[prompt-async-gate] ${sessionName} dispatched`, { sessionID, source })
|
||||||
|
return { status: "dispatched", response }
|
||||||
|
} catch (error) {
|
||||||
|
log(`[prompt-async-gate] ${sessionName} failed`, { sessionID, source, error: String(error) })
|
||||||
|
return { status: "failed", error, dispatchAttempted }
|
||||||
|
} finally {
|
||||||
|
finishPromptReservation(sessionID, reservation, dispatchAttempted, postDispatchHoldMs)
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -0,0 +1,46 @@
|
|||||||
|
export const DEFAULT_PROMPT_ASYNC_POST_DISPATCH_HOLD_MS = 2_000
|
||||||
|
export const DEFAULT_PROMPT_DISPATCH_TIMEOUT_MS = 30_000
|
||||||
|
export const DEFAULT_PROMPT_GATE_MESSAGES_FETCH_TIMEOUT_MS = 5_000
|
||||||
|
export const DEFAULT_PROMPT_QUEUE_RETRY_MS = 250
|
||||||
|
|
||||||
|
declare function setTimeout(callback: () => void, delay?: number): unknown
|
||||||
|
declare function clearTimeout(timeout: unknown): void
|
||||||
|
|
||||||
|
let promptGateMessagesFetchTimeoutMsForTesting: number | undefined
|
||||||
|
|
||||||
|
export function _setPromptGateMessagesFetchTimeoutMsForTesting(value: number | undefined): void {
|
||||||
|
promptGateMessagesFetchTimeoutMsForTesting = value
|
||||||
|
}
|
||||||
|
|
||||||
|
export function getPromptGateMessagesFetchTimeoutMs(): number {
|
||||||
|
return promptGateMessagesFetchTimeoutMsForTesting ?? DEFAULT_PROMPT_GATE_MESSAGES_FETCH_TIMEOUT_MS
|
||||||
|
}
|
||||||
|
|
||||||
|
export function resetPromptGateTimingForTesting(): void {
|
||||||
|
promptGateMessagesFetchTimeoutMsForTesting = undefined
|
||||||
|
}
|
||||||
|
|
||||||
|
export async function withDispatchTimeout<T>(
|
||||||
|
operation: Promise<T>,
|
||||||
|
dispatchTimeoutMs: number,
|
||||||
|
operationName: string,
|
||||||
|
): Promise<T> {
|
||||||
|
if (dispatchTimeoutMs <= 0) {
|
||||||
|
return operation
|
||||||
|
}
|
||||||
|
|
||||||
|
let timeoutID: unknown
|
||||||
|
const timeoutPromise = new Promise<never>((_, reject) => {
|
||||||
|
timeoutID = setTimeout(() => {
|
||||||
|
reject(new Error(`${operationName} timed out after ${dispatchTimeoutMs}ms`))
|
||||||
|
}, dispatchTimeoutMs)
|
||||||
|
})
|
||||||
|
|
||||||
|
try {
|
||||||
|
return await Promise.race([operation, timeoutPromise])
|
||||||
|
} finally {
|
||||||
|
if (timeoutID !== undefined) {
|
||||||
|
clearTimeout(timeoutID)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -0,0 +1,99 @@
|
|||||||
|
export type PromptAsyncInput = {
|
||||||
|
readonly path?: { readonly id?: string }
|
||||||
|
readonly body?: unknown
|
||||||
|
readonly query?: unknown
|
||||||
|
readonly signal?: unknown
|
||||||
|
readonly [key: string]: unknown
|
||||||
|
}
|
||||||
|
|
||||||
|
export type PromptMessagesQuery = {
|
||||||
|
readonly directory: string
|
||||||
|
readonly limit?: number
|
||||||
|
}
|
||||||
|
|
||||||
|
export type PromptAsyncClient<TInput> = {
|
||||||
|
readonly session?: {
|
||||||
|
readonly status?: () => Promise<unknown>
|
||||||
|
readonly messages?: (input: { readonly path: { readonly id: string }; readonly query: PromptMessagesQuery }) => Promise<unknown>
|
||||||
|
readonly promptAsync?: (input: TInput) => Promise<unknown>
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
export type PromptClient<TInput> = {
|
||||||
|
readonly session?: {
|
||||||
|
readonly status?: () => Promise<unknown>
|
||||||
|
readonly messages?: (input: { readonly path: { readonly id: string }; readonly query: PromptMessagesQuery }) => Promise<unknown>
|
||||||
|
readonly prompt?: (input: TInput) => Promise<unknown>
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
export type InternalPromptDispatchMode = "async" | "sync"
|
||||||
|
export type InternalPromptQueueBehavior = "enqueue" | "defer"
|
||||||
|
export type PromptSessionName = "promptAsync" | "prompt"
|
||||||
|
|
||||||
|
type InternalPromptDispatchCommonArgs<TInput> = {
|
||||||
|
readonly sessionID: string
|
||||||
|
readonly input: TInput
|
||||||
|
readonly source: string
|
||||||
|
readonly dedupeKey?: string
|
||||||
|
readonly queueBehavior?: InternalPromptQueueBehavior
|
||||||
|
readonly queue?: boolean
|
||||||
|
readonly queueRetryMs?: number
|
||||||
|
readonly settleMs?: number
|
||||||
|
readonly postDispatchHoldMs?: number
|
||||||
|
readonly dispatchTimeoutMs?: number
|
||||||
|
readonly checkStatus?: boolean
|
||||||
|
readonly checkToolState?: boolean
|
||||||
|
}
|
||||||
|
|
||||||
|
export type InternalPromptDispatchArgs<TInput = PromptAsyncInput> = InternalPromptDispatchCommonArgs<TInput> & (
|
||||||
|
| { readonly mode: "async"; readonly client: PromptAsyncClient<TInput> }
|
||||||
|
| { readonly mode: "sync"; readonly client: PromptClient<TInput> }
|
||||||
|
)
|
||||||
|
|
||||||
|
export type PromptAsyncReservation = {
|
||||||
|
readonly source: string
|
||||||
|
readonly dedupeKey: string
|
||||||
|
readonly reservedAt: number
|
||||||
|
readonly token: symbol
|
||||||
|
readonly expiresAt?: number
|
||||||
|
}
|
||||||
|
|
||||||
|
export type InternalPromptDispatchResult =
|
||||||
|
| { readonly status: "dispatched"; readonly response: unknown }
|
||||||
|
| { readonly status: "queued"; readonly queuedBy: string; readonly position: number }
|
||||||
|
| { readonly status: "active" }
|
||||||
|
| { readonly status: "reserved"; readonly reservedBy: string }
|
||||||
|
| { readonly status: "unavailable" }
|
||||||
|
| { readonly status: "failed"; readonly error: unknown; readonly dispatchAttempted: boolean }
|
||||||
|
|
||||||
|
export type PromptAsyncGateResult = InternalPromptDispatchResult
|
||||||
|
|
||||||
|
export type PromptAsyncReservationReleaseOptions = {
|
||||||
|
readonly reservedBy?: string | readonly string[]
|
||||||
|
readonly reservedByPrefix?: string | readonly string[]
|
||||||
|
}
|
||||||
|
|
||||||
|
export type PromptDispatchClient = {
|
||||||
|
readonly session?: {
|
||||||
|
readonly status?: () => Promise<unknown>
|
||||||
|
readonly messages?: (input: { readonly path: { readonly id: string }; readonly query: PromptMessagesQuery }) => Promise<unknown>
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
export type QueuedInternalPrompt = {
|
||||||
|
readonly id: number
|
||||||
|
readonly sessionID: string
|
||||||
|
readonly sessionName: PromptSessionName
|
||||||
|
readonly client: PromptDispatchClient
|
||||||
|
readonly input: unknown
|
||||||
|
readonly source: string
|
||||||
|
readonly dedupeKey: string
|
||||||
|
readonly settleMs: number
|
||||||
|
readonly postDispatchHoldMs: number
|
||||||
|
readonly dispatchTimeoutMs: number
|
||||||
|
readonly queueRetryMs: number
|
||||||
|
readonly checkStatus: boolean
|
||||||
|
readonly checkToolState: boolean
|
||||||
|
readonly dispatch: (input: unknown) => Promise<unknown>
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user