{"version":3,"file":"generation.d.ts","sourceRoot":"","sources":["../../../../src/harness/runtime/drive/generation.ts"],"names":[],"mappings":"AAMA,OAAO,EAEN,KAAK,uBAAuB,EAC5B,KAAK,2BAA2B,EAMhC,MAAM,wBAAwB,CAAC;AAEhC,OAAO,KAAK,EAAE,IAAI,EAAE,MAAM,YAAY,CAAC;AAEvC,OAAO,KAAK,EAA2B,KAAK,EAAE,eAAe,EAAE,MAAM,aAAa,CAAC;AAqNnF,2FAA2F;AAC3F,wBAAsB,YAAY,CAAC,QAAQ,SAAS,MAAM,GAAG,SAAS,EACrE,IAAI,EAAE,IAAI,CAAC,QAAQ,CAAC,EACpB,KAAK,EAAE,KAAK,EACZ,UAAU,EAAE,2BAA2B,GACrC,OAAO,CAAC,eAAe,CAAC,CA4C1B;AAED,gFAAgF;AAChF,wBAAsB,aAAa,CAAC,QAAQ,SAAS,MAAM,GAAG,SAAS,EACtE,IAAI,EAAE,IAAI,CAAC,QAAQ,CAAC,EACpB,KAAK,EAAE,KAAK,EACZ,UAAU,EAAE,uBAAuB,GAAG,2BAA2B,GAC/D,OAAO,CAAC,eAAe,CAAC,CAa1B","sourcesContent":["import type { Api, Model, Tool } from \"@earendil-works/pi-ai\";\nimport type { AgentMessage } from \"../../../types.ts\";\nimport { type Context, getTelemetryContext, withAbortSignal } from \"../../context.ts\";\nimport { type HarnessAssistantStreamConfig, streamHarnessAssistant } from \"../../execution/assistant.ts\";\nimport { applyStreamOptionsPatch } from \"../../hooks.ts\";\nimport { SessionInvariantError } from \"../../session/session.ts\";\nimport {\n\ttype AssistantEffectPendingOperation,\n\ttype AssistantReadyOperation,\n\ttype AssistantRetryWaitOperation,\n\ttype JsonValue,\n\ttype OperationError,\n\ttype OperationScope,\n\toperationScopeOf,\n\ttype SettledAssistantMessage,\n} from \"../../session/types.ts\";\nimport type { AgentHarnessStreamOptions } from \"../../types.ts\";\nimport type { Lane } from \"../lane.ts\";\nimport { readBoundedContext } from \"../transcript.ts\";\nimport type { ContinueOperationResult, Drive, ProcedureResult } from \"../types.ts\";\nimport { openAssistantResponse, publishConfigurationFailure, publishResponse } from \"./response.ts\";\nimport { waitUntil } from \"./retry.ts\";\n\n/** Assistant effect-pending payload without the run-wide scope fields. */\ntype AssistantEffectPending = Omit<AssistantEffectPendingOperation, keyof OperationScope>;\n\ntype PreparedGeneration = {\n\tkind: \"ready\";\n\tmodel: Model<Api>;\n\ttools: Tool[];\n\tmessages: AgentMessage[];\n\tsystemPrompt: string;\n\tstreamOptions: AgentHarnessStreamOptions;\n\ttoProviderMessages: HarnessAssistantStreamConfig[\"toProviderMessages\"];\n};\n\ntype GenerationPreparation =\n\t| PreparedGeneration\n\t| { kind: \"configuration_failure\"; error: OperationError }\n\t| { kind: \"cancel_requested\" };\n\nfunction configurationError(\n\tcode: \"model_unavailable\" | \"configured_tools_unavailable\",\n\tdetails: JsonValue,\n): OperationError {\n\treturn {\n\t\tcode,\n\t\tmessage:\n\t\t\tcode === \"model_unavailable\"\n\t\t\t\t? \"The configured model is unavailable in this process\"\n\t\t\t\t: \"One or more configured tools are unavailable in this process\",\n\t\tdetails,\n\t};\n}\n\nasync function resolveSystemPrompt<TContext extends object | undefined>(\n\tlane: Lane<TContext>,\n\tcontext: Context,\n): Promise<string> {\n\tconst config = lane.readConfig();\n\tif (config.systemPrompt === undefined) return \"\";\n\tif (typeof config.systemPrompt === \"string\") return config.systemPrompt;\n\tconst source = config.toolContext;\n\tconst toolContext = typeof source === \"function\" ? await source(context) : source;\n\treturn config.systemPrompt(toolContext as TContext, context);\n}\n\nasync function prepareGeneration<TContext extends object | undefined>(\n\tlane: Lane<TContext>,\n\tdrive: Drive,\n\tgeneration: AssistantReadyOperation,\n): Promise<GenerationPreparation> {\n\tconst identity = generation.generationContext.configuration.model;\n\tconst model = lane.models.getModel(identity.provider, identity.modelId);\n\tif (model === undefined) {\n\t\treturn { kind: \"configuration_failure\", error: configurationError(\"model_unavailable\", identity) };\n\t}\n\n\tconst config = lane.readConfig();\n\tconst toolsByName = new Map(config.tools.map((tool) => [tool.name, tool]));\n\tconst missingTools = generation.generationContext.configuration.activeToolNames.filter(\n\t\t(name) => !toolsByName.has(name),\n\t);\n\tif (missingTools.length !== 0) {\n\t\treturn {\n\t\t\tkind: \"configuration_failure\",\n\t\t\terror: configurationError(\"configured_tools_unavailable\", { tools: missingTools }),\n\t\t};\n\t}\n\tconst tools: Tool[] = generation.generationContext.configuration.activeToolNames.map((name) => {\n\t\tconst tool = toolsByName.get(name);\n\t\tif (tool === undefined) throw new SessionInvariantError(`Configured tool ${name} disappeared during resolution`);\n\t\treturn {\n\t\t\tname: tool.name,\n\t\t\tdescription: tool.description,\n\t\t\tparameters: tool.parameters,\n\t\t\t...(tool.constrainedSampling === undefined ? {} : { constrainedSampling: tool.constrainedSampling }),\n\t\t};\n\t});\n\n\tconst messages = await readBoundedContext(lane, drive, generation);\n\tif (messages.kind === \"cancel_requested\") return messages;\n\tconst systemPrompt = await resolveSystemPrompt(lane, drive.context);\n\tconst beforeRequest = await lane.hooks.runWithGate(\n\t\t\"before_request\",\n\t\t{\n\t\t\tlane: lane.name,\n\t\t\trunId: drive.operationId,\n\t\t\tmodel,\n\t\t\tstep: \"assistant\",\n\t\t\tattempt: generation.nextAttempt,\n\t\t\tstreamOptions: generation.generationContext.streamOptions,\n\t\t},\n\t\tdrive.gate,\n\t\tdrive.context,\n\t);\n\tconst streamOptions =\n\t\tbeforeRequest?.streamOptions === undefined\n\t\t\t? generation.generationContext.streamOptions\n\t\t\t: applyStreamOptionsPatch(generation.generationContext.streamOptions, beforeRequest.streamOptions);\n\treturn {\n\t\tkind: \"ready\",\n\t\tmodel,\n\t\ttools,\n\t\tmessages: messages.value,\n\t\tsystemPrompt,\n\t\tstreamOptions,\n\t\ttoProviderMessages: config.toProviderMessages,\n\t};\n}\n\nasync function publishGenerationIntent<TContext extends object | undefined>(\n\tlane: Lane<TContext>,\n\tdrive: Drive,\n\tready: AssistantReadyOperation,\n\tprepared: PreparedGeneration,\n): Promise<ContinueOperationResult<AssistantEffectPendingOperation>> {\n\tconst at = Date.now();\n\tconst pending: AssistantEffectPending = {\n\t\tat: \"assistant.effect_pending\",\n\t\tgenerationContext: ready.generationContext,\n\t\tattempt: ready.nextAttempt,\n\t\tresponseEntryId: lane.session.idGenerator.next(at),\n\t\tusageId: lane.session.idGenerator.next(at),\n\t\tintendedOutputLimit: prepared.model.maxTokens,\n\t\tcontextWindow: prepared.model.contextWindow,\n\t};\n\treturn lane.continueOperation(\n\t\tready,\n\t\t(_state, current) => {\n\t\t\tconst nextState: AssistantEffectPendingOperation = { ...operationScopeOf(current), ...pending };\n\t\t\treturn {\n\t\t\t\tkind: \"commit\",\n\t\t\t\twrites: [],\n\t\t\t\toperationState: nextState,\n\t\t\t\tmaterialize: () => nextState,\n\t\t\t\tevents: () =>\n\t\t\t\t\tready.nextAttempt === 1\n\t\t\t\t\t\t? [\n\t\t\t\t\t\t\t\t{\n\t\t\t\t\t\t\t\t\ttype: \"turn_start\",\n\t\t\t\t\t\t\t\t\tlane: lane.name,\n\t\t\t\t\t\t\t\t\trunId: drive.operationId,\n\t\t\t\t\t\t\t\t\tturnId: ready.generationContext.stepId,\n\t\t\t\t\t\t\t\t},\n\t\t\t\t\t\t\t]\n\t\t\t\t\t\t: [],\n\t\t\t};\n\t\t},\n\t\tdrive.context,\n\t);\n}\n\nasync function performGeneration<TContext extends object | undefined>(\n\tlane: Lane<TContext>,\n\tdrive: Drive,\n\tintent: AssistantEffectPendingOperation,\n\tprepared: PreparedGeneration,\n): Promise<SettledAssistantMessage> {\n\tconst response = openAssistantResponse(lane, drive, intent.responseEntryId);\n\ttry {\n\t\treturn await streamHarnessAssistant(\n\t\t\tprepared.messages,\n\t\t\t{\n\t\t\t\tmodel: prepared.model,\n\t\t\t\tsystemPrompt: prepared.systemPrompt,\n\t\t\t\ttools: prepared.tools,\n\t\t\t\tthinkingLevel: intent.generationContext.configuration.thinkingLevel,\n\t\t\t\tstreamOptions: prepared.streamOptions,\n\t\t\t\ttransformContext: async (requestContext, context) => {\n\t\t\t\t\tconst result = await lane.hooks.runWithGate(\n\t\t\t\t\t\t\"transform_context\",\n\t\t\t\t\t\t{ lane: lane.name, runId: drive.operationId, ...requestContext },\n\t\t\t\t\t\tdrive.gate,\n\t\t\t\t\t\tcontext,\n\t\t\t\t\t);\n\t\t\t\t\treturn {\n\t\t\t\t\t\tmessages: result?.messages ?? requestContext.messages,\n\t\t\t\t\t\tsystemPrompt: result?.systemPrompt ?? requestContext.systemPrompt,\n\t\t\t\t\t};\n\t\t\t\t},\n\t\t\t\ttoProviderMessages: prepared.toProviderMessages,\n\t\t\t\tbeforePayload: async (payload, requestModel, context) => {\n\t\t\t\t\tconst result = await lane.hooks.runWithGate(\n\t\t\t\t\t\t\"before_payload\",\n\t\t\t\t\t\t{ lane: lane.name, runId: drive.operationId, model: requestModel, payload },\n\t\t\t\t\t\tdrive.gate,\n\t\t\t\t\t\tcontext,\n\t\t\t\t\t);\n\t\t\t\t\treturn result?.payload;\n\t\t\t\t},\n\t\t\t\tafterResponse: response.afterResponse,\n\t\t\t\trequest: (aiContext, options, context) => {\n\t\t\t\t\tconst admitted = withAbortSignal(drive.gate.signal, context);\n\t\t\t\t\treturn drive.gate.admit(() =>\n\t\t\t\t\t\tlane.models.streamSimple(prepared.model, aiContext, {\n\t\t\t\t\t\t\t...options,\n\t\t\t\t\t\t\tsessionId: `${lane.session.metadata.id}:${lane.name}`,\n\t\t\t\t\t\t\tsignal: admitted.abortSignal,\n\t\t\t\t\t\t\ttelemetryContext: getTelemetryContext(admitted),\n\t\t\t\t\t\t}),\n\t\t\t\t\t);\n\t\t\t\t},\n\t\t\t\tobserver: response.observer,\n\t\t\t},\n\t\t\tdrive.context,\n\t\t);\n\t} finally {\n\t\tawait response.close();\n\t}\n}\n\n/** Advance one durable assistant retry wait according to this pass's local wait policy. */\nexport async function runRetryWait<TContext extends object | undefined>(\n\tlane: Lane<TContext>,\n\tdrive: Drive,\n\tgeneration: AssistantRetryWaitOperation,\n): Promise<ProcedureResult> {\n\tif (Date.now() < generation.notBefore) {\n\t\tif (!drive.waitForRetry) {\n\t\t\treturn {\n\t\t\t\tkind: \"waiting\",\n\t\t\t\toutcome: {\n\t\t\t\t\tkind: \"waiting\",\n\t\t\t\t\toperationId: drive.operationId,\n\t\t\t\t\treason: \"retry\",\n\t\t\t\t\tnotBefore: generation.notBefore,\n\t\t\t\t},\n\t\t\t};\n\t\t}\n\t\tawait drive.gate.admit(() => waitUntil(generation.notBefore, drive.gate.signal));\n\t}\n\n\tconst result = await lane.continueOperation(\n\t\tgeneration,\n\t\t(_state, current) => {\n\t\t\tconst nextState: AssistantReadyOperation = {\n\t\t\t\t...operationScopeOf(current),\n\t\t\t\tat: \"assistant.ready\",\n\t\t\t\tgenerationContext: generation.generationContext,\n\t\t\t\tnextAttempt: generation.nextAttempt,\n\t\t\t};\n\t\t\treturn {\n\t\t\t\tkind: \"commit\",\n\t\t\t\twrites: [],\n\t\t\t\toperationState: nextState,\n\t\t\t\tmaterialize: () => ({ kind: \"continue\" }) as const,\n\t\t\t\tevents: () => [\n\t\t\t\t\t{\n\t\t\t\t\t\ttype: \"retry_start\",\n\t\t\t\t\t\tlane: lane.name,\n\t\t\t\t\t\trunId: drive.operationId,\n\t\t\t\t\t\tstep: generation.generationContext.stepId,\n\t\t\t\t\t\tattempt: generation.nextAttempt,\n\t\t\t\t\t},\n\t\t\t\t],\n\t\t\t};\n\t\t},\n\t\tdrive.context,\n\t);\n\treturn result.kind === \"cancel_requested\" ? { kind: \"continue\" } : result.value;\n}\n\n/** Execute one ready assistant generation or advance its durable retry wait. */\nexport async function runGeneration<TContext extends object | undefined>(\n\tlane: Lane<TContext>,\n\tdrive: Drive,\n\tgeneration: AssistantReadyOperation | AssistantRetryWaitOperation,\n): Promise<ProcedureResult> {\n\tif (generation.at === \"assistant.retry_wait\") return runRetryWait(lane, drive, generation);\n\n\tconst prepared = await prepareGeneration(lane, drive, generation);\n\tif (prepared.kind === \"configuration_failure\") {\n\t\treturn publishConfigurationFailure(lane, drive, generation, prepared.error);\n\t}\n\tif (prepared.kind === \"cancel_requested\") return { kind: \"continue\" };\n\n\tconst intent = await publishGenerationIntent(lane, drive, generation, prepared);\n\tif (intent.kind === \"cancel_requested\") return { kind: \"continue\" };\n\tconst response = await performGeneration(lane, drive, intent.value, prepared);\n\treturn publishResponse(lane, drive, intent.value, response);\n}\n"]}