diff --git a/TELEMETRY.md b/TELEMETRY.md index ed524fc456..1d3a0645cd 100644 --- a/TELEMETRY.md +++ b/TELEMETRY.md @@ -26,6 +26,7 @@ use the query recipes below to find the failing turn and next query. | ----------------------------------- | ----------------------------- | ------------------------- | --------------------- | | `event_id` | captured Sentry error | failed Slack reply | open event | | `gen_ai.conversation.id` | Slack thread/run conversation | Slack footer, logs, spans | query trace/logs | +| `app.ai.turn.id` | one durable Junior turn | logs, agent/tool spans | query turn | | `trace_id` | end-to-end trace | errors, logs, spans | open trace | | `span_id` | one span in a trace | logs, spans | inspect span | | `messaging.message.conversation_id` | Slack thread | logs, spans | thread logs | @@ -34,6 +35,14 @@ use the query recipes below to find the failing turn and next query. | `gen_ai.tool.name` | tool name | tool spans/logs | tool failures | | `app.credential.provider` | auth provider | auth logs | auth/resume search | +## Semantic Conventions + +Use OpenTelemetry GenAI attributes for the concepts the specification defines: +operation, agent identity, conversation ID, request model, token usage, and +errors. OpenTelemetry does not currently define a durable agent-turn ID, +cumulative agent-step count, or cumulative turn runtime, so Junior records +those gaps under `app.ai.turn.*` rather than inventing new `gen_ai.*` fields. + ## Query Recipes Conversation timeline from a Slack thread, footer link, or conversation ID. @@ -52,6 +61,21 @@ fields=timestamp,level,event.name,trace_id,span_id,error.type,exception.message sort=timestamp ``` +One durable turn across execution slices and child model/tool spans. + +```text +dataset=spans query='app.ai.turn.id:""' +fields=timestamp,trace,span.op,span.description,span.duration,gen_ai.conversation.id,gen_ai.request.model,app.ai.turn.slice_id,app.ai.turn.step_count,app.ai.turn.runtime_ms,app.ai.turn.state,error.type +sort=timestamp +``` + +Observed completed-turn step and cumulative-runtime distributions. + +```text +dataset=spans query='span.op:chat.turn app.ai.turn.state:completed' +fields=count(),p50(app.ai.turn.step_count),p90(app.ai.turn.step_count),p95(app.ai.turn.step_count),p99(app.ai.turn.step_count),max(app.ai.turn.step_count),p50(app.ai.turn.runtime_ms),p90(app.ai.turn.runtime_ms),p95(app.ai.turn.runtime_ms),p99(app.ai.turn.runtime_ms),max(app.ai.turn.runtime_ms) +``` + Trace log history after opening a Sentry event or trace. ```text @@ -84,6 +108,14 @@ fields=timestamp,trace,gen_ai.conversation.id,gen_ai.tool.name,gen_ai.tool.call. sort=-gen_ai.tool.call.result.size ``` +System budgets exceeded while admitting or running work. + +```text +dataset=logs query='event.name:system.budget.exceeded' +fields=timestamp,event.name,gen_ai.conversation.id,app.ai.turn.id,app.budget.name,app.budget.outcome,app.budget.value,app.budget.limit +sort=-timestamp +``` + Search tool volume, truncation, and raw output size. ```text @@ -135,6 +167,8 @@ Spans: `chat.turn`, `chat.reply`, `chat.slash_command`, `chat.app_home_opened`, `chat.app_home_disconnect` Attributes: `trace_id`, `span_id`, `gen_ai.conversation.id`, +`app.ai.turn.id`, `app.ai.turn.state`, `app.ai.turn.step_count`, +`app.ai.turn.runtime_ms`, `messaging.message.conversation_id`, `messaging.destination.name`, `app.slack.reply_stage`, `app.slack.error_code`, `app.slack.api_error` @@ -145,6 +179,7 @@ The turn timed out, returned no useful answer, or used unexpected reasoning. Events: `agent.message.received`, `agent.message.generated`, `agent.turn.timed_out`, `agent.turn.provider_error`, `agent.turn.execution.failed`, +`system.budget.exceeded`, `agent.turn.empty_output.retrying`, `agent.turn.empty_output.exhausted`, `assistant.reply.generation.failed`, `guardian.action_review.retrying` @@ -153,15 +188,18 @@ Spans: `ai.generate_assistant_reply`, `ai.chat_completion`, `chat.route_thinking`, `gen_ai.invoke_agent`, `gen_ai.chat` Attributes: `gen_ai.operation.name`, `gen_ai.request.model`, -`gen_ai.response.finish_reasons`, `app.ai.outcome`, -`app.ai.reasoning_effort`, `app.ai.model_profile`, `gen_ai.usage.input_tokens`, -`gen_ai.usage.output_tokens`, `gen_ai.usage.input_tokens.cached`, -`gen_ai.usage.input_tokens.cache_write`, `app.ai.reasoning_tokens`, +`gen_ai.agent.name`, `gen_ai.conversation.id`, `gen_ai.response.finish_reasons`, +`gen_ai.usage.input_tokens`, `gen_ai.usage.output_tokens`, +`app.ai.turn.id`, `app.ai.turn.step_count`, `app.ai.turn.runtime_ms`, +`app.ai.outcome`, +`app.ai.reasoning_effort`, `app.ai.model_profile`, +`gen_ai.usage.cache_read.input_tokens`, +`gen_ai.usage.cache_creation.input_tokens`, +`gen_ai.usage.reasoning.output_tokens`, `app.ai.reasoning_tokens`, `app.ai.empty_output.attempt`, `app.ai.provider_error.kind`, `app.guardian.review_attempt`, -`app.ai.cost.input_usd`, `app.ai.cost.output_usd`, -`app.ai.cost.cache_read_usd`, `app.ai.cost.cache_write_usd`, -`app.ai.cost.total_usd` +`app.cost.input_usd`, `app.cost.output_usd`, `app.cost.cache_read_usd`, +`app.cost.cache_write_usd`, `app.cost.total_usd` ### Tools, MCP, And Sandbox diff --git a/TERMINOLOGY.md b/TERMINOLOGY.md index 88416a505e..0ecdc26f3f 100644 --- a/TERMINOLOGY.md +++ b/TERMINOLOGY.md @@ -26,6 +26,12 @@ Canonical words used across Junior's code and documentation. normal ordering and waits when a turn is already active. - **Turn**: one request-to-final-response cycle. It may span multiple runs and execution slices; one model invocation is not a turn. +- **Agent step**: one model response attempt inside a turn, including any tool + calls and results that follow that response. A turn may contain many agent + steps across runs and execution slices. +- **System budget**: one configured resource boundary checked against current + runtime usage. Exceeding a capacity budget queues work; exceeding a turn + budget stops the turn. - **Run**: one bounded attempt to advance a turn. A later run may resume the same turn after a pause, yield, or recoverable failure. - **Execution slice**: one serverless invocation segment of a run. @@ -57,7 +63,8 @@ Canonical words used across Junior's code and documentation. ## Naming Guidance -- Use `turn`, `run`, and `slice` only with the meanings above. +- Use `turn`, `agent step`, `system budget`, `run`, and `slice` only with the + meanings above. - Use `message` for platform chat content. Use `user_message`, `assistant_message`, and `tool_result` for replayable agent history. - Use `agent history item` when referring to those three native event types as diff --git a/packages/docs/src/content/docs/reference/config-and-env.md b/packages/docs/src/content/docs/reference/config-and-env.md index 4b14594b63..c00e1af8ac 100644 --- a/packages/docs/src/content/docs/reference/config-and-env.md +++ b/packages/docs/src/content/docs/reference/config-and-env.md @@ -45,6 +45,34 @@ for its own turn. Set it to `steer` to preserve collaborative steering across actors. In `follow_up` mode, a user can start one message with `!!` to steer the active turn explicitly. +## System protections + +These budgets are parsed into `botConfig.budgets` at startup, checked through +the shared `checkBudgets()` decision path, and shown on the dashboard's +**System** page. + +Junior's private internal budget registry owns each budget's environment name, +default limit, unit, display copy, outcome, runtime stage, and measurement +function. Nitro configuration contains only the resulting numeric limits; +budget functions stay in the bundled runtime and are never serialized. + +| Variable | Default | Protection | +| ------------------------------------------ | ---------- | ----------------------------------------------------------------------------------------------------- | +| `JUNIOR_MAX_ACTIVE_CONVERSATIONS` | `100` | Maximum conversations holding execution leases globally. Additional conversations remain queued. | +| `JUNIOR_MAX_ACTIVE_CONVERSATIONS_PER_USER` | `5` | Maximum active conversations for one platform-scoped user when a stable user ID is available. | +| `JUNIOR_MAX_STEPS_PER_TURN` | `500` | Maximum agent steps in one turn across model responses, tool calls, retries, and resumes. | +| `JUNIOR_MAX_TURN_RUNTIME_MS` | `21600000` | Maximum cumulative active runtime for one turn across timeout, yield, and recovery resumes (6 hours). | + +Active-conversation budgets are admission controls, not rejection limits. Inbound +messages remain durable, and a conversation that cannot acquire a lease is +queued again after a short delay. The per-user limit applies when the inbound +source provides a stable user ID; the global limit still covers system work. +Step and runtime budgets stop one runaway turn instead of leaving it queued. +Those hard-budget hits are available from `/api/stats` under the `junior` +namespace and `budget_exceeded` metric. Every exceeded budget emits the shared +event `system.budget.exceeded` with its name, value, limit, and `queue` or +`stop` outcome. Stop outcomes are also captured as Sentry issues. + Model profile names are durable conversation bindings. Each later turn resolves the stored name through current configuration; the exact model ID recorded when an epoch opens is audit evidence, not a runtime pin. Changing a mapping retargets diff --git a/packages/junior-dashboard/src/api/schema.ts b/packages/junior-dashboard/src/api/schema.ts index 29096489fa..5887919ce0 100644 --- a/packages/junior-dashboard/src/api/schema.ts +++ b/packages/junior-dashboard/src/api/schema.ts @@ -21,6 +21,21 @@ export const dashboardConfigSchema = z basePath: z.string(), componentGallery: z.boolean(), sentryConversationLinks: z.boolean(), + systemBudgets: z + .array( + z + .object({ + description: z.string().min(1), + label: z.string().min(1), + limit: z.number().positive(), + name: z.string().min(1), + outcome: z.enum(["queue", "stop"]), + stage: z.enum(["conversation_admission", "turn"]), + unit: z.enum(["count", "milliseconds", "usd"]), + }) + .strict(), + ) + .optional(), timeZone: z.string(), }) .strict(); diff --git a/packages/junior-dashboard/src/app.ts b/packages/junior-dashboard/src/app.ts index b6d4d04124..f9c357abdb 100644 --- a/packages/junior-dashboard/src/app.ts +++ b/packages/junior-dashboard/src/app.ts @@ -14,7 +14,11 @@ import type { PluginRouteApp, } from "@sentry/junior-plugin-api"; import { pluginApiRouteRequestContextSchema } from "@sentry/junior-plugin-api"; -import { dashboardConfigSchema, dashboardIdentitySchema } from "./api/schema"; +import { + dashboardConfigSchema, + dashboardIdentitySchema, + type DashboardConfig, +} from "./api/schema"; import { dashboardAvatarHeaderAsset, dashboardClientAsset, @@ -55,6 +59,7 @@ export interface JuniorDashboardOptions { interface DashboardRuntimeOptions extends JuniorDashboardOptions { pluginRoutes?: DashboardPluginRoute[]; + systemBudgets?: NonNullable; } interface DashboardPluginRoute { @@ -767,6 +772,9 @@ export function createDashboardApp( basePath, componentGallery: options.componentGallery === true, sentryConversationLinks: hasSentryConversationLinks(), + ...(options.systemBudgets + ? { systemBudgets: options.systemBudgets } + : {}), timeZone: dashboardTimeZone(), }); }); diff --git a/packages/junior-dashboard/src/client/pages/system/SystemBudgets.tsx b/packages/junior-dashboard/src/client/pages/system/SystemBudgets.tsx new file mode 100644 index 0000000000..5a3e1a654e --- /dev/null +++ b/packages/junior-dashboard/src/client/pages/system/SystemBudgets.tsx @@ -0,0 +1,67 @@ +import { ShieldCheck } from "lucide-react"; + +import type { DashboardConfig } from "../../types"; +import { Card } from "../../components/layout/Card"; +import { SectionIntro } from "../../components/layout/SectionIntro"; + +type SystemBudgetDescriptions = NonNullable; + +function formatLimit(budget: SystemBudgetDescriptions[number]): string { + if (budget.unit === "milliseconds") { + const hours = budget.limit / (60 * 60 * 1000); + return Number.isInteger(hours) + ? `${hours} hours` + : `${hours.toFixed(1)} hours`; + } + if (budget.unit === "usd") { + return `$${budget.limit.toLocaleString(undefined, { + maximumFractionDigits: 2, + })}`; + } + return budget.limit.toLocaleString(); +} + +/** Show the configured budgets that queue work or stop runaway turns. */ +export function SystemBudgets(props: { budgets: SystemBudgetDescriptions }) { + return ( +
+ + +
+ {props.budgets.map((budget) => ( +
+
+ + + + {budget.outcome === "queue" ? "Queue" : "Stop"} + +
+
+ {formatLimit(budget)} +
+

+ {budget.description} +

+
+ ))} +
+ +
+ ); +} diff --git a/packages/junior-dashboard/src/client/pages/system/SystemPage.tsx b/packages/junior-dashboard/src/client/pages/system/SystemPage.tsx index 3b0d35a464..52165dcd12 100644 --- a/packages/junior-dashboard/src/client/pages/system/SystemPage.tsx +++ b/packages/junior-dashboard/src/client/pages/system/SystemPage.tsx @@ -16,6 +16,7 @@ import { PluginReports } from "./PluginReports"; import { SkillInventory } from "./SkillInventory"; import { SystemActivity } from "./SystemActivity"; import { SystemPageLayout } from "./SystemPageLayout"; +import { SystemBudgets } from "./SystemBudgets"; import { buildSystemPlugins, normalizeSystemPath, @@ -94,6 +95,9 @@ function OverviewSystemPage(props: { eyebrow={`${agentNamePossessive()} engine room`} title="System" /> + {props.data.config.systemBudgets ? ( + + ) : null} { + it("returns configured system budgets", async () => { + const app = createDashboardApp({ + authRequired: false, + systemBudgets: [ + { + description: "Stops runaway model and tool loops.", + label: "Agent steps per turn", + limit: 500, + name: "turn_steps", + outcome: "stop", + stage: "turn", + unit: "count", + }, + ], + }); + + await expect( + (await app.fetch(new Request("http://localhost/api/config"))).json(), + ).resolves.toMatchObject({ + systemBudgets: [ + { + limit: 500, + name: "turn_steps", + outcome: "stop", + }, + ], + }); + }); +}); diff --git a/packages/junior-dashboard/tests/system-budgets.test.tsx b/packages/junior-dashboard/tests/system-budgets.test.tsx new file mode 100644 index 0000000000..7bd664f822 --- /dev/null +++ b/packages/junior-dashboard/tests/system-budgets.test.tsx @@ -0,0 +1,60 @@ +import { renderToStaticMarkup } from "react-dom/server"; +import { describe, expect, it } from "vitest"; +import { SystemBudgets } from "../src/client/pages/system/SystemBudgets"; + +describe("system budgets", () => { + it("shows configured queue and stop budgets", () => { + const html = renderToStaticMarkup( + , + ); + + expect(html).toContain("System budgets"); + expect(html).toContain("Active globally"); + expect(html).toContain("Agent steps per turn"); + expect(html).toContain("Runtime per turn"); + expect(html).toContain("Daily spend"); + expect(html).toContain("$250"); + expect(html).toContain("6 hours"); + expect(html.match(/>QueueStop { , ); expect(systemHtml).toContain("Usage over time"); + expect(systemHtml).toContain("System budgets"); + expect(systemHtml).toContain("Agent steps per turn"); + expect(systemHtml).toContain("6 hours"); expect(systemHtml).toContain("Token usage"); expect(systemHtml).toContain("Model spend"); expect(systemHtml).toContain("Runtime"); diff --git a/packages/junior/src/app.ts b/packages/junior/src/app.ts index 4a4fbb036d..6f782432f1 100644 --- a/packages/junior/src/app.ts +++ b/packages/junior/src/app.ts @@ -14,6 +14,10 @@ import { getSlackReactionConfig, setSlackReactionConfig, } from "@/chat/config"; +import { + describeBudgets, + type BudgetDescription, +} from "@/chat/services/budgets"; import { getDb } from "@/chat/db"; import { logException } from "@/chat/logging"; import { executeAgentRun } from "@/chat/agent"; @@ -140,6 +144,7 @@ export interface JuniorDashboardOptions { interface JuniorDashboardRuntimeOptions extends JuniorDashboardOptions { agentName?: string; pluginRoutes?: PluginApiRouteRegistration[]; + systemBudgets?: BudgetDescription[]; } type JuniorVirtualDashboardOptions = JuniorDashboardOptions; @@ -509,6 +514,7 @@ function dashboardRouteRegistrations(args: { ...args.dashboard, agentName: botConfig.userName, pluginRoutes: args.pluginRoutes, + systemBudgets: describeBudgets(botConfig.budgets), }); if (!app || typeof app.fetch !== "function") { throw new Error("createDashboardApp() must return an app with fetch()"); diff --git a/packages/junior/src/chat/agent/index.ts b/packages/junior/src/chat/agent/index.ts index ffee906492..e2b80e1914 100644 --- a/packages/junior/src/chat/agent/index.ts +++ b/packages/junior/src/chat/agent/index.ts @@ -81,6 +81,7 @@ import { isProviderRetryError, } from "@/chat/services/provider-error"; import { nextProviderRetry } from "@/chat/services/provider-retry"; +import { isBudgetExceededError } from "@/chat/services/budgets"; import { nextEmptyOutputContinuation } from "@/chat/services/empty-output-continuation"; import { getDiscardedRetryUsage } from "@/chat/agent/retry-usage"; import { annotateTurnDeadlineToolResult } from "@/chat/tool-support/turn-deadline-result"; @@ -231,6 +232,7 @@ export async function executeAgentRun( userName: userActor?.userName, userEmail: userActor?.email, runId: request.runId, + turnId: request.turnId, actorType: credentialActor ? "type" in credentialActor ? credentialActor.type @@ -1004,12 +1006,17 @@ async function executeAgentRunInPrivacyContext( // Pi converts prepareNextTurn exceptions into error turns instead of // rejecting. Preserve Junior's yield so runAgentStep can restore it after // the Pi run settles without leaking Pi mechanics into the resume API. + const tracedStreamFn = createTracedStreamFn({ + conversationPrivacy, + ...(streamFn ? { base: streamFn } : {}), + }); + const limitedStreamFn: StreamFn = async (...args) => { + await runResume.startStep(); + return await tracedStreamFn(...args); + }; agent = new Agent({ ...(apiKeyOverride ? { getApiKey: () => apiKeyOverride } : {}), - streamFn: createTracedStreamFn({ - conversationPrivacy, - ...(streamFn ? { base: streamFn } : {}), - }), + streamFn: limitedStreamFn, steeringMode: "all", beforeToolCall: async ({ assistantMessage }) => { const toolCalls = assistantMessage.content.filter( @@ -1427,7 +1434,7 @@ async function executeAgentRunInPrivacyContext( ? { "app.conversation.privacy": conversationPrivacy } : {}), "app.ai.session.conversation_id": conversationId, - "app.ai.turn.session_id": turnId, + "app.ai.turn.id": turnId, ...(currentSliceId ? { "app.ai.turn.slice_id": currentSliceId } : {}), ...toGenAiMessagesTraceAttributes("gen_ai.input", inputMessages), ...(inputMessagesAttribute @@ -1463,6 +1470,7 @@ async function executeAgentRunInPrivacyContext( executionProfile: turnRoute, assistantUserName: botConfig.userName, modelId: activeModelId, + stepCount: runResume.stepCount, }); return { status: "completed", @@ -1510,6 +1518,9 @@ async function executeAgentRunInPrivacyContext( if (isTurnInputCommitLostError(error)) { throw error; } + if (isBudgetExceededError(error)) { + throw error; + } if (error instanceof AuthorizationFlowDisabledError) { throw error; } @@ -1544,6 +1555,7 @@ async function executeAgentRunInPrivacyContext( toolCalls: [], toolResultCount: 0, toolErrorCount: 0, + stepCount: resume?.stepCount ?? 0, usedPrimaryText: false, durationMs: Date.now() - replyStartedAtMs, errorMessage: message, diff --git a/packages/junior/src/chat/agent/resume.ts b/packages/junior/src/chat/agent/resume.ts index 13091b821f..ecb5868f30 100644 --- a/packages/junior/src/chat/agent/resume.ts +++ b/packages/junior/src/chat/agent/resume.ts @@ -39,7 +39,12 @@ import { RetryableDeliveryError, type AgentRunDurability, } from "@/chat/agent/request"; -import { TurnSliceLimitExceededError } from "@/chat/services/turn-limit"; +import { + BudgetExceededError, + checkBudgets, + type TurnBudgets, +} from "@/chat/services/budgets"; +import { reportBudgetExceeded } from "@/chat/services/budget-reporting"; import type { PluginTurnContext } from "@/chat/plugins/prompt"; import type { ConversationPrivacy } from "@/chat/conversation-privacy"; @@ -63,6 +68,7 @@ interface ResumeStateArgs { sessionRecordState: LoadedSessionRecordState; startedAtMs: number; surface: AgentTurnSurface; + turnBudgets?: TurnBudgets; } type AuthPauseOutcome = Extract; @@ -88,7 +94,11 @@ export function createResumeState(args: ResumeStateArgs) { let turnStartMessageIndex: number | undefined; const currentSliceId = args.sessionRecordState.currentSliceId; + const turnBudgets = args.turnBudgets ?? botConfig.budgets; const currentDurationMs = () => Date.now() - args.startedAtMs; + const priorDurationMs = + args.sessionRecordState.existingSessionRecord?.cumulativeDurationMs ?? 0; + let stepCount = args.sessionRecordState.existingSessionRecord?.stepCount ?? 0; const sessionRecordBase = () => ({ channelName: args.channelName, @@ -100,6 +110,7 @@ export function createResumeState(args: ResumeStateArgs) { ...(args.dispatchId ? { dispatchId: args.dispatchId } : {}), source: args.runSource, sessionId: args.turnId, + stepCount, loadedSkillNames: args.getLoadedSkillNames(), modelId: args.getModelId(), ...(args.getReasoningLevel() @@ -119,6 +130,22 @@ export function createResumeState(args: ResumeStateArgs) { get timedOut(): boolean { return timedOut; }, + get stepCount(): number { + return stepCount; + }, + async startStep(): Promise { + const cumulativeRuntimeMs = priorDurationMs + currentDurationMs(); + const exceeded = await checkBudgets(turnBudgets, { + runtimeMs: cumulativeRuntimeMs, + stage: "turn", + steps: stepCount, + }); + if (exceeded) { + await reportBudgetExceeded(exceeded); + throw new BudgetExceededError(exceeded); + } + stepCount += 1; + }, setTurnStartMessageIndex(index: number | undefined): void { turnStartMessageIndex = index; }, @@ -313,7 +340,15 @@ export function createResumeState(args: ResumeStateArgs) { ...(usage ? { usage } : {}), }; } - throw new TurnSliceLimitExceededError(botConfig.maxSlicesPerTurn); + const exceeded = await checkBudgets(botConfig.budgets, { + runtimeMs: sessionRecord.cumulativeDurationMs, + stage: "turn", + steps: sessionRecord.stepCount, + }); + if (!exceeded) { + throw new Error("Turn suspension failed without an exceeded budget"); + } + throw new BudgetExceededError(exceeded); } return undefined; diff --git a/packages/junior/src/chat/config.ts b/packages/junior/src/chat/config.ts index 9c6980ea47..642d0cd242 100644 --- a/packages/junior/src/chat/config.ts +++ b/packages/junior/src/chat/config.ts @@ -12,10 +12,10 @@ import { modelProfileSchema, STANDARD_MODEL_PROFILE, } from "@/chat/model-profile"; +import { readBudgetLimits, type BudgetLimits } from "@/chat/services/budgets"; const MIN_AGENT_TURN_TIMEOUT_MS = 10 * 1000; const DEFAULT_AGENT_TURN_TIMEOUT_MS = 12 * 60 * 1000; -const MAX_SLICES_PER_TURN = 100; const DEFAULT_FUNCTION_MAX_DURATION_SECONDS = 300; const DEFAULT_SLACK_SLASH_COMMAND = "/jr"; const DEFAULT_PROCESSING_REACTION_EMOJI = "eyes"; @@ -44,6 +44,7 @@ const DEFAULT_ASSISTANT_LOADING_MESSAGES = [ ] as const; export interface BotConfig { + budgets: BudgetLimits; contextWindowTokens: number; crossActorMidRunMode: CrossActorMidRunMode; embeddingModelId: string; @@ -54,7 +55,6 @@ export interface BotConfig { profiles: Readonly>; reasoningLevel?: TurnReasoningLevel; visionModelId?: string; - maxSlicesPerTurn: number; turnTimeoutMs: number; userName: string; webSearchModelId: string; @@ -302,6 +302,7 @@ function readBotConfig( validateGatewayModelId(env.AI_HANDOFF_MODEL) ?? DEFAULT_HANDOFF_MODEL_ID; return { + budgets: readBudgetLimits(env, parseOptionalPositiveInteger), userName: toOptionalTrimmed(env.JUNIOR_BOT_NAME) ?? "junior", crossActorMidRunMode: parseCrossActorMidRunMode( env.JUNIOR_CROSS_ACTOR_MID_RUN_MODE, @@ -326,7 +327,6 @@ function readBotConfig( DEFAULT_EMBEDDING_MODEL_ID, loadingMessages: parseLoadingMessages(env.JUNIOR_LOADING_MESSAGES), visionModelId: validateGatewayModelId(env.AI_VISION_MODEL), - maxSlicesPerTurn: MAX_SLICES_PER_TURN, turnTimeoutMs: parseAgentTurnTimeoutMs( env.AGENT_TURN_TIMEOUT_MS, maxTurnTimeoutMs, diff --git a/packages/junior/src/chat/log-context.ts b/packages/junior/src/chat/log-context.ts index c8be01e2aa..44fd2d5ea9 100644 --- a/packages/junior/src/chat/log-context.ts +++ b/packages/junior/src/chat/log-context.ts @@ -14,6 +14,7 @@ export interface LogContext { userName?: string; userEmail?: string; runId?: string; + turnId?: string; actorType?: string; actorId?: string; assistantUserName?: string; @@ -60,6 +61,7 @@ export function logContextToAttributes(context: LogContext): LogAttributes { "enduser.id": context.userId, "enduser.pseudo.id": context.userName, "app.run.id": context.runId, + "app.ai.turn.id": context.turnId, "app.actor.type": context.actorType, "app.actor.id": context.actorId, "gen_ai.agent.name": context.assistantUserName, diff --git a/packages/junior/src/chat/runtime/reply-executor.ts b/packages/junior/src/chat/runtime/reply-executor.ts index cf1d70a3f7..fab01b1d1f 100644 --- a/packages/junior/src/chat/runtime/reply-executor.ts +++ b/packages/junior/src/chat/runtime/reply-executor.ts @@ -78,6 +78,10 @@ import { updateConversationStats, } from "@/chat/services/conversation-memory"; import type { ContextCompactor } from "@/chat/services/context-compaction"; +import { + getBudgetAttributes, + isBudgetExceededError, +} from "@/chat/services/budgets"; import { countPotentialImageAttachments, hasPotentialImageAttachment, @@ -955,6 +959,7 @@ export function createReplyToThread(deps: ReplyExecutorDeps) { } setTags({ conversationId, + turnId, }); if (shouldEmitDevAgentTrace()) { logInfo("agent.turn.started", { @@ -1559,6 +1564,7 @@ export function createReplyToThread(deps: ReplyExecutorDeps) { source, sessionId: turnId, sliceId: 1, + stepCount: reply.diagnostics.stepCount, dispatchOutcome: reply.diagnostics.outcome === "success" ? "completed" @@ -1583,6 +1589,7 @@ export function createReplyToThread(deps: ReplyExecutorDeps) { cumulativeUsage: reply.diagnostics.usage, sessionId: turnId, sliceId: 1, + stepCount: reply.diagnostics.stepCount, startedAtMs: message.metadata.dateSent.getTime(), state: "completed", actor: executionActor, @@ -1733,11 +1740,21 @@ export function createReplyToThread(deps: ReplyExecutorDeps) { } const failureCode = classifiedFailure?.failureCode ?? boundaryFailureCode; + const budgetExceeded = isBudgetExceededError(failureCause); const failureEventId = classifiedFailure?.eventId ?? - logException(failureCause, "slack.turn.execution.failed", { - "app.ai.failure_code": failureCode, - }); + logException( + failureCause, + budgetExceeded + ? "system.budget.exceeded" + : "slack.turn.execution.failed", + { + "app.ai.failure_code": failureCode, + ...(budgetExceeded + ? getBudgetAttributes(failureCause.budget) + : {}), + }, + ); const createdCanvasUrl = getCurrentTurnCanvasUrl({ before: preparedState.artifacts, after: latestArtifacts, diff --git a/packages/junior/src/chat/runtime/slack-resume.ts b/packages/junior/src/chat/runtime/slack-resume.ts index 4a229e1ffd..46905ff940 100644 --- a/packages/junior/src/chat/runtime/slack-resume.ts +++ b/packages/junior/src/chat/runtime/slack-resume.ts @@ -55,9 +55,10 @@ import type { SlackMessageTs } from "@/chat/slack/timestamp"; import { buildAuthPauseResponse } from "@/chat/services/auth-pause-response"; import { getTurnRequestDeadline } from "@/chat/runtime/request-deadline"; import { - TurnSliceLimitExceededError, - buildTurnLimitResponse, -} from "@/chat/services/turn-limit"; + buildBudgetExceededResponse, + getBudgetAttributes, + isBudgetExceededError, +} from "@/chat/services/budgets"; import { coerceThreadConversationState } from "@/chat/state/conversation"; import { hydrateConversationMessages, @@ -242,10 +243,9 @@ async function postResumeFailureReply(args: { await postSlackApiMessage({ channelId: args.channelId, threadTs: args.threadTs, - text: - args.error instanceof TurnSliceLimitExceededError - ? buildTurnLimitResponse(args.eventId) - : buildTurnFailureResponse(args.eventId), + text: isBudgetExceededError(args.error) + ? buildBudgetExceededResponse(args.eventId) + : buildTurnFailureResponse(args.eventId), }); } @@ -258,7 +258,14 @@ async function handleResumeFailure(args: { failureCode: ConversationTurnFailureCode; resumeArgs: ResumeSlackTurnArgs; }): Promise { - const capturedEventId = logException(args.error, args.eventName); + const budgetError = isBudgetExceededError(args.error) + ? args.error + : undefined; + const capturedEventId = logException( + args.error, + budgetError ? "system.budget.exceeded" : args.eventName, + budgetError ? getBudgetAttributes(budgetError.budget) : {}, + ); const eventId = requireTurnFailureEventId(capturedEventId, args.eventName); let failureStatePersistError: unknown; try { @@ -753,6 +760,7 @@ async function resumeSlackTurnInContext( actor: resumeActor, surface: replyContext.routing.surface ?? "slack", sliceId: runArgs.sliceId, + stepCount: reply.diagnostics.stepCount, }); } else if (replyContext.routing.dispatch?.id) { await recordAgentTurnSessionSummary({ @@ -767,6 +775,9 @@ async function resumeSlackTurnInContext( : {}), sessionId: runArgs.turnId, sliceId: runArgs.sliceId ?? 1, + stepCount: reply.diagnostics.stepCount, + currentDurationMs: reply.diagnostics.durationMs, + currentUsage: reply.diagnostics.usage, source: replyContext.routing.source, state: reply.diagnostics.outcome === "success" ? "completed" : "failed", diff --git a/packages/junior/src/chat/services/budget-reporting.ts b/packages/junior/src/chat/services/budget-reporting.ts new file mode 100644 index 0000000000..cb26b402d8 --- /dev/null +++ b/packages/junior/src/chat/services/budget-reporting.ts @@ -0,0 +1,29 @@ +import { logWarn } from "@/chat/logging"; +import { incrementStat } from "@/stats"; +import { + getBudgetAttributes, + type BudgetExceeded, +} from "@/chat/services/budgets"; + +/** Record one exceeded budget without allowing reporting failure to affect work. */ +export async function reportBudgetExceeded( + budget: BudgetExceeded, +): Promise { + logWarn("system.budget.exceeded", getBudgetAttributes(budget)); + if (budget.outcome !== "stop" || !process.env.DATABASE_URL) { + return; + } + try { + await incrementStat({ + namespace: "junior", + metric: "budget_exceeded", + name: budget.name, + }); + } catch (error) { + logWarn("system.budget.stat.failed", { + "app.budget.name": budget.name, + "exception.message": + error instanceof Error ? error.message : String(error), + }); + } +} diff --git a/packages/junior/src/chat/services/budgets.ts b/packages/junior/src/chat/services/budgets.ts new file mode 100644 index 0000000000..7d548f7bdb --- /dev/null +++ b/packages/junior/src/chat/services/budgets.ts @@ -0,0 +1,209 @@ +type MaybePromise = T | Promise; + +export type BudgetStage = "conversation_admission" | "turn"; +export type BudgetOutcome = "queue" | "stop"; +export type BudgetUnit = "count" | "milliseconds" | "usd"; + +export type BudgetContext = + | { + activeConversations: number; + activeConversationsForUser?: number; + stage: "conversation_admission"; + } + | { + runtimeMs: number; + stage: "turn"; + steps: number; + }; + +interface InternalBudget { + defaultLimit: number; + description: string; + envName: string; + label: string; + measure(context: BudgetContext): MaybePromise; + name: Name; + outcome: BudgetOutcome; + stage: BudgetStage; + unit: BudgetUnit; +} + +function defineBudget( + budget: InternalBudget, +): InternalBudget { + return budget; +} + +export const internalBudgets = [ + defineBudget({ + defaultLimit: 100, + description: "Queues additional conversations.", + envName: "JUNIOR_MAX_ACTIVE_CONVERSATIONS", + label: "Active globally", + measure: (context) => + context.stage === "conversation_admission" + ? context.activeConversations + : undefined, + name: "active_conversations_global", + outcome: "queue", + stage: "conversation_admission", + unit: "count", + }), + defineBudget({ + defaultLimit: 5, + description: "Applies when a stable user ID is available.", + envName: "JUNIOR_MAX_ACTIVE_CONVERSATIONS_PER_USER", + label: "Active per user", + measure: (context) => + context.stage === "conversation_admission" + ? context.activeConversationsForUser + : undefined, + name: "active_conversations_user", + outcome: "queue", + stage: "conversation_admission", + unit: "count", + }), + defineBudget({ + defaultLimit: 21_600_000, + description: "Cumulative active time across resumes.", + envName: "JUNIOR_MAX_TURN_RUNTIME_MS", + label: "Runtime per turn", + measure: (context) => + context.stage === "turn" ? context.runtimeMs : undefined, + name: "turn_runtime", + outcome: "stop", + stage: "turn", + unit: "milliseconds", + }), + defineBudget({ + defaultLimit: 500, + description: "Stops runaway model and tool loops.", + envName: "JUNIOR_MAX_STEPS_PER_TURN", + label: "Agent steps per turn", + measure: (context) => + context.stage === "turn" ? context.steps : undefined, + name: "turn_steps", + outcome: "stop", + stage: "turn", + unit: "count", + }), +] as const; + +export type BudgetName = (typeof internalBudgets)[number]["name"]; +export type BudgetLimits = Record; +export type ConversationAdmissionBudgets = Pick< + BudgetLimits, + "active_conversations_global" | "active_conversations_user" +>; +export type TurnBudgets = Pick; + +export interface BudgetDescription { + description: string; + label: string; + limit: number; + name: BudgetName; + outcome: BudgetOutcome; + stage: BudgetStage; + unit: BudgetUnit; +} + +export interface BudgetExceeded { + limit: number; + name: BudgetName; + outcome: "queue" | "stop"; + value: number; +} + +interface BudgetLimitParser { + (envName: string, rawValue: string | undefined): number | undefined; +} + +/** Parse configured budget limits from the internal registry. */ +export function readBudgetLimits( + env: NodeJS.ProcessEnv, + parse: BudgetLimitParser, +): BudgetLimits { + return Object.fromEntries( + internalBudgets.map((budget) => [ + budget.name, + parse(budget.envName, env[budget.envName]) ?? budget.defaultLimit, + ]), + ) as BudgetLimits; +} + +/** Return safe configured budget descriptions for reporting and UI surfaces. */ +export function describeBudgets(limits: BudgetLimits): BudgetDescription[] { + return internalBudgets.map((budget) => ({ + description: budget.description, + label: budget.label, + limit: limits[budget.name], + name: budget.name, + outcome: budget.outcome, + stage: budget.stage, + unit: budget.unit, + })); +} + +/** Return the first exceeded internal budget for the supplied runtime context. */ +export async function checkBudgets( + limits: Partial, + context: BudgetContext, +): Promise { + for (const budget of internalBudgets) { + if (budget.stage !== context.stage) { + continue; + } + const limit = limits[budget.name]; + if (limit === undefined) { + continue; + } + const value = await budget.measure(context); + if (value !== undefined && value >= limit) { + return { + limit, + name: budget.name, + outcome: budget.outcome, + value, + }; + } + } + return undefined; +} + +/** Terminal failure carrying the budget decision that stopped a turn. */ +export class BudgetExceededError extends Error { + constructor(readonly budget: BudgetExceeded) { + super( + `System budget exceeded: ${budget.name} (${budget.value}/${budget.limit})`, + ); + this.name = "BudgetExceededError"; + } +} + +/** Return whether an error carries a terminal system budget decision. */ +export function isBudgetExceededError( + error: unknown, +): error is BudgetExceededError { + return error instanceof BudgetExceededError; +} + +/** Return stable telemetry attributes for one budget decision. */ +export function getBudgetAttributes( + budget: BudgetExceeded, +): Record { + return { + "app.budget.limit": budget.limit, + "app.budget.name": budget.name, + "app.budget.outcome": budget.outcome, + "app.budget.value": budget.value, + }; +} + +/** Explain a terminal turn budget with actionable recovery guidance. */ +export function buildBudgetExceededResponse(eventId: string): string { + return ( + "I couldn't finish this request because this turn reached a system budget. " + + "Please try again with a smaller or more specific request. " + + `Reference: \`event_id=${eventId}\`.` + ); +} diff --git a/packages/junior/src/chat/services/turn-limit.ts b/packages/junior/src/chat/services/turn-limit.ts deleted file mode 100644 index 155f8c2500..0000000000 --- a/packages/junior/src/chat/services/turn-limit.ts +++ /dev/null @@ -1,16 +0,0 @@ -/** Terminal failure raised when one agent turn exhausts its slice budget. */ -export class TurnSliceLimitExceededError extends Error { - constructor(maxSlices: number) { - super(`Agent turn exceeded execution limit (${maxSlices} slices)`); - this.name = "TurnSliceLimitExceededError"; - } -} - -/** Explain a terminal turn execution limit with actionable recovery guidance. */ -export function buildTurnLimitResponse(eventId: string): string { - return ( - "I couldn't finish this request because this turn reached its execution limit. " + - "Please try again with a smaller or more specific request. " + - `Reference: \`event_id=${eventId}\`.` - ); -} diff --git a/packages/junior/src/chat/services/turn-result.ts b/packages/junior/src/chat/services/turn-result.ts index 756c1bf059..8c9245bc6d 100644 --- a/packages/junior/src/chat/services/turn-result.ts +++ b/packages/junior/src/chat/services/turn-result.ts @@ -29,6 +29,7 @@ export interface AgentTurnDiagnostics { outcome: "success" | "execution_failure" | "provider_error"; reasoningLevel?: TurnRoute["reasoningLevel"]; stopReason?: string; + stepCount?: number; toolCalls: string[]; toolErrorCount: number; toolResultCount: number; @@ -59,6 +60,7 @@ export interface TurnResultInput { executionProfile: TurnRoute; assistantUserName?: string; modelId: string; + stepCount?: number; } /** Process raw agent messages into a structured AgentRunResult. */ @@ -73,6 +75,7 @@ export function buildTurnResult(input: TurnResultInput): AgentRunResult { usage, executionProfile, modelId, + stepCount, } = input; const toolResults = newMessages.filter(isToolResultMessage); @@ -184,6 +187,7 @@ export function buildTurnResult(input: TurnResultInput): AgentRunResult { durationMs, usage, stopReason, + stepCount, errorMessage, providerError: resolvedOutcome === "provider_error" && errorMessage diff --git a/packages/junior/src/chat/services/turn-session-record.ts b/packages/junior/src/chat/services/turn-session-record.ts index 7c8f82693b..92329f96f6 100644 --- a/packages/junior/src/chat/services/turn-session-record.ts +++ b/packages/junior/src/chat/services/turn-session-record.ts @@ -17,7 +17,8 @@ import { } from "@/chat/pi/transcript"; import { addAgentTurnUsage, type AgentTurnUsage } from "@/chat/usage"; import { persistWithRetry } from "@/chat/services/persist-retry"; -import { TurnSliceLimitExceededError } from "@/chat/services/turn-limit"; +import { BudgetExceededError, checkBudgets } from "@/chat/services/budgets"; +import { reportBudgetExceeded } from "@/chat/services/budget-reporting"; import { botConfig } from "@/chat/config"; import type { PluginTurnContext } from "@/chat/plugins/prompt"; @@ -107,6 +108,7 @@ export async function persistRunningSessionRecord(args: { source?: Source; sessionId: string; sliceId: number; + stepCount?: number; messages: PiMessage[]; /** Provenance for trailing newly committed messages, such as steering. */ trailingMessageProvenance?: ConversationMessageProvenance[]; @@ -146,6 +148,7 @@ export async function persistRunningSessionRecord(args: { : {}), sessionId: args.sessionId, sliceId: args.sliceId, + stepCount: args.stepCount ?? latestSessionRecord?.stepCount, state: "running", piMessages: args.messages, ...(args.trailingMessageProvenance @@ -215,6 +218,7 @@ export async function persistCompletedSessionRecord(args: { sessionId: string; /** Defaults to the latest stored slice when the deliverer does not know it. */ sliceId?: number; + stepCount?: number; allMessages: PiMessage[]; loadedSkillNames?: string[]; modelId: string; @@ -281,6 +285,7 @@ export async function persistCompletedSessionRecord(args: { : {}), sessionId: args.sessionId, sliceId, + stepCount: args.stepCount ?? latestSessionRecord?.stepCount, state: "completed", piMessages: args.allMessages, ...((args.surface ?? latestSessionRecord?.surface) @@ -332,6 +337,7 @@ export async function completeDeliveredTurn(args: { resultMessageId?: string; sessionId: string; sliceId: number; + stepCount?: number; source: Source; surface: AgentTurnSurface; turnStartMessageIndex?: number; @@ -351,6 +357,7 @@ export async function completeDeliveredTurn(args: { source: args.source, sessionId: args.sessionId, sliceId: args.sliceId, + stepCount: args.stepCount, allMessages: args.messages, loadedSkillNames: args.loadedSkillNames, modelId: args.modelId, @@ -370,6 +377,7 @@ export async function persistAuthPauseSessionRecord(args: { conversationId: string; sessionId: string; currentSliceId: number; + stepCount?: number; currentDurationMs?: number; currentUsage?: AgentTurnUsage; destination?: Destination; @@ -422,6 +430,7 @@ export async function persistAuthPauseSessionRecord(args: { : {}), sessionId: args.sessionId, sliceId: nextSliceId, + stepCount: args.stepCount ?? latestSessionRecord?.stepCount, state: "awaiting_resume", piMessages, ...((args.surface ?? latestSessionRecord?.surface) @@ -466,6 +475,7 @@ interface ContinuationRecordInput { conversationId: string; sessionId: string; currentSliceId: number; + stepCount?: number; currentDurationMs?: number; currentUsage?: AgentTurnUsage; destination?: Destination; @@ -481,7 +491,7 @@ interface ContinuationRecordInput { surface?: AgentTurnSurface; } -/** Persist a timeout or delivery retry under the turn's shared slice limit. */ +/** Persist a timeout or delivery retry under the turn's runtime limit. */ export async function persistContinuationSessionRecord( args: ContinuationRecordInput & { resumeReason: "retry" | "timeout"; @@ -501,15 +511,22 @@ export async function persistContinuationSessionRecord( if (piMessages.length === 0 || !isContinuablePiBoundary(piMessages)) { return undefined; } - const cumulativeDurationMs = addDurationMs( - latestSessionRecord?.cumulativeDurationMs, - args.currentDurationMs, - ); + const cumulativeDurationMs = + addDurationMs( + latestSessionRecord?.cumulativeDurationMs, + args.currentDurationMs, + ) ?? 0; const cumulativeUsage = addAgentTurnUsage( latestSessionRecord?.cumulativeUsage, args.currentUsage, ); - if (nextSliceId > botConfig.maxSlicesPerTurn) { + const exceeded = await checkBudgets(botConfig.budgets, { + runtimeMs: cumulativeDurationMs, + stage: "turn", + steps: args.stepCount ?? latestSessionRecord?.stepCount ?? 0, + }); + if (exceeded) { + await reportBudgetExceeded(exceeded); return await upsertAgentTurnSessionRecord({ ...((args.channelName ?? latestSessionRecord?.channelName) ? { @@ -533,6 +550,7 @@ export async function persistContinuationSessionRecord( : {}), sessionId: args.sessionId, sliceId: args.currentSliceId, + stepCount: args.stepCount ?? latestSessionRecord?.stepCount, state: "failed", piMessages, ...((args.surface ?? latestSessionRecord?.surface) @@ -550,9 +568,7 @@ export async function persistContinuationSessionRecord( : {}), resumeReason: args.resumeReason, resumedFromSliceId: latestSessionRecord?.resumedFromSliceId, - errorMessage: new TurnSliceLimitExceededError( - botConfig.maxSlicesPerTurn, - ).message, + errorMessage: new BudgetExceededError(exceeded).message, ...((args.actor ?? latestSessionRecord?.actor) ? { actor: args.actor ?? latestSessionRecord?.actor } : {}), @@ -580,6 +596,7 @@ export async function persistContinuationSessionRecord( : {}), sessionId: args.sessionId, sliceId: nextSliceId, + stepCount: args.stepCount ?? latestSessionRecord?.stepCount, state: "awaiting_resume", piMessages, ...((args.surface ?? latestSessionRecord?.surface) @@ -627,6 +644,7 @@ export async function persistYieldSessionRecord(args: { conversationId: string; sessionId: string; currentSliceId: number; + stepCount?: number; currentDurationMs?: number; currentUsage?: AgentTurnUsage; destination?: Destination; @@ -675,6 +693,7 @@ export async function persistYieldSessionRecord(args: { : {}), sessionId: args.sessionId, sliceId: args.currentSliceId, + stepCount: args.stepCount ?? latestSessionRecord?.stepCount, state: "awaiting_resume", piMessages, ...((args.surface ?? latestSessionRecord?.surface) diff --git a/packages/junior/src/chat/state/turn-session.ts b/packages/junior/src/chat/state/turn-session.ts index d248c3b8ec..46c59f5c61 100644 --- a/packages/junior/src/chat/state/turn-session.ts +++ b/packages/junior/src/chat/state/turn-session.ts @@ -29,10 +29,18 @@ import { } from "@/chat/conversations/projection"; import type { PluginTurnContext } from "@/chat/plugins/prompt"; import { projectConversationEvents } from "@/chat/pi/conversation-events"; -import { agentTurnUsageSchema, type AgentTurnUsage } from "@/chat/usage"; +import { + addAgentTurnUsage, + agentTurnUsageSchema, + type AgentTurnUsage, +} from "@/chat/usage"; import { getStateAdapter } from "./adapter"; import { getConversationEventStore, getConversationStore } from "@/chat/db"; -import { logWarn } from "@/chat/logging"; +import { + extractGenAiUsageAttributes, + logWarn, + setSpanAttributes, +} from "@/chat/logging"; import { retainRuntimeTurnContext, stripRuntimeTurnContext, @@ -103,6 +111,7 @@ export interface AgentTurnSessionRecord { resumedFromSliceId?: number; sessionId: string; sliceId: number; + stepCount: number; startedAtMs: number; state: AgentTurnSessionStatus; surface?: AgentTurnSurface; @@ -186,6 +195,7 @@ const agentTurnSessionSummarySchema = z resumedFromSliceId: z.number().int().nonnegative().optional(), sessionId: z.string().min(1), sliceId: z.number().int().nonnegative(), + stepCount: z.number().int().nonnegative().default(0), startedAtMs: nonNegativeNumberSchema, state: agentTurnSessionStatusSchema, surface: agentTurnSurfaceSchema.optional(), @@ -235,6 +245,51 @@ function parseAgentTurnSessionSummary(value: unknown): AgentTurnSessionSummary { return agentTurnSessionSummarySchema.parse(value); } +type TurnTelemetrySummary = Pick< + AgentTurnSessionSummary, + | "conversationId" + | "cumulativeDurationMs" + | "cumulativeUsage" + | "modelId" + | "resumeReason" + | "sessionId" + | "sliceId" + | "state" + | "stepCount" + | "surface" +>; + +/** Return OTEL-aligned and Junior-specific attributes for one durable turn. */ +export function getTurnTelemetryAttributes( + summary: TurnTelemetrySummary, +): Record { + return { + "gen_ai.conversation.id": summary.conversationId, + ...(summary.modelId ? { "gen_ai.request.model": summary.modelId } : {}), + ...extractGenAiUsageAttributes(summary.cumulativeUsage), + "app.ai.turn.id": summary.sessionId, + "app.ai.turn.runtime_ms": summary.cumulativeDurationMs, + "app.ai.turn.slice_id": summary.sliceId, + "app.ai.turn.state": summary.state, + "app.ai.turn.step_count": summary.stepCount, + ...(summary.resumeReason + ? { "app.ai.turn.resume_reason": summary.resumeReason } + : {}), + ...(summary.surface ? { "app.ai.turn.surface": summary.surface } : {}), + }; +} + +function recordTerminalTurnTelemetry(summary: TurnTelemetrySummary): void { + if ( + summary.state !== "completed" && + summary.state !== "failed" && + summary.state !== "abandoned" + ) { + return; + } + setSpanAttributes(getTurnTelemetryAttributes(summary)); +} + async function appendAgentTurnSessionSummary( summary: AgentTurnSessionSummary, ttlMs: number, @@ -314,6 +369,7 @@ function materializeAgentTurnSessionRecord( conversationId: stored.conversationId, sessionId: stored.sessionId, sliceId: stored.sliceId, + stepCount: stored.stepCount, state: stored.state, startedAtMs: stored.startedAtMs, lastProgressAtMs: stored.lastProgressAtMs, @@ -497,6 +553,7 @@ function buildStoredRecord(args: { actors?: Actor[]; sessionId: string; sliceId: number; + stepCount?: number; startedAtMs?: number; state: AgentTurnSessionStatus; surface?: AgentTurnSurface; @@ -514,6 +571,7 @@ function buildStoredRecord(args: { conversationId: args.conversationId, sessionId: args.sessionId, sliceId: args.sliceId, + stepCount: args.stepCount ?? 0, state: args.state, startedAtMs: args.startedAtMs ?? nowMs, lastProgressAtMs: args.lastProgressAtMs ?? nowMs, @@ -586,6 +644,7 @@ async function setStoredRecord(args: { ...summary } = args.record; await appendAgentTurnSessionSummary(summary, args.ttlMs); + recordTerminalTurnTelemetry(summary); return materializeAgentTurnSessionRecord( args.record, { @@ -699,6 +758,7 @@ export async function upsertAgentTurnSessionRecord(args: { conversationStore?: ConversationStore; sessionId: string; sliceId: number; + stepCount?: number; state: AgentTurnSessionStatus; surface?: AgentTurnSurface; piMessages: PiMessage[]; @@ -798,6 +858,7 @@ export async function upsertAgentTurnSessionRecord(args: { conversationId: args.conversationId, sessionId: args.sessionId, sliceId: args.sliceId, + stepCount: args.stepCount ?? existingRecord?.stepCount ?? 0, state: args.state, ...(existingRecord?.startedAtMs !== undefined ? { startedAtMs: existingRecord.startedAtMs } @@ -877,6 +938,8 @@ export async function recordAgentTurnSessionSummary(args: { conversationId: string; cumulativeDurationMs?: number; cumulativeUsage?: AgentTurnUsage; + currentDurationMs?: number; + currentUsage?: AgentTurnUsage; destination?: Destination; dispatchId?: string; dispatchOutcome?: AgentDispatchOutcome; @@ -893,6 +956,7 @@ export async function recordAgentTurnSessionSummary(args: { reasoningLevel?: string; sessionId: string; sliceId: number; + stepCount?: number; startedAtMs?: number; state: AgentTurnSessionStatus; surface?: AgentTurnSurface; @@ -923,6 +987,14 @@ export async function recordAgentTurnSessionSummary(args: { } const nowMs = Date.now(); const ttlMs = Math.max(1, args.ttlMs ?? AGENT_TURN_SESSION_TTL_MS); + const cumulativeDurationMs = + args.cumulativeDurationMs ?? + (args.currentDurationMs !== undefined + ? (existing?.cumulativeDurationMs ?? 0) + args.currentDurationMs + : (existing?.cumulativeDurationMs ?? 0)); + const cumulativeUsage = + args.cumulativeUsage ?? + addAgentTurnUsage(existing?.cumulativeUsage, args.currentUsage); const summary: AgentTurnSessionSummary = { version: existing?.version ?? 0, ...((args.channelName ?? existing?.channelName) @@ -931,15 +1003,13 @@ export async function recordAgentTurnSessionSummary(args: { conversationId: args.conversationId, sessionId: args.sessionId, sliceId: args.sliceId, + stepCount: args.stepCount ?? existing?.stepCount ?? 0, startedAtMs: existing?.startedAtMs ?? args.startedAtMs ?? nowMs, lastProgressAtMs: args.lastProgressAtMs ?? nowMs, state: args.state, updatedAtMs: nowMs, - cumulativeDurationMs: - args.cumulativeDurationMs ?? existing?.cumulativeDurationMs ?? 0, - ...((args.cumulativeUsage ?? existing?.cumulativeUsage) - ? { cumulativeUsage: args.cumulativeUsage ?? existing?.cumulativeUsage } - : {}), + cumulativeDurationMs, + ...(cumulativeUsage ? { cumulativeUsage } : {}), ...((args.destination ?? existing?.destination) ? { destination: args.destination ?? existing?.destination } : {}), @@ -984,6 +1054,7 @@ export async function recordAgentTurnSessionSummary(args: { summary, }); await appendAgentTurnSessionSummary(summary, ttlMs); + recordTerminalTurnTelemetry(summary); } async function readAgentTurnSessionSummariesFromIndex( diff --git a/packages/junior/src/chat/task-execution/state.ts b/packages/junior/src/chat/task-execution/state.ts index aa80652688..bc4e38b649 100644 --- a/packages/junior/src/chat/task-execution/state.ts +++ b/packages/junior/src/chat/task-execution/state.ts @@ -12,7 +12,12 @@ import type { Lock, StateAdapter } from "chat"; import { destinationSchema, type Destination } from "@sentry/junior-plugin-api"; import { z } from "zod"; import { isRecord, toOptionalNumber, toOptionalString } from "@/chat/coerce"; -import { getChatConfig } from "@/chat/config"; +import { botConfig, getChatConfig } from "@/chat/config"; +import { + checkBudgets, + type BudgetExceeded, + type ConversationAdmissionBudgets, +} from "@/chat/services/budgets"; import { parseDestination, sameDestination } from "@/chat/destination"; import { parseStoredSlackActor, type StoredSlackActor } from "@/chat/actor"; import { @@ -27,6 +32,11 @@ const CONVERSATION_ACTIVITY_INDEX_MAX_LENGTH = 10_000; const CONVERSATION_INDEX_LOCK_TTL_MS = 10_000; const CONVERSATION_INDEX_LOCK_WAIT_MS = 2_000; const CONVERSATION_INDEX_LOCK_RETRY_MS = 25; +const CONVERSATION_ADMISSION_LOCK_KEY = `${CONVERSATION_PREFIX}:admission`; +const CONVERSATION_ADMISSION_STATE_KEY = `${CONVERSATION_PREFIX}:admission:leases`; +const CONVERSATION_ADMISSION_LOCK_TTL_MS = 30_000; +const CONVERSATION_ADMISSION_LOCK_WAIT_MS = 10_000; +const CONVERSATION_ADMISSION_LOCK_RETRY_MS = 25; const CONVERSATION_MUTATION_LOCK_TTL_MS = 10_000; const CONVERSATION_MUTATION_WAIT_MS = 10_000; const CONVERSATION_MUTATION_RETRY_MS = 25; @@ -127,6 +137,7 @@ export interface Lease { } export interface ConversationExecution { + activeUserKey?: string; inboundMessageIds: string[]; lastCheckpointAtMs?: number; lastEnqueuedAtMs?: number; @@ -167,6 +178,8 @@ export interface ConversationWorkState extends Conversation { } export interface StartConversationWorkAcquired { + activeConversationCount: number; + activeUserConversationCount?: number; leaseExpiresAtMs: number; leaseToken: string; status: "acquired"; @@ -181,9 +194,17 @@ export interface StartConversationWorkNoWork { status: "no_work"; } +export interface StartConversationWorkLimited { + activeConversationCount: number; + activeUserConversationCount?: number; + budget: BudgetExceeded; + status: "limited"; +} + export type StartConversationWorkResult = | StartConversationWorkAcquired | StartConversationWorkActive + | StartConversationWorkLimited | StartConversationWorkNoWork; export interface AppendInboundMessageResult { @@ -203,6 +224,13 @@ interface ConversationIndexEntry { score: number; } +interface ConversationAdmissionEntry { + conversationId: string; + expiresAtMs: number; + leaseToken: string; + userKey?: string; +} + interface ConversationIndexStore { list(args: { indexKey: string; @@ -432,6 +460,7 @@ function normalizeExecution( pendingCount: pendingMessages.length, pendingMessages, lease, + activeUserKey: toOptionalString(value.activeUserKey), lastCheckpointAtMs: toOptionalNumber(value.lastCheckpointAtMs), lastEnqueuedAtMs: toOptionalNumber(value.lastEnqueuedAtMs), runId: toOptionalString(value.runId), @@ -636,6 +665,32 @@ async function withIndexLock( } } +async function withConversationAdmission( + state: StateAdapter, + callback: (lock: Lock) => Promise, +): Promise { + const startedAtMs = now(); + let lock: Lock | null; + while (true) { + lock = await state.acquireLock( + CONVERSATION_ADMISSION_LOCK_KEY, + CONVERSATION_ADMISSION_LOCK_TTL_MS, + ); + if (lock) { + break; + } + if (now() - startedAtMs >= CONVERSATION_ADMISSION_LOCK_WAIT_MS) { + throw new Error("Could not acquire conversation admission lock"); + } + await sleep(CONVERSATION_ADMISSION_LOCK_RETRY_MS); + } + try { + return await callback(lock); + } finally { + await state.releaseLock(lock); + } +} + function normalizeIndexEntry( value: unknown, ): ConversationIndexEntry | undefined { @@ -939,6 +994,124 @@ async function readConversation( return conversation; } +function normalizeConversationAdmissionEntry( + value: unknown, +): ConversationAdmissionEntry | undefined { + if (!isRecord(value)) { + return undefined; + } + const conversationId = toOptionalString(value.conversationId); + const expiresAtMs = toOptionalNumber(value.expiresAtMs); + const leaseToken = toOptionalString(value.leaseToken); + if (!conversationId || expiresAtMs === undefined || !leaseToken) { + return undefined; + } + return { + conversationId, + expiresAtMs, + leaseToken, + ...(toOptionalString(value.userKey) + ? { userKey: toOptionalString(value.userKey) } + : {}), + }; +} + +async function readConversationAdmissions( + state: StateAdapter, + nowMs: number, +): Promise { + const value = await state.get(CONVERSATION_ADMISSION_STATE_KEY); + if (!Array.isArray(value)) { + return []; + } + const entries = new Map(); + for (const item of value) { + const entry = normalizeConversationAdmissionEntry(item); + if (!entry || entry.expiresAtMs <= nowMs) { + continue; + } + let conversation: Conversation | undefined; + try { + conversation = await readConversation(state, entry.conversationId); + } catch (error) { + if (!(error instanceof InvalidConversationRecordError)) { + throw error; + } + continue; + } + if (!conversation) { + continue; + } + const lease = conversation.execution.lease; + if ( + !lease || + lease.token !== entry.leaseToken || + lease.expiresAtMs <= nowMs + ) { + continue; + } + const activeEntry: ConversationAdmissionEntry = { + conversationId: entry.conversationId, + expiresAtMs: lease.expiresAtMs, + leaseToken: lease.token, + ...(conversation.execution.activeUserKey + ? { userKey: conversation.execution.activeUserKey } + : {}), + }; + const existing = entries.get(entry.conversationId); + if (!existing || activeEntry.expiresAtMs > existing.expiresAtMs) { + entries.set(entry.conversationId, activeEntry); + } + } + return [...entries.values()]; +} + +async function writeConversationAdmissions( + state: StateAdapter, + entries: ConversationAdmissionEntry[], +): Promise { + if (entries.length === 0) { + await state.delete(CONVERSATION_ADMISSION_STATE_KEY); + return; + } + await state.set( + CONVERSATION_ADMISSION_STATE_KEY, + entries, + JUNIOR_THREAD_STATE_TTL_MS, + ); +} + +function conversationAdmissionOccupancy(args: { + entries: ConversationAdmissionEntry[]; + userKey?: string; +}): { + activeConversationCount: number; + activeUserConversationCount?: number; +} { + return { + activeConversationCount: args.entries.length, + ...(args.userKey + ? { + activeUserConversationCount: args.entries.filter( + (entry) => entry.userKey === args.userKey, + ).length, + } + : {}), + }; +} + +async function removeConversationAdmissionWhileLocked(args: { + conversationId: string; + nowMs: number; + state: StateAdapter; +}): Promise { + const entries = await readConversationAdmissions(args.state, args.nowMs); + await writeConversationAdmissions( + args.state, + entries.filter((entry) => entry.conversationId !== args.conversationId), + ); +} + /** * Persist a conversation and refresh its reporting and active-recovery indexes. * @@ -994,6 +1167,22 @@ async function writeConversation( }); } +async function writeConversationAndReleaseAdmission(args: { + conversation: Conversation; + lock: Lock; + nowMs: number; + state: StateAdapter; +}): Promise { + await withConversationAdmission(args.state, async () => { + await writeConversation(args.state, args.lock, args.conversation); + await removeConversationAdmissionWhileLocked({ + conversationId: args.conversation.conversationId, + nowMs: args.nowMs, + state: args.state, + }); + }); +} + function assertSameConversationDestination(args: { conversationId: string; current: Destination | undefined; @@ -1394,6 +1583,8 @@ export async function clearConsumedConversationWake(args: { /** Try to acquire the durable execution lease for one conversation. */ export async function startConversationWork(args: { + activeUserKey?: string; + budgets?: ConversationAdmissionBudgets; conversationId: string; nowMs?: number; state?: StateAdapter; @@ -1414,35 +1605,87 @@ export async function startConversationWork(args: { return { status: "no_work" }; } - const lease: Lease = { - token: randomUUID(), - acquiredAtMs: nowMs, - lastCheckInAtMs: nowMs, - expiresAtMs: nowMs + CONVERSATION_WORK_LEASE_TTL_MS, - }; - await writeConversation( - state, - lock, - withExecutionUpdate( - current, + return await withConversationAdmission(state, async (admissionLock) => { + const budgets = args.budgets ?? botConfig.budgets; + const entries = (await readConversationAdmissions(state, nowMs)).filter( + (entry) => entry.conversationId !== args.conversationId, + ); + const occupancy = conversationAdmissionOccupancy({ + entries, + userKey: args.activeUserKey, + }); + const exceeded = await checkBudgets(budgets, { + activeConversations: occupancy.activeConversationCount, + ...(args.activeUserKey + ? { + activeConversationsForUser: + occupancy.activeUserConversationCount ?? 0, + } + : {}), + stage: "conversation_admission", + }); + if (exceeded) { + return { + ...occupancy, + budget: exceeded, + status: "limited", + }; + } + + const fenced = await state.extendLock( + admissionLock, + CONVERSATION_ADMISSION_LOCK_TTL_MS, + ); + if (!fenced) { + throw new Error("Conversation admission lock was lost before write"); + } + const lease: Lease = { + token: randomUUID(), + acquiredAtMs: nowMs, + lastCheckInAtMs: nowMs, + expiresAtMs: nowMs + CONVERSATION_WORK_LEASE_TTL_MS, + }; + await writeConversationAdmissions(state, [ + ...entries, { - ...current.execution, - lease, - status: - current.execution.status === "awaiting_resume" - ? "awaiting_resume" - : "running", - runId: current.execution.runId ?? randomUUID(), - lastEnqueuedAtMs: undefined, + conversationId: args.conversationId, + expiresAtMs: lease.expiresAtMs, + leaseToken: lease.token, + ...(args.activeUserKey ? { userKey: args.activeUserKey } : {}), }, - nowMs, - ), - ); - return { - status: "acquired", - leaseToken: lease.token, - leaseExpiresAtMs: lease.expiresAtMs, - }; + ]); + await writeConversation( + state, + lock, + withExecutionUpdate( + current, + { + ...current.execution, + activeUserKey: args.activeUserKey, + lease, + status: + current.execution.status === "awaiting_resume" + ? "awaiting_resume" + : "running", + runId: current.execution.runId ?? randomUUID(), + lastEnqueuedAtMs: undefined, + }, + nowMs, + ), + ); + return { + status: "acquired", + activeConversationCount: occupancy.activeConversationCount + 1, + ...(args.activeUserKey + ? { + activeUserConversationCount: + (occupancy.activeUserConversationCount ?? 0) + 1, + } + : {}), + leaseToken: lease.token, + leaseExpiresAtMs: lease.expiresAtMs, + }; + }); }); } @@ -1456,25 +1699,43 @@ export async function checkInConversationWork(args: { const nowMs = args.nowMs ?? now(); return await withConversationMutation(args, async (state, lock) => { const current = await readConversation(state, args.conversationId); - if (!current || current.execution.lease?.token !== args.leaseToken) { + const lease = current?.execution.lease; + if (!current || !lease || lease.token !== args.leaseToken) { return false; } - await writeConversation( - state, - lock, - withExecutionUpdate( - current, + const expiresAtMs = nowMs + CONVERSATION_WORK_LEASE_TTL_MS; + await withConversationAdmission(state, async () => { + const entries = (await readConversationAdmissions(state, nowMs)).filter( + (entry) => entry.conversationId !== args.conversationId, + ); + await writeConversationAdmissions(state, [ + ...entries, { - ...current.execution, - lease: { - ...current.execution.lease, - lastCheckInAtMs: nowMs, - expiresAtMs: nowMs + CONVERSATION_WORK_LEASE_TTL_MS, - }, + conversationId: args.conversationId, + expiresAtMs, + leaseToken: lease.token, + ...(current.execution.activeUserKey + ? { userKey: current.execution.activeUserKey } + : {}), }, - nowMs, - ), - ); + ]); + await writeConversation( + state, + lock, + withExecutionUpdate( + current, + { + ...current.execution, + lease: { + ...lease, + lastCheckInAtMs: nowMs, + expiresAtMs, + }, + }, + nowMs, + ), + ); + }); return true; }); } @@ -1683,13 +1944,18 @@ export async function releaseConversationWork(args: { if (!current || current.execution.lease?.token !== args.leaseToken) { return false; } - await writeConversation( + await writeConversationAndReleaseAdmission({ state, lock, - withExecutionUpdate( + nowMs, + conversation: withExecutionUpdate( current, { ...current.execution, + activeUserKey: + current.execution.status === "awaiting_resume" + ? current.execution.activeUserKey + : undefined, lease: undefined, status: current.execution.status === "running" @@ -1698,7 +1964,7 @@ export async function releaseConversationWork(args: { }, nowMs, ), - ); + }); return true; }); } @@ -1719,13 +1985,15 @@ export async function completeConversationWork(args: { const hasPending = pendingMessages(current).length > 0; const needsRun = current.execution.status === "awaiting_resume"; const runnable = needsRun || hasPending; - await writeConversation( + await writeConversationAndReleaseAdmission({ state, lock, - withExecutionUpdate( + nowMs, + conversation: withExecutionUpdate( current, { ...current.execution, + activeUserKey: needsRun ? current.execution.activeUserKey : undefined, lease: undefined, status: needsRun ? "awaiting_resume" @@ -1736,7 +2004,7 @@ export async function completeConversationWork(args: { }, nowMs, ), - ); + }); return runnable ? "pending" : "completed"; }); } @@ -1843,20 +2111,22 @@ export async function deadLetterAttempt(args: { return "lost_lease"; } const runnable = pendingMessages(current).length > 0; - await writeConversation( + await writeConversationAndReleaseAdmission({ state, lock, - withExecutionUpdate( + nowMs, + conversation: withExecutionUpdate( current, { ...current.execution, + activeUserKey: undefined, lease: undefined, status: runnable ? "pending" : "failed", runId: runnable ? current.execution.runId : undefined, }, nowMs, ), - ); + }); return runnable ? "pending" : "failed"; }); } @@ -1876,10 +2146,11 @@ export async function clearExpiredConversationLease(args: { ) { return false; } - await writeConversation( + await writeConversationAndReleaseAdmission({ state, lock, - withExecutionUpdate( + nowMs, + conversation: withExecutionUpdate( current, { ...current.execution, @@ -1888,7 +2159,7 @@ export async function clearExpiredConversationLease(args: { }, nowMs, ), - ); + }); return true; }); } @@ -1899,16 +2170,24 @@ export async function deleteConversationState(args: { state?: StateAdapter; }): Promise { await withConversationMutation(args, async (state) => { - await state.delete(conversationKey(args.conversationId)); - await removeIndexEntry({ - state, - indexKey: CONVERSATION_ACTIVE_INDEX_KEY, - conversationId: args.conversationId, - }); - await removeIndexEntry({ - state, - indexKey: CONVERSATION_BY_ACTIVITY_INDEX_KEY, - conversationId: args.conversationId, + await withConversationAdmission(state, async () => { + const nowMs = now(); + await state.delete(conversationKey(args.conversationId)); + await removeConversationAdmissionWhileLocked({ + conversationId: args.conversationId, + nowMs, + state, + }); + await removeIndexEntry({ + state, + indexKey: CONVERSATION_ACTIVE_INDEX_KEY, + conversationId: args.conversationId, + }); + await removeIndexEntry({ + state, + indexKey: CONVERSATION_BY_ACTIVITY_INDEX_KEY, + conversationId: args.conversationId, + }); }); }); } diff --git a/packages/junior/src/chat/task-execution/store.ts b/packages/junior/src/chat/task-execution/store.ts index 05c256c610..a8e1cdf4df 100644 --- a/packages/junior/src/chat/task-execution/store.ts +++ b/packages/junior/src/chat/task-execution/store.ts @@ -28,6 +28,7 @@ export { type Source, type StartConversationWorkAcquired, type StartConversationWorkActive, + type StartConversationWorkLimited, type StartConversationWorkNoWork, type StartConversationWorkResult, } from "@/chat/task-execution/state"; @@ -344,12 +345,11 @@ async function markConversationWorkEnqueued(args: { } /** Try to acquire the durable execution lease for one conversation. */ -export async function startConversationWork(args: { - conversationId: string; - conversationStore?: ConversationStore; - nowMs?: number; - state?: StateAdapter; -}) { +export async function startConversationWork( + args: { + conversationStore?: ConversationStore; + } & Parameters[0], +) { const result = await workState.startConversationWork(args); await recordExecutionMetadata(args); return result; diff --git a/packages/junior/src/chat/task-execution/worker.ts b/packages/junior/src/chat/task-execution/worker.ts index 79248a8904..0dc04ece93 100644 --- a/packages/junior/src/chat/task-execution/worker.ts +++ b/packages/junior/src/chat/task-execution/worker.ts @@ -1,9 +1,11 @@ import type { StateAdapter } from "chat"; import type { Destination } from "@sentry/junior-plugin-api"; -import { getChatConfig } from "@/chat/config"; +import { botConfig, getChatConfig } from "@/chat/config"; import { logException, logInfo, logWarn, withLogContext } from "@/chat/logging"; import type { ConversationStore } from "@/chat/conversations/store"; import { isProviderRetryError } from "@/chat/services/provider-error"; +import type { ConversationAdmissionBudgets } from "@/chat/services/budgets"; +import { reportBudgetExceeded } from "@/chat/services/budget-reporting"; import { ConversationQueueMessageRejectedError, type ConversationQueueMessage, @@ -60,6 +62,7 @@ export interface ConversationWorkerResult { export interface ConversationWorkProcessResult { status: | "active" + | "capacity_deferred" | "completed" | "failed" | "lost_lease" @@ -69,6 +72,7 @@ export interface ConversationWorkProcessResult { } export interface ProcessConversationWorkOptions { + conversationBudgets?: ConversationAdmissionBudgets; checkInIntervalMs?: number; conversationStore?: ConversationStore; nowMs?: () => number; @@ -112,6 +116,33 @@ function selectAttemptMessages(work: ConversationWorkState): InboundMessage[] { : selectContiguousActorBatch(messages); } +function activeUserKeyForMessage( + message: InboundMessage | undefined, +): string | undefined { + if (!message) { + return undefined; + } + const userId = message.input.authorId?.trim(); + if ( + !userId || + (message.source !== "api" && + message.source !== "local" && + message.source !== "slack") + ) { + return undefined; + } + return message.destination.platform === "slack" + ? `slack:${message.destination.teamId}:${userId}` + : `${message.source}:${userId}`; +} + +function activeUserKeyForWork(work: ConversationWorkState): string | undefined { + return ( + work.execution.activeUserKey ?? + activeUserKeyForMessage(selectAttemptMessages(work)[0]) + ); +} + function nudgeIdempotencyKey( reason: string, conversationId: string, @@ -291,10 +322,14 @@ async function processConversationWorkInContext( ); } const destination = initial.destination; + const activeUserKey = activeUserKeyForWork(initial); + const conversationBudgets = options.conversationBudgets ?? botConfig.budgets; const lease = await startConversationWork({ + activeUserKey, conversationId, conversationStore: options.conversationStore, + budgets: conversationBudgets, nowMs: now(options), state: options.state, }); @@ -324,6 +359,25 @@ async function processConversationWorkInContext( }); return { status: "active" }; } + if (lease.status === "limited") { + const nudgeNowMs = now(options); + await ensureConversationWake({ + conversationId, + conversationStore: options.conversationStore, + delayMs: CONVERSATION_WORK_DEFER_DELAY_MS, + idempotencyKey: nudgeIdempotencyKey( + `budget_${lease.budget.name}`, + conversationId, + nudgeNowMs, + ), + nowMs: nudgeNowMs, + queue: options.queue, + replaceExistingWake: true, + state: options.state, + }); + await reportBudgetExceeded(lease.budget); + return { status: "capacity_deferred" }; + } const startedAtMs = now(options); const softYieldDeadlineMs = @@ -344,6 +398,17 @@ async function processConversationWorkInContext( options, }); logInfo("conversation.work.lease.acquired", { + "app.conversation.active_count": lease.activeConversationCount, + "app.conversation.active_limit": + conversationBudgets.active_conversations_global, + ...(lease.activeUserConversationCount !== undefined + ? { + "app.conversation.active_user_count": + lease.activeUserConversationCount, + "app.conversation.active_user_limit": + conversationBudgets.active_conversations_user, + } + : {}), "app.lease.expires_at_ms": lease.leaseExpiresAtMs, "app.worker.soft_yield_deadline_ms": softYieldDeadlineMs, }); diff --git a/packages/junior/tests/component/config/chat-config.test.ts b/packages/junior/tests/component/config/chat-config.test.ts index 08bf79899f..8bb1627100 100644 --- a/packages/junior/tests/component/config/chat-config.test.ts +++ b/packages/junior/tests/component/config/chat-config.test.ts @@ -392,9 +392,37 @@ describe("chat config", () => { ); }); - it("sets max slices per turn from core config", async () => { + it("uses default system budgets", async () => { const { botConfig } = await loadConfig(); - expect(botConfig.maxSlicesPerTurn).toBe(100); + expect(botConfig.budgets).toEqual({ + active_conversations_global: 100, + active_conversations_user: 5, + turn_runtime: 21_600_000, + turn_steps: 500, + }); + }); + + it("reads conversation safety limit overrides", async () => { + process.env.JUNIOR_MAX_ACTIVE_CONVERSATIONS = "20"; + process.env.JUNIOR_MAX_ACTIVE_CONVERSATIONS_PER_USER = "2"; + process.env.JUNIOR_MAX_STEPS_PER_TURN = "750"; + process.env.JUNIOR_MAX_TURN_RUNTIME_MS = "3600000"; + + const { botConfig } = await loadConfig(); + expect(botConfig.budgets).toEqual({ + active_conversations_global: 20, + active_conversations_user: 2, + turn_runtime: 3_600_000, + turn_steps: 750, + }); + }); + + it("rejects invalid conversation safety limits", async () => { + process.env.JUNIOR_MAX_ACTIVE_CONVERSATIONS = "0"; + + await expect(loadConfig()).rejects.toThrow( + "JUNIOR_MAX_ACTIVE_CONVERSATIONS must be a positive integer", + ); }); it("uses default AGENT_TURN_TIMEOUT_MS when env var is unset", async () => { diff --git a/packages/junior/tests/component/runtime/agent-resume.test.ts b/packages/junior/tests/component/runtime/agent-resume.test.ts index 834893817b..a15340e6c4 100644 --- a/packages/junior/tests/component/runtime/agent-resume.test.ts +++ b/packages/junior/tests/component/runtime/agent-resume.test.ts @@ -2,9 +2,13 @@ import { afterEach, beforeEach, describe, expect, it } from "vitest"; import { createLocalSource } from "@sentry/junior-plugin-api"; import { createResumeState } from "@/chat/agent/resume"; import { AuthorizationPauseError } from "@/chat/services/auth-pause"; +import { BudgetExceededError } from "@/chat/services/budgets"; import { loadTurnSessionRecord } from "@/chat/services/turn-session-record"; import { disconnectStateAdapter } from "@/chat/state/adapter"; -import { getAgentTurnSessionRecord } from "@/chat/state/turn-session"; +import { + getAgentTurnSessionRecord, + upsertAgentTurnSessionRecord, +} from "@/chat/state/turn-session"; const originalStateAdapter = process.env.JUNIOR_STATE_ADAPTER; @@ -56,4 +60,88 @@ describe("agent resume", () => { getAgentTurnSessionRecord(conversationId, turnId), ).resolves.toBeUndefined(); }); + + it("limits model steps and cumulative runtime for one turn", async () => { + const conversationId = "local:test:turn-limits"; + const turnId = "turn-limits"; + const destination = { platform: "local" as const, conversationId }; + const base = { + destination, + durability: {}, + getLoadedSkillNames: () => [], + getModelId: () => "test/model", + getReasoningLevel: () => undefined, + recordActiveMcpProviders: async () => {}, + runSource: createLocalSource(conversationId), + conversationId, + turnId, + sessionRecordState: await loadTurnSessionRecord({ + conversationId, + sessionId: turnId, + }), + surface: "internal" as const, + }; + const steps = createResumeState({ + ...base, + startedAtMs: Date.now(), + turnBudgets: { turn_runtime: 60_000, turn_steps: 1 }, + }); + + await expect(steps.startStep()).resolves.toBeUndefined(); + await expect(steps.startStep()).rejects.toMatchObject({ + budget: { name: "turn_steps", outcome: "stop" }, + }); + + const runtime = createResumeState({ + ...base, + startedAtMs: Date.now() - 1_000, + turnBudgets: { turn_runtime: 500, turn_steps: 10 }, + }); + await expect(runtime.startStep()).rejects.toMatchObject({ + budget: { name: "turn_runtime", outcome: "stop" }, + }); + }); + + it("keeps the step count across resumed history replacements", async () => { + const conversationId = "local:test:durable-step-limit"; + const turnId = "turn-durable-step-limit"; + await upsertAgentTurnSessionRecord({ + conversationId, + modelId: "test/model", + piMessages: [ + { + role: "user", + content: [{ type: "text", text: "continue" }], + timestamp: 1, + }, + ], + resumeReason: "yield", + sessionId: turnId, + sliceId: 2, + state: "awaiting_resume", + stepCount: 1, + }); + const resume = createResumeState({ + conversationId, + destination: { platform: "local", conversationId }, + durability: {}, + getLoadedSkillNames: () => [], + getModelId: () => "test/model", + getReasoningLevel: () => undefined, + recordActiveMcpProviders: async () => {}, + runSource: createLocalSource(conversationId), + sessionRecordState: await loadTurnSessionRecord({ + conversationId, + sessionId: turnId, + }), + startedAtMs: Date.now(), + surface: "internal", + turnId, + turnBudgets: { turn_runtime: 60_000, turn_steps: 1 }, + }); + + await expect(resume.startStep()).rejects.toBeInstanceOf( + BudgetExceededError, + ); + }); }); diff --git a/packages/junior/tests/component/runtime/agent-run-agent-continue.test.ts b/packages/junior/tests/component/runtime/agent-run-agent-continue.test.ts index 62c0c4a924..1c1fd8a893 100644 --- a/packages/junior/tests/component/runtime/agent-run-agent-continue.test.ts +++ b/packages/junior/tests/component/runtime/agent-run-agent-continue.test.ts @@ -360,7 +360,7 @@ describe("agent continuation composition", () => { ]); }); - it("throws terminal timeout failures instead of returning an error reply after the execution limit", async () => { + it("throws terminal timeout failures after the cumulative runtime limit", async () => { promptMode.value = "continueSettlesAfterAbort"; const piMessages: PiMessage[] = [ { @@ -373,10 +373,11 @@ describe("agent continuation composition", () => { modelId: "test/model", conversationId: "conversation-timeout-cap", sessionId: "turn-timeout-cap", - sliceId: botConfig.maxSlicesPerTurn, + sliceId: 2, state: "awaiting_resume", piMessages, resumeReason: "timeout", + cumulativeDurationMs: botConfig.budgets.turn_runtime - 10_000, }); const replyPromise = executeAgentRun({ @@ -394,11 +395,11 @@ describe("agent continuation composition", () => { await vi.advanceTimersByTimeAsync(10_000); const error = await replyPromise; - const { TurnSliceLimitExceededError } = - await import("@/chat/services/turn-limit"); - expect(error).toBeInstanceOf(TurnSliceLimitExceededError); + const { BudgetExceededError } = await import("@/chat/services/budgets"); + expect(error).toBeInstanceOf(BudgetExceededError); + expect(error).toMatchObject({ budget: { name: "turn_runtime" } }); expect(error).not.toHaveProperty("text"); - expect(error.message).toContain("execution limit"); + expect(error.message).toContain("turn_runtime"); const sessionRecord = await getAgentTurnSessionRecord( "conversation-timeout-cap", @@ -407,8 +408,8 @@ describe("agent continuation composition", () => { expect(sessionRecord).toMatchObject({ state: "failed", resumeReason: "timeout", - sliceId: botConfig.maxSlicesPerTurn, - errorMessage: expect.stringContaining("execution limit"), + sliceId: 2, + errorMessage: expect.stringContaining("turn_runtime"), }); }); diff --git a/packages/junior/tests/component/services/turn-session-record.test.ts b/packages/junior/tests/component/services/turn-session-record.test.ts index 0b6f48090b..c39e0e344e 100644 --- a/packages/junior/tests/component/services/turn-session-record.test.ts +++ b/packages/junior/tests/component/services/turn-session-record.test.ts @@ -729,7 +729,45 @@ describe("persistAuthPauseSessionRecord", () => { }); }); - it("fails timeout sessions instead of scheduling beyond the execution limit", async () => { + it("adds current-slice totals when completing from summary-only resume state", async () => { + const { + listAgentTurnSessionSummariesForConversation, + recordAgentTurnSessionSummary, + } = await import("@/chat/state/turn-session"); + const conversationId = "conversation-summary-totals"; + const sessionId = "turn-summary-totals"; + + await recordAgentTurnSessionSummary({ + conversationId, + cumulativeDurationMs: 1_500, + cumulativeUsage: { inputTokens: 10 }, + sessionId, + sliceId: 1, + state: "awaiting_resume", + stepCount: 2, + }); + await recordAgentTurnSessionSummary({ + conversationId, + currentDurationMs: 2_250, + currentUsage: { outputTokens: 7 }, + sessionId, + sliceId: 2, + state: "completed", + stepCount: 3, + }); + + const summary = ( + await listAgentTurnSessionSummariesForConversation(conversationId) + ).find((candidate) => candidate.sessionId === sessionId); + expect(summary).toMatchObject({ + cumulativeDurationMs: 3_750, + cumulativeUsage: { inputTokens: 10, outputTokens: 7 }, + state: "completed", + stepCount: 3, + }); + }); + + it("fails timeout sessions after the cumulative runtime limit", async () => { const { persistContinuationSessionRecord } = await import("@/chat/services/turn-session-record"); const { botConfig } = await import("@/chat/config"); @@ -748,11 +786,11 @@ describe("persistAuthPauseSessionRecord", () => { modelId: "test/model", conversationId: "conversation-timeout-cap", sessionId: "turn-timeout-cap", - sliceId: botConfig.maxSlicesPerTurn, + sliceId: 9, state: "awaiting_resume", piMessages, resumeReason: "timeout", - cumulativeDurationMs: 12_000, + cumulativeDurationMs: botConfig.budgets.turn_runtime - 3_000, }); await expect( @@ -761,16 +799,16 @@ describe("persistAuthPauseSessionRecord", () => { modelId: "test-model", conversationId: "conversation-timeout-cap", sessionId: "turn-timeout-cap", - currentSliceId: botConfig.maxSlicesPerTurn, + currentSliceId: 9, currentDurationMs: 3_000, messages: piMessages, errorMessage: "timed out again", }), ).resolves.toMatchObject({ state: "failed", - sliceId: botConfig.maxSlicesPerTurn, - cumulativeDurationMs: 15_000, - errorMessage: expect.stringContaining("execution limit"), + sliceId: 9, + cumulativeDurationMs: botConfig.budgets.turn_runtime, + errorMessage: expect.stringContaining("turn_runtime"), piMessages, }); @@ -778,9 +816,9 @@ describe("persistAuthPauseSessionRecord", () => { getAgentTurnSessionRecord("conversation-timeout-cap", "turn-timeout-cap"), ).resolves.toMatchObject({ state: "failed", - sliceId: botConfig.maxSlicesPerTurn, - cumulativeDurationMs: 15_000, - errorMessage: expect.stringContaining("execution limit"), + sliceId: 9, + cumulativeDurationMs: botConfig.budgets.turn_runtime, + errorMessage: expect.stringContaining("turn_runtime"), piMessages, }); }); diff --git a/packages/junior/tests/component/task-execution/conversation-work.test.ts b/packages/junior/tests/component/task-execution/conversation-work.test.ts index 6f35384219..82370bd32b 100644 --- a/packages/junior/tests/component/task-execution/conversation-work.test.ts +++ b/packages/junior/tests/component/task-execution/conversation-work.test.ts @@ -1554,12 +1554,17 @@ describe("conversation work execution", () => { it("extends the lease with worker check-ins during long execution", async () => { vi.useFakeTimers({ now: 1_000 }); const queue = createConversationWorkQueueTestAdapter(); + const limits = { + active_conversations_global: 1, + active_conversations_user: 5, + }; await appendInboundMessage({ message: inboundMessage("m1"), nowMs: 1_000 }); const entered = deferred(); const finish = deferred(); const running = processConversationWork(conversationQueueMessage(), { checkInIntervalMs: 15_000, + conversationBudgets: limits, queue, run: async (context) => { await context.attempt.drain(async () => {}); @@ -1585,10 +1590,203 @@ describe("conversation work execution", () => { 16_000 + CONVERSATION_WORK_LEASE_TTL_MS, ); + const otherConversationId = "slack:C123:check-in-capacity"; + await appendInboundMessage({ + message: inboundMessage("check-in-capacity", { + conversationId: otherConversationId, + input: { authorId: "U456", text: "wait for capacity" }, + }), + nowMs: 17_000, + }); + await expect( + startConversationWork({ + activeUserKey: "slack:T123:U456", + conversationId: otherConversationId, + budgets: limits, + nowMs: 92_000, + }), + ).resolves.toMatchObject({ + budget: { name: "active_conversations_global" }, + status: "limited", + }); + + vi.setSystemTime(92_000); finish.resolve(); await expect(running).resolves.toEqual({ status: "completed" }); }); + it("keeps conversation work queued when the global active limit is full", async () => { + const activeConversationId = "slack:C123:active-global"; + const queuedConversationId = "slack:C123:queued-global"; + const limits = { + active_conversations_global: 1, + active_conversations_user: 5, + }; + await appendInboundMessage({ + message: inboundMessage("active-global", { + conversationId: activeConversationId, + }), + nowMs: 1_000, + }); + await appendInboundMessage({ + message: inboundMessage("queued-global", { + conversationId: queuedConversationId, + input: { authorId: "U456", text: "queued global" }, + }), + nowMs: 1_000, + }); + await expect( + startConversationWork({ + activeUserKey: "slack:T123:U123", + conversationId: activeConversationId, + budgets: limits, + nowMs: 2_000, + }), + ).resolves.toMatchObject({ status: "acquired" }); + const queue = createConversationWorkQueueTestAdapter(); + const run = vi.fn(); + + await expect( + processConversationWork( + conversationQueueMessage({ conversationId: queuedConversationId }), + { + conversationBudgets: limits, + nowMs: () => 3_000, + queue, + run, + }, + ), + ).resolves.toEqual({ status: "capacity_deferred" }); + + expect(run).not.toHaveBeenCalled(); + expect(queue.sentRecords()).toEqual([ + { + conversationId: queuedConversationId, + delayMs: CONVERSATION_WORK_DEFER_DELAY_MS, + idempotencyKey: `budget_active_conversations_global:${queuedConversationId}:3000`, + }, + ]); + const queuedState = await getConversationWorkState({ + conversationId: queuedConversationId, + }); + expect(queuedState).toMatchObject({ needsRun: true }); + expect(queuedState?.lease).toBeUndefined(); + }); + + it("ignores an admission entry when the matching conversation lease was not written", async () => { + const conversationId = "slack:C123:orphaned-admission"; + const state = getStateAdapter(); + await appendInboundMessage({ + message: inboundMessage("orphaned-admission", { conversationId }), + nowMs: 1_000, + state, + }); + let failConversationWrite = true; + const failingState = new Proxy(state, { + get(target, property, receiver) { + if (property === "set") { + return async (key: string, value: unknown, ttlMs?: number) => { + if ( + failConversationWrite && + key === `junior:conversation:${conversationId}` + ) { + failConversationWrite = false; + throw new Error("conversation write failed"); + } + return await target.set(key, value, ttlMs); + }; + } + const value = Reflect.get(target, property, receiver); + return typeof value === "function" ? value.bind(target) : value; + }, + }) as StateAdapter; + const budgets = { + active_conversations_global: 1, + active_conversations_user: 1, + }; + + await expect( + startConversationWork({ + activeUserKey: "slack:T123:U123", + budgets, + conversationId, + nowMs: 2_000, + state: failingState, + }), + ).rejects.toThrow("conversation write failed"); + + await expect( + startConversationWork({ + activeUserKey: "slack:T123:U123", + budgets, + conversationId, + nowMs: 3_000, + state, + }), + ).resolves.toMatchObject({ status: "acquired" }); + }); + + it("keeps a user's extra conversation queued while other users can run", async () => { + const activeConversationId = "slack:C123:active-user"; + const queuedConversationId = "slack:C123:queued-user"; + const otherUserConversationId = "slack:C123:other-user"; + const limits = { + active_conversations_global: 2, + active_conversations_user: 1, + }; + await appendInboundMessage({ + message: inboundMessage("active-user", { + conversationId: activeConversationId, + }), + nowMs: 1_000, + }); + await appendInboundMessage({ + message: inboundMessage("queued-user", { + conversationId: queuedConversationId, + }), + nowMs: 1_000, + }); + await appendInboundMessage({ + message: inboundMessage("other-user", { + conversationId: otherUserConversationId, + input: { authorId: "U456", text: "other user" }, + }), + nowMs: 1_000, + }); + await expect( + startConversationWork({ + activeUserKey: "slack:T123:U123", + conversationId: activeConversationId, + budgets: limits, + nowMs: 2_000, + }), + ).resolves.toMatchObject({ status: "acquired" }); + + await expect( + startConversationWork({ + activeUserKey: "slack:T123:U123", + conversationId: queuedConversationId, + budgets: limits, + nowMs: 2_001, + }), + ).resolves.toMatchObject({ + budget: { name: "active_conversations_user" }, + status: "limited", + }); + await expect( + startConversationWork({ + activeUserKey: "slack:T123:U456", + conversationId: otherUserConversationId, + budgets: limits, + nowMs: 2_002, + }), + ).resolves.toMatchObject({ + activeConversationCount: 2, + activeUserConversationCount: 1, + status: "acquired", + }); + }); + it("reports lost lease after periodic check-in loses ownership", async () => { vi.useFakeTimers({ now: 1_000 }); const queue = createConversationWorkQueueTestAdapter(); diff --git a/packages/junior/tests/unit/app-config.test.ts b/packages/junior/tests/unit/app-config.test.ts index d24593b416..285b99002c 100644 --- a/packages/junior/tests/unit/app-config.test.ts +++ b/packages/junior/tests/unit/app-config.test.ts @@ -6,6 +6,7 @@ import { githubPlugin } from "@sentry/junior-github"; import { afterEach, describe, expect, it, vi } from "vitest"; import { createApp, defineJuniorPlugins } from "@/app"; import { botConfig } from "@/chat/config"; +import { describeBudgets } from "@/chat/services/budgets"; import { getConfigDefaults, setConfigDefaults, @@ -870,6 +871,7 @@ describe("createApp plugin config", () => { expect.objectContaining({ agentName: botConfig.userName, componentGallery: true, + systemBudgets: describeBudgets(botConfig.budgets), }), ); diff --git a/packages/junior/tests/unit/logging/log-context.test.ts b/packages/junior/tests/unit/logging/log-context.test.ts index b665c3b130..e6872167fc 100644 --- a/packages/junior/tests/unit/logging/log-context.test.ts +++ b/packages/junior/tests/unit/logging/log-context.test.ts @@ -2,12 +2,25 @@ import { describe, expect, it } from "vitest"; import { getBoundLogAttributes, getBoundLogContext, + logContextToAttributes, runWithLogAttributes, runWithLogContext, updateLogAttributes, } from "@/chat/log-context"; describe("log context", () => { + it("maps the canonical Junior turn id into telemetry attributes", () => { + expect( + logContextToAttributes({ + conversationId: "conversation-1", + turnId: "turn-1", + }), + ).toEqual({ + "app.ai.turn.id": "turn-1", + "gen_ai.conversation.id": "conversation-1", + }); + }); + it("restores nested async scopes", async () => { expect(getBoundLogAttributes()).toEqual({}); diff --git a/packages/junior/tests/unit/services/budgets.test.ts b/packages/junior/tests/unit/services/budgets.test.ts new file mode 100644 index 0000000000..79844d2d28 --- /dev/null +++ b/packages/junior/tests/unit/services/budgets.test.ts @@ -0,0 +1,69 @@ +import { describe, expect, it } from "vitest"; +import { + BudgetExceededError, + buildBudgetExceededResponse, + checkBudgets, + describeBudgets, + getBudgetAttributes, + isBudgetExceededError, + readBudgetLimits, +} from "@/chat/services/budgets"; + +describe("system budgets", () => { + it("uses one registry for config, descriptions, queue, and stop decisions", async () => { + const limits = readBudgetLimits({}, () => undefined); + expect(limits).toMatchObject({ + active_conversations_global: 100, + turn_runtime: 21_600_000, + turn_steps: 500, + }); + expect(describeBudgets(limits)).toEqual( + expect.arrayContaining([ + expect.objectContaining({ + label: "Agent steps per turn", + limit: 500, + name: "turn_steps", + outcome: "stop", + }), + ]), + ); + await expect( + checkBudgets( + { active_conversations_global: 100 }, + { activeConversations: 100, stage: "conversation_admission" }, + ), + ).resolves.toEqual({ + limit: 100, + name: "active_conversations_global", + outcome: "queue", + value: 100, + }); + const exceeded = await checkBudgets( + { turn_steps: 500 }, + { runtimeMs: 0, stage: "turn", steps: 500 }, + ); + expect(exceeded).toEqual({ + limit: 500, + name: "turn_steps", + outcome: "stop", + value: 500, + }); + const error = new BudgetExceededError(exceeded!); + expect(isBudgetExceededError(error)).toBe(true); + expect(getBudgetAttributes(error.budget)).toEqual({ + "app.budget.limit": 500, + "app.budget.name": "turn_steps", + "app.budget.outcome": "stop", + "app.budget.value": 500, + }); + }); + + it("gives users an actionable response without internal implementation details", () => { + const response = buildBudgetExceededResponse("abc123"); + + expect(response).toContain("reached a system budget"); + expect(response).toContain("smaller or more specific request"); + expect(response).toContain("event_id=abc123"); + expect(response).not.toContain("continuation"); + }); +}); diff --git a/packages/junior/tests/unit/services/turn-limit.test.ts b/packages/junior/tests/unit/services/turn-limit.test.ts deleted file mode 100644 index 76f06d47b6..0000000000 --- a/packages/junior/tests/unit/services/turn-limit.test.ts +++ /dev/null @@ -1,23 +0,0 @@ -import { describe, expect, it } from "vitest"; -import { - TurnSliceLimitExceededError, - buildTurnLimitResponse, -} from "@/chat/services/turn-limit"; - -describe("turn execution limit", () => { - it("keeps the internal limit in diagnostics", () => { - expect(new TurnSliceLimitExceededError(100)).toMatchObject({ - name: "TurnSliceLimitExceededError", - message: "Agent turn exceeded execution limit (100 slices)", - }); - }); - - it("gives users an actionable response without internal implementation details", () => { - const response = buildTurnLimitResponse("abc123"); - - expect(response).toContain("reached its execution limit"); - expect(response).toContain("smaller or more specific request"); - expect(response).toContain("event_id=abc123"); - expect(response).not.toContain("continuation"); - }); -}); diff --git a/packages/junior/tests/unit/state/turn-session-telemetry.test.ts b/packages/junior/tests/unit/state/turn-session-telemetry.test.ts new file mode 100644 index 0000000000..73de2459af --- /dev/null +++ b/packages/junior/tests/unit/state/turn-session-telemetry.test.ts @@ -0,0 +1,38 @@ +import { describe, expect, it } from "vitest"; +import { getTurnTelemetryAttributes } from "@/chat/state/turn-session"; + +describe("turn session telemetry", () => { + it("uses OTEL GenAI attributes and explicit Junior turn attributes", () => { + expect( + getTurnTelemetryAttributes({ + conversationId: "conversation-1", + cumulativeDurationMs: 90_000, + cumulativeUsage: { + inputTokens: 10, + outputTokens: 4, + cost: { total: 0.25 }, + }, + modelId: "openai/gpt-test", + resumeReason: "yield", + sessionId: "turn-1", + sliceId: 3, + state: "completed", + stepCount: 12, + surface: "slack", + }), + ).toEqual({ + "app.ai.turn.id": "turn-1", + "app.ai.turn.resume_reason": "yield", + "app.ai.turn.runtime_ms": 90_000, + "app.ai.turn.slice_id": 3, + "app.ai.turn.state": "completed", + "app.ai.turn.step_count": 12, + "app.ai.turn.surface": "slack", + "app.cost.total_usd": 0.25, + "gen_ai.conversation.id": "conversation-1", + "gen_ai.request.model": "openai/gpt-test", + "gen_ai.usage.input_tokens": 10, + "gen_ai.usage.output_tokens": 4, + }); + }); +});