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 9b92b330cf..c04580b8b3 100644 --- a/packages/docs/src/content/docs/reference/config-and-env.md +++ b/packages/docs/src/content/docs/reference/config-and-env.md @@ -42,6 +42,7 @@ related: | `CRON_SECRET` or `JUNIOR_SCHEDULER_SECRET` | Conditional | Bearer token for the internal heartbeat route; use `CRON_SECRET` with Vercel Cron, or `JUNIOR_SCHEDULER_SECRET` for a non-Vercel heartbeat caller. | | `JUNIOR_TIMEZONE` | No | Default IANA timezone for scheduler authoring when the scheduler plugin is enabled. Defaults to `America/Los_Angeles`. | | `AI_GATEWAY_API_KEY` | No | Fallback AI Gateway auth when Vercel OIDC is unavailable (local/CI/non-Vercel hosts). On Vercel, prefer project OIDC so usage attributes to the project. | +| `TYPESAFE_API_KEY` | No | Use TypeSafe JEV for the experimental passive Slack reply classifier. This has no effect unless `passive-routing` is enabled. | | `BLOB_STORE_ID` | Conditional | Vercel Blob store for durable conversation attachments and published public artifacts. Vercel sets this when an OIDC-enabled Blob store is connected to the project. | | `BLOB_READ_WRITE_TOKEN` | Conditional | Static Vercel Blob credential when OIDC is unavailable. Vercel sets this for a token-connected store. | @@ -166,6 +167,12 @@ const app = await createApp({ }); ``` +`passive-routing` applies only to Slack threads that Junior has already joined. +It does not make Junior process every message in a channel. If `TYPESAFE_API_KEY` +is set, Junior uses TypeSafe JEV for the passive reply decision. It uses a +conservative reply threshold and stays quiet if the provider request fails. +Without this key, Junior uses the configured fast model. + `junior chat` enables experimental `subagents` automatically because it is the local createApp-equivalent entrypoint and already wires the child-worker path. diff --git a/packages/junior/src/chat/app/services.ts b/packages/junior/src/chat/app/services.ts index 6ccdc16d6a..01f41016e6 100644 --- a/packages/junior/src/chat/app/services.ts +++ b/packages/junior/src/chat/app/services.ts @@ -18,6 +18,7 @@ import { type SubscribedReplyPolicy, type SubscribedReplyPolicyDeps, } from "@/chat/services/subscribed-reply-policy"; +import { createJevSubscribedReplyClassifier } from "@/chat/services/jev-subscribed-reply-classifier"; import { createVisionContextService, type VisionContextDeps, @@ -53,6 +54,7 @@ export interface JuniorRuntimeServiceOverrides { export function createJuniorRuntimeServices( overrides: JuniorRuntimeServiceOverrides = {}, ): JuniorRuntimeServices { + const typesafeApiKey = process.env.TYPESAFE_API_KEY?.trim(); const conversationMemory = createConversationMemoryService({ completeText: overrides.conversationMemory?.completeText ?? completeText, }); @@ -83,6 +85,11 @@ export function createJuniorRuntimeServices( executeTurn: async (run, saveResult, timeoutMs) => await executeTurn(agentRunner, run, saveResult, timeoutMs), subscribedReplyPolicy: createSubscribedReplyPolicy({ + classifyReply: + overrides.subscribedReplyPolicy?.classifyReply ?? + (typesafeApiKey + ? createJevSubscribedReplyClassifier({ apiKey: typesafeApiKey }) + : undefined), completeObject: overrides.subscribedReplyPolicy?.completeObject ?? completeObject, }), diff --git a/packages/junior/src/chat/services/jev-subscribed-reply-classifier.ts b/packages/junior/src/chat/services/jev-subscribed-reply-classifier.ts new file mode 100644 index 0000000000..a9c650a32f --- /dev/null +++ b/packages/junior/src/chat/services/jev-subscribed-reply-classifier.ts @@ -0,0 +1,116 @@ +import { z } from "zod"; +import type { + RouterEvidence, + SubscribedReplyClassification, +} from "@/chat/services/subscribed-decision"; + +const TYPESAFE_SYSTEM_ONE_URL = "https://api.typesafe.ai/v1/systemone"; +const JEV_MODEL = "jev-latest"; + +const jevResponseSchema = z + .object({ + answers: z + .object({ + should_reply: z + .object({ type: z.literal("noul"), noul: z.number().min(0).max(1) }) + .strict(), + should_unsubscribe: z + .object({ type: z.literal("noul"), noul: z.number().min(0).max(1) }) + .strict(), + }) + .strict(), + }) + .passthrough(); + +const REPLY_THRESHOLD = 0.8; +const UNSUBSCRIBE_THRESHOLD = 0.9; +const REQUEST_TIMEOUT_MS = 2_000; + +export interface JevSubscribedReplyClassifierOptions { + apiKey: string; + fetch?: typeof fetch; +} + +/** Create a JEV classifier for passive replies in subscribed Slack threads. */ +export function createJevSubscribedReplyClassifier( + options: JevSubscribedReplyClassifierOptions, +): (args: { + botUserName: string; + evidence: RouterEvidence; + latestMessage: string; +}) => Promise { + const fetchImpl = options.fetch ?? globalThis.fetch; + + return async ({ botUserName, evidence, latestMessage }) => { + const response = await fetchImpl(TYPESAFE_SYSTEM_ONE_URL, { + method: "POST", + headers: { + authorization: `Bearer ${options.apiKey}`, + "content-type": "application/json", + }, + body: JSON.stringify({ + model: JEV_MODEL, + state: { + assistant_name: botUserName, + transcript: evidence.entries, + signals: { + assistant_was_last_speaker: evidence.assistantWasLastSpeaker, + current_message_has_attachments: + evidence.currentMessageHasAttachments, + current_message_has_directed_follow_up_cue: + evidence.currentMessageHasDirectedFollowUpCue, + current_message_is_terse_clarification: + evidence.currentMessageIsTerseClarification, + human_messages_since_last_assistant: + evidence.humanMessagesSinceLastAssistant ?? null, + latest_prior_message_role: evidence.latestPriorMessageRole, + }, + latest_prior_assistant_message: evidence.latestPriorAssistantMessage, + latest_message: latestMessage, + }, + questions: { + should_reply: { + type: "noul", + instructions: + "Should the assistant reply to the latest message in this Slack thread?", + criteria: { + true: "The latest message asks the assistant to act, answer, clarify, or continue its work.", + false: + "The message is side conversation, an acknowledgment, or directed to another person.", + }, + }, + should_unsubscribe: { + type: "noul", + instructions: + "Does the latest message clearly ask the assistant to stop participating in this thread?", + criteria: { + true: "The user clearly asks the assistant to leave or stop replying in this thread.", + false: "The user does not clearly opt out of assistant replies.", + }, + }, + }, + }), + signal: AbortSignal.timeout(REQUEST_TIMEOUT_MS), + }); + + if (!response.ok) { + throw new Error( + `TypeSafe System One request failed (${response.status})`, + ); + } + + const result = jevResponseSchema.parse(await response.json()); + const replyProbability = result.answers.should_reply.noul; + const unsubscribeProbability = result.answers.should_unsubscribe.noul; + + const shouldUnsubscribe = unsubscribeProbability >= UNSUBSCRIBE_THRESHOLD; + return { + shouldReply: replyProbability >= REPLY_THRESHOLD, + shouldUnsubscribe, + confidence: shouldUnsubscribe + ? Math.max(unsubscribeProbability, 1 - unsubscribeProbability) + : Math.max(replyProbability, 1 - replyProbability), + reason: `jev reply=${replyProbability.toFixed(2)} unsubscribe=${unsubscribeProbability.toFixed(2)}`, + }; + }; +} diff --git a/packages/junior/src/chat/services/subscribed-decision.ts b/packages/junior/src/chat/services/subscribed-decision.ts index c110e433b7..30236b1492 100644 --- a/packages/junior/src/chat/services/subscribed-decision.ts +++ b/packages/junior/src/chat/services/subscribed-decision.ts @@ -39,13 +39,67 @@ export interface SubscribedDecisionResult { reasonDetail?: string; } +export interface SubscribedReplyClassification { + confidence: number; + reason: string; + shouldReply: boolean; + shouldUnsubscribe: boolean; +} + +function mapSubscribedReplyClassification( + classification: SubscribedReplyClassification, + costUsd?: number, +): SubscribedDecisionResult { + const cost = costUsd === undefined ? {} : { costUsd }; + const reason = classification.reason.trim() || "classifier"; + if (classification.shouldUnsubscribe) { + if (classification.confidence < ROUTER_CONFIDENCE_THRESHOLD) { + return { + ...cost, + shouldReply: false, + reason: SubscribedReplyReason.LowConfidence, + reasonDetail: `${classification.confidence.toFixed(2)}: ${reason}`, + }; + } + return { + ...cost, + shouldReply: false, + shouldUnsubscribe: true, + reason: SubscribedReplyReason.ThreadOptOut, + reasonDetail: reason, + }; + } + if (!classification.shouldReply) { + return { + ...cost, + shouldReply: false, + reason: SubscribedReplyReason.SideConversation, + reasonDetail: reason, + }; + } + if (classification.confidence < ROUTER_CONFIDENCE_THRESHOLD) { + return { + ...cost, + shouldReply: false, + reason: SubscribedReplyReason.LowConfidence, + reasonDetail: `${classification.confidence.toFixed(2)}: ${reason}`, + }; + } + return { + ...cost, + shouldReply: true, + reason: SubscribedReplyReason.Classifier, + reasonDetail: reason, + }; +} + interface TranscriptMessage { author: string; role: "assistant" | "user"; text: string; } -interface RouterEvidence { +export interface RouterEvidence { assistantWasLastSpeaker: boolean; currentMessageHasAttachments: boolean; currentMessageHasDirectedFollowUpCue: boolean; @@ -90,8 +144,7 @@ const TRANSCRIPT_MESSAGE_LINE_RE = /** `!stop` may appear anywhere in the message. */ const BANG_STOP_RE = /(?:^|\s)!stop(?=\s|$|[.!?,;:])/i; /** Drop a leading `@jr` / mention before matching bare `stop`. */ -const LEADING_ADDRESS_RE = - /^(?:(?:<@[^>]+>|@[\w.-]+)\s*[,:\-–—]?\s*)+/i; +const LEADING_ADDRESS_RE = /^(?:(?:<@[^>]+>|@[\w.-]+)\s*[,:\-–—]?\s*)+/i; /** Whole message is only `stop` after any leading address. */ const BARE_STOP_RE = /^stop(?:\s*[.!…]+)?$/i; const ACKNOWLEDGMENT_ONLY_RE = @@ -434,6 +487,11 @@ export async function decideSubscribedThreadReply(args: { botUserName: string; modelId: string; input: SubscribedDecisionInput; + classifyReply?: (args: { + botUserName: string; + evidence: RouterEvidence; + latestMessage: string; + }) => Promise; completeObject: (args: { modelId: string; schema: typeof replyDecisionSchema; @@ -510,6 +568,16 @@ export async function decideSubscribedThreadReply(args: { } try { + if (args.classifyReply) { + return mapSubscribedReplyClassification( + await args.classifyReply({ + botUserName: args.botUserName, + evidence, + latestMessage: rawText.trim() || "[attachment-only message]", + }), + ); + } + const result = await args.completeObject({ modelId: args.modelId, schema: replyDecisionSchema, @@ -528,50 +596,15 @@ export async function decideSubscribedThreadReply(args: { }); const parsed = replyDecisionSchema.parse(result.object); - const reason = parsed.reason?.trim() || "classifier"; - if (parsed.should_unsubscribe) { - if (parsed.confidence < ROUTER_CONFIDENCE_THRESHOLD) { - return { - ...(result.costUsd !== undefined ? { costUsd: result.costUsd } : undefined), - shouldReply: false, - reason: SubscribedReplyReason.LowConfidence, - reasonDetail: `${parsed.confidence.toFixed(2)}: ${reason}`, - }; - } - - return { - ...(result.costUsd !== undefined ? { costUsd: result.costUsd } : undefined), - shouldReply: false, - shouldUnsubscribe: true, - reason: SubscribedReplyReason.ThreadOptOut, - reasonDetail: reason, - }; - } - - if (!parsed.should_reply) { - return { - ...(result.costUsd !== undefined ? { costUsd: result.costUsd } : undefined), - shouldReply: false, - reason: SubscribedReplyReason.SideConversation, - reasonDetail: reason, - }; - } - - if (parsed.confidence < ROUTER_CONFIDENCE_THRESHOLD) { - return { - ...(result.costUsd !== undefined ? { costUsd: result.costUsd } : undefined), - shouldReply: false, - reason: SubscribedReplyReason.LowConfidence, - reasonDetail: `${parsed.confidence.toFixed(2)}: ${reason}`, - }; - } - - return { - ...(result.costUsd !== undefined ? { costUsd: result.costUsd } : undefined), - shouldReply: true, - reason: SubscribedReplyReason.Classifier, - reasonDetail: reason, - }; + return mapSubscribedReplyClassification( + { + confidence: parsed.confidence, + reason: parsed.reason, + shouldReply: parsed.should_reply, + shouldUnsubscribe: parsed.should_unsubscribe, + }, + result.costUsd, + ); } catch (error) { if (isProviderRetryError(error)) { throw error; diff --git a/packages/junior/src/chat/services/subscribed-reply-policy.ts b/packages/junior/src/chat/services/subscribed-reply-policy.ts index 6526c5c754..2c3d101921 100644 --- a/packages/junior/src/chat/services/subscribed-reply-policy.ts +++ b/packages/junior/src/chat/services/subscribed-reply-policy.ts @@ -3,10 +3,17 @@ import { logWarn } from "@/chat/logging"; import { decideSubscribedThreadReply, type SubscribedDecisionInput, + type SubscribedReplyClassification, + type RouterEvidence, } from "@/chat/services/subscribed-decision"; import type { completeObject } from "@/chat/pi/client"; export interface SubscribedReplyPolicyDeps { + classifyReply?: (args: { + botUserName: string; + evidence: RouterEvidence; + latestMessage: string; + }) => Promise; completeObject: typeof completeObject; } @@ -29,6 +36,7 @@ export function createSubscribedReplyPolicy( botUserName: botConfig.userName, modelId: botConfig.fastModelId, input: args, + classifyReply: deps.classifyReply, completeObject: deps.completeObject, logClassifierFailure: (error) => { logWarn("subscribed_message.classifier.failed", { @@ -42,7 +50,9 @@ export function createSubscribedReplyPolicy( ? `${decision.reason}:${decision.reasonDetail}` : decision.reason; return { - ...(decision.costUsd !== undefined ? { costUsd: decision.costUsd } : undefined), + ...(decision.costUsd !== undefined + ? { costUsd: decision.costUsd } + : undefined), shouldReply: decision.shouldReply, shouldUnsubscribe: decision.shouldUnsubscribe, reason, diff --git a/packages/junior/tests/component/routing/jev-subscribed-reply-classifier.test.ts b/packages/junior/tests/component/routing/jev-subscribed-reply-classifier.test.ts new file mode 100644 index 0000000000..def1caea6a --- /dev/null +++ b/packages/junior/tests/component/routing/jev-subscribed-reply-classifier.test.ts @@ -0,0 +1,80 @@ +import { describe, expect, it, vi } from "vitest"; +import { createJevSubscribedReplyClassifier } from "@/chat/services/jev-subscribed-reply-classifier"; +import type { RouterEvidence } from "@/chat/services/subscribed-decision"; + +const evidence: RouterEvidence = { + assistantWasLastSpeaker: true, + currentMessageHasAttachments: false, + currentMessageHasDirectedFollowUpCue: false, + currentMessageIsTerseClarification: false, + entries: [{ author: "junior", role: "assistant", text: "I opened the PR." }], + humanMessagesSinceLastAssistant: 0, + latestPriorAssistantMessage: "I opened the PR.", + latestPriorMessageRole: "assistant", + omittedEntries: 0, +}; + +describe("JEV subscribed reply classifier", () => { + it("maps calibrated probabilities to a reply decision", async () => { + const fetchMock = vi.fn( + async (_input: RequestInfo | URL, _init?: RequestInit) => + Response.json({ + model: "jev-latest", + answers: { + should_reply: { type: "noul", noul: 0.91 }, + should_unsubscribe: { type: "noul", noul: 0.03 }, + }, + usage: { input_tokens: 120, output_tokens: 2 }, + }), + ); + const classify = createJevSubscribedReplyClassifier({ + apiKey: "test-key", + fetch: fetchMock, + }); + + await expect( + classify({ + botUserName: "junior", + evidence, + latestMessage: "can you link it?", + }), + ).resolves.toEqual({ + confidence: 0.91, + reason: "jev reply=0.91 unsubscribe=0.03", + shouldReply: true, + shouldUnsubscribe: false, + }); + + const [, init] = fetchMock.mock.calls[0]!; + expect(init?.headers).toEqual({ + authorization: "Bearer test-key", + "content-type": "application/json", + }); + expect(JSON.parse(String(init?.body))).toMatchObject({ + model: "jev-latest", + state: { + assistant_name: "junior", + latest_message: "can you link it?", + }, + questions: { + should_reply: { type: "noul" }, + should_unsubscribe: { type: "noul" }, + }, + }); + }); + + it("fails when TypeSafe returns a provider error", async () => { + const classify = createJevSubscribedReplyClassifier({ + apiKey: "test-key", + fetch: vi.fn(async () => new Response("overloaded", { status: 529 })), + }); + + await expect( + classify({ + botUserName: "junior", + evidence, + latestMessage: "please continue", + }), + ).rejects.toThrow("TypeSafe System One request failed (529)"); + }); +}); diff --git a/packages/junior/tests/unit/routing/subscribed-decision.test.ts b/packages/junior/tests/unit/routing/subscribed-decision.test.ts index df0807cdbe..36dad4a5ae 100644 --- a/packages/junior/tests/unit/routing/subscribed-decision.test.ts +++ b/packages/junior/tests/unit/routing/subscribed-decision.test.ts @@ -310,6 +310,41 @@ describe("subscribed reply decision", () => { ).toEqual([]); }); + it("uses an injected classifier instead of the fast model", async () => { + const classifyReply = vi.fn(async () => ({ + shouldReply: true, + shouldUnsubscribe: false, + confidence: 0.93, + reason: "direct question", + })); + const completeObject = vi.fn(); + + await expect( + decideSubscribedThreadReply({ + botUserName: "junior", + modelId: "router-model", + input: makeInput({ + rawText: "can you link the PR?", + text: "can you link the PR?", + }), + classifyReply, + completeObject, + logClassifierFailure: vi.fn(), + }), + ).resolves.toEqual({ + shouldReply: true, + reason: SubscribedReplyReason.Classifier, + reasonDetail: "direct question", + }); + expect(classifyReply).toHaveBeenCalledWith( + expect.objectContaining({ + botUserName: "junior", + latestMessage: "can you link the PR?", + }), + ); + expect(completeObject).not.toHaveBeenCalled(); + }); + it("projects guardian-style user/assistant evidence without tool lines", async () => { const completeObject = vi.fn( async (_request: { prompt: string; system: string }) => ({