diff --git a/AgentCompany-New/src/main/app/provider-background-worker.ts b/AgentCompany-New/src/main/app/provider-background-worker.ts new file mode 100644 index 00000000..fa454bfc --- /dev/null +++ b/AgentCompany-New/src/main/app/provider-background-worker.ts @@ -0,0 +1,38 @@ +import type { BrowserWindow } from "electron"; +import type { DatabaseSync } from "node:sqlite"; +import { IPC_CHANNELS } from "../../shared/ipc"; +import { ProviderBackgroundService } from "../services/provider-background-service"; + +const DEFAULT_INTERVAL_MS = 8_000; + +export interface ProviderBackgroundWorkerInput { + db: DatabaseSync; + mainWindow: BrowserWindow; + intervalMs?: number; +} + +export function startProviderBackgroundWorker(input: ProviderBackgroundWorkerInput): () => void { + const service = new ProviderBackgroundService(input.db); + let running = false; + const intervalMs = input.intervalMs ?? DEFAULT_INTERVAL_MS; + + const tick = async (): Promise => { + if (running) return; + running = true; + try { + const changedCount = await service.recoverBackgroundResponses(); + if (changedCount > 0 && !input.mainWindow.isDestroyed()) { + input.mainWindow.webContents.send(IPC_CHANNELS.rendererEvent, { type: "snapshot-changed" }); + } + } finally { + running = false; + } + }; + + const interval = setInterval(() => { + void tick(); + }, intervalMs); + void tick(); + + return () => clearInterval(interval); +} diff --git a/AgentCompany-New/src/main/index.ts b/AgentCompany-New/src/main/index.ts index 779fdbf6..79a86a81 100644 --- a/AgentCompany-New/src/main/index.ts +++ b/AgentCompany-New/src/main/index.ts @@ -4,6 +4,7 @@ import { fileURLToPath } from "node:url"; import { DatabaseConnection } from "./database/connection"; import { createMainWindow } from "./app/create-window"; import { registerIpcHandlers } from "./app/ipc-handlers"; +import { startProviderBackgroundWorker } from "./app/provider-background-worker"; const __dirname = dirname(fileURLToPath(import.meta.url)); const hasSingleInstanceLock = app.requestSingleInstanceLock(); @@ -26,6 +27,10 @@ app.whenReady().then(() => { mainWindow: win, artifactRoot: join(profileDir, "artifacts"), }); + const stopProviderBackgroundWorker = startProviderBackgroundWorker({ + db: connection.db, + mainWindow: win, + }); app.on("second-instance", () => { if (win.isMinimized()) win.restore(); @@ -33,6 +38,7 @@ app.whenReady().then(() => { }); app.on("before-quit", () => { + stopProviderBackgroundWorker(); connection.close(); }); }); diff --git a/AgentCompany-New/src/main/services/provider-adapters.ts b/AgentCompany-New/src/main/services/provider-adapters.ts index 90fb182a..f4ecd435 100644 --- a/AgentCompany-New/src/main/services/provider-adapters.ts +++ b/AgentCompany-New/src/main/services/provider-adapters.ts @@ -1,6 +1,27 @@ import { PROVIDER_AUTH_TYPE, type ProviderAuthType } from "../../shared/domain"; import { listProviderModels, type ProviderModelEntry } from "../../shared/provider-models"; import type { ProviderObservedRateLimitState } from "../../shared/provider-runtime-domain"; +import { + parseAccessToken, + parseAccountToken, + parseApiKey, + parseAzureProfile, + parseBedrockProfile, + parseVertexProfile, + requireAccessTokenProfile, + requireAccountTokenProfile, + requireApiKeyProfile, + requireAzureProfile, + requireBedrockProfile, + requireVertexProfile, + type ProviderAuthProfile, +} from "./provider-auth-profiles"; +import { + backgroundUnsupported, + retrieveUnsupported, + type ProviderBackgroundCreateResult, + type ProviderBackgroundRetrieveResult, +} from "./provider-background"; import { authorizeWithParser, refreshUnsupported, @@ -21,6 +42,7 @@ import { } from "./provider-streaming"; export { ProviderHttpError, providerRateLimitStateFromError } from "./provider-http"; +export type { ProviderAuthProfile } from "./provider-auth-profiles"; export interface ProviderExecutionInput { authProfile: ProviderAuthProfile; @@ -36,12 +58,18 @@ export interface ProviderExecutionResult { } export type ProviderStreamInput = ProviderExecutionInput; +export type ProviderBackgroundCreateInput = ProviderExecutionInput; export interface ProviderCancelInput { authProfile: ProviderAuthProfile; responseId: string; } +export interface ProviderBackgroundRetrieveInput { + authProfile: ProviderAuthProfile; + responseId: string; +} + export interface ProviderHealthInput { authProfile: ProviderAuthProfile; model: string | null; @@ -77,19 +105,13 @@ export interface ProviderAdapter { checkHealth(input: ProviderHealthInput): Promise; streamResponse(input: ProviderStreamInput): Promise; cancelResponse(input: ProviderCancelInput): Promise; + createBackgroundResponse(input: ProviderBackgroundCreateInput): Promise; + retrieveResponse(input: ProviderBackgroundRetrieveInput): Promise; createResponse?: (input: ProviderExecutionInput) => Promise; } export type ProviderAdapterRegistry = Record; -export type ProviderAuthProfile = - | { type: "api_key"; apiKey: string } - | { type: "access_token"; accessToken: string } - | { type: "account_token"; accessToken: string } - | { type: "azure_openai"; baseUrl: string; deployment: string; apiKey: string } - | { type: "aws_bedrock"; region: string; accessKeyId: string; secretAccessKey: string; sessionToken: string | null } - | { type: "vertex_ai"; projectId: string; region: string; serviceAccountJson: string }; - export const providerAdapters: ProviderAdapterRegistry = { openai: { providerKey: "openai", @@ -121,6 +143,40 @@ export const providerAdapters: ProviderAdapterRegistry = { }, {}); return { responseId, rawResponseJson: JSON.stringify(response.data), rateLimitState: response.rateLimitState }; }, + async createBackgroundResponse(input) { + const profile = requireApiKeyProfile(input.authProfile, "OpenAI"); + const response = await postJson("https://api.openai.com/v1/responses", { + authorization: `Bearer ${profile.apiKey}`, + }, { + model: input.model, + input: input.prompt, + background: true, + store: true, + ...(input.instructions ? { instructions: input.instructions } : {}), + }); + const data = response.data; + return { + responseId: requireStringField(data, "id", "OpenAI response id"), + status: requireStringField(data, "status", "OpenAI response status"), + rawResponseJson: JSON.stringify(data), + rateLimitState: response.rateLimitState, + }; + }, + async retrieveResponse(input) { + const profile = requireApiKeyProfile(input.authProfile, "OpenAI"); + const responseId = requireResponseId(input.responseId); + const response = await getJson(`https://api.openai.com/v1/responses/${encodeURIComponent(responseId)}`, { + authorization: `Bearer ${profile.apiKey}`, + }); + const data = response.data; + return { + responseId: requireStringField(data, "id", "OpenAI response id"), + status: requireStringField(data, "status", "OpenAI response status"), + text: optionalText(collectText(data.output_text ?? data.output)), + rawResponseJson: JSON.stringify(data), + rateLimitState: response.rateLimitState, + }; + }, async createResponse(input) { const profile = requireApiKeyProfile(input.authProfile, "OpenAI"); const response = await postJson("https://api.openai.com/v1/responses", { @@ -167,6 +223,8 @@ export const providerAdapters: ProviderAdapterRegistry = { return { events: mapAnthropicMessageStream(response.messages), rateLimitState: response.rateLimitState }; }, cancelResponse: cancelUnsupported("Anthropic"), + createBackgroundResponse: backgroundUnsupported("Anthropic"), + retrieveResponse: retrieveUnsupported("Anthropic"), async createResponse(input) { const profile = requireApiKeyProfile(input.authProfile, "Anthropic"); const response = await postJson("https://api.anthropic.com/v1/messages", { @@ -209,6 +267,8 @@ export const providerAdapters: ProviderAdapterRegistry = { getRateLimitState: providerRateLimitUnknown, streamResponse: streamUnsupported("Azure OpenAI"), cancelResponse: cancelUnsupported("Azure OpenAI"), + createBackgroundResponse: backgroundUnsupported("Azure OpenAI"), + retrieveResponse: retrieveUnsupported("Azure OpenAI"), async checkHealth(input) { const profile = requireAzureProfile(input.authProfile); const response = await getJson(`${profile.baseUrl}/openai/deployments?api-version=2024-10-21`, { "api-key": profile.apiKey }); @@ -226,6 +286,8 @@ export const providerAdapters: ProviderAdapterRegistry = { getRateLimitState: providerRateLimitUnknown, streamResponse: streamUnsupported("AWS Bedrock"), cancelResponse: cancelUnsupported("AWS Bedrock"), + createBackgroundResponse: backgroundUnsupported("AWS Bedrock"), + retrieveResponse: retrieveUnsupported("AWS Bedrock"), checkHealth: async (input) => { requireBedrockProfile(input.authProfile); return healthy(null); @@ -242,6 +304,8 @@ export const providerAdapters: ProviderAdapterRegistry = { getRateLimitState: providerRateLimitUnknown, streamResponse: streamUnsupported("Vertex AI"), cancelResponse: cancelUnsupported("Vertex AI"), + createBackgroundResponse: backgroundUnsupported("Vertex AI"), + retrieveResponse: retrieveUnsupported("Vertex AI"), checkHealth: async (input) => { requireVertexProfile(input.authProfile); return healthy(null); @@ -267,6 +331,17 @@ function requireText(parts: string[], providerKey: string): string { return text; } +function optionalText(parts: string[]): string | null { + const text = parts.map((part) => part.trim()).filter(Boolean).join("\n"); + return text || null; +} + +function requireStringField(record: JsonRecord, key: string, label: string): string { + const value = record[key]; + if (typeof value !== "string" || !value.trim()) throw new Error(`${label} is required.`); + return value.trim(); +} + function openAiCompatibleAdapter({ providerKey, label, @@ -301,6 +376,8 @@ function openAiCompatibleAdapter({ return { events: mapOpenAiChatCompletionStream(response.messages), rateLimitState: response.rateLimitState }; }, cancelResponse: cancelUnsupported(label), + createBackgroundResponse: backgroundUnsupported(label), + retrieveResponse: retrieveUnsupported(label), async createResponse(input) { const profile = requireApiKeyProfile(input.authProfile, label); const messages = responseMessages(input.instructions, input.prompt); @@ -344,6 +421,8 @@ function browserTokenAdapter(providerKey: string, authType: ProviderAuthType): P getRateLimitState: providerRateLimitUnknown, streamResponse: streamUnsupported(providerKey), cancelResponse: cancelUnsupported(providerKey), + createBackgroundResponse: backgroundUnsupported(providerKey), + retrieveResponse: retrieveUnsupported(providerKey), checkHealth: async (input) => { if (authType === PROVIDER_AUTH_TYPE.ACCOUNT_SUBSCRIPTION) requireAccountTokenProfile(input.authProfile); else requireAccessTokenProfile(input.authProfile); @@ -352,118 +431,6 @@ function browserTokenAdapter(providerKey: string, authType: ProviderAuthType): P }; } -function parseApiKey(secret: string): ProviderAuthProfile { - return { type: "api_key", apiKey: nonEmptySecret(secret, "API key") }; -} - -function parseAccessToken(secret: string): ProviderAuthProfile { - return { type: "access_token", accessToken: nonEmptySecret(secret, "Access token") }; -} - -function parseAccountToken(secret: string): ProviderAuthProfile { - return { type: "account_token", accessToken: nonEmptySecret(secret, "Account token") }; -} - -function parseAzureProfile(secret: string): ProviderAuthProfile { - const parsed = parseCloudSecret(secret); - const baseUrl = requiredString(parsed.baseUrl, "Azure base URL").replace(/\/+$/, ""); - const deployment = requiredString(parsed.deployment, "Azure deployment"); - return { type: "azure_openai", baseUrl: normalizedHttpsUrl(baseUrl, "Azure base URL"), deployment, apiKey: requiredString(parsed.secret, "Azure API key") }; -} - -function parseBedrockProfile(secret: string): ProviderAuthProfile { - const parsed = parseCloudSecret(secret); - const credential = parseCredentialJson(requiredString(parsed.secret, "AWS credential")); - return { - type: "aws_bedrock", - region: requiredString(parsed.region, "AWS region"), - accessKeyId: requiredString(credential.accessKeyId, "AWS access key id"), - secretAccessKey: requiredString(credential.secretAccessKey, "AWS secret access key"), - sessionToken: typeof credential.sessionToken === "string" && credential.sessionToken.trim() ? credential.sessionToken.trim() : null, - }; -} - -function parseVertexProfile(secret: string): ProviderAuthProfile { - const parsed = parseCloudSecret(secret); - const serviceAccountJson = requiredString(parsed.secret, "Vertex service account JSON"); - parseCredentialJson(serviceAccountJson); - return { - type: "vertex_ai", - projectId: requiredString(parsed.projectId, "Vertex project id"), - region: requiredString(parsed.region, "Vertex region"), - serviceAccountJson, - }; -} - -function parseCloudSecret(secret: string): JsonRecord { - try { - const parsed = JSON.parse(secret) as unknown; - if (!isRecord(parsed)) throw new Error("Cloud credential profile is invalid."); - return parsed; - } catch (error) { - if (error instanceof SyntaxError) throw new Error("Cloud credential profile must be JSON."); - throw error; - } -} - -function parseCredentialJson(value: string): JsonRecord { - try { - const parsed = JSON.parse(value) as unknown; - if (!isRecord(parsed)) throw new Error("Credential JSON is invalid."); - return parsed; - } catch (error) { - if (error instanceof SyntaxError) throw new Error("Credential must be JSON."); - throw error; - } -} - -function requiredString(value: unknown, label: string): string { - if (typeof value !== "string" || !value.trim()) throw new Error(`${label} is required.`); - return value.trim(); -} - -function nonEmptySecret(secret: string, label: string): string { - const value = secret.trim(); - if (!value) throw new Error(`${label} is required.`); - return value; -} - -function normalizedHttpsUrl(value: string, label: string): string { - const url = new URL(value); - if (url.protocol !== "https:") throw new Error(`${label} must use HTTPS.`); - return url.toString().replace(/\/+$/, ""); -} - -function requireApiKeyProfile(profile: ProviderAuthProfile, label: string): Extract { - if (profile.type !== "api_key") throw new Error(`${label} API key profile is required.`); - return profile; -} - -function requireAccessTokenProfile(profile: ProviderAuthProfile): Extract { - if (profile.type !== "access_token") throw new Error("Access token profile is required."); - return profile; -} - -function requireAccountTokenProfile(profile: ProviderAuthProfile): Extract { - if (profile.type !== "account_token") throw new Error("Account token profile is required."); - return profile; -} - -function requireAzureProfile(profile: ProviderAuthProfile): Extract { - if (profile.type !== "azure_openai") throw new Error("Azure OpenAI profile is required."); - return profile; -} - -function requireBedrockProfile(profile: ProviderAuthProfile): Extract { - if (profile.type !== "aws_bedrock") throw new Error("AWS Bedrock profile is required."); - return profile; -} - -function requireVertexProfile(profile: ProviderAuthProfile): Extract { - if (profile.type !== "vertex_ai") throw new Error("Vertex AI profile is required."); - return profile; -} - function requireResponseId(value: string): string { const responseId = value.trim(); if (!responseId) throw new Error("Provider response id is required."); diff --git a/AgentCompany-New/src/main/services/provider-auth-profiles.ts b/AgentCompany-New/src/main/services/provider-auth-profiles.ts new file mode 100644 index 00000000..c663e385 --- /dev/null +++ b/AgentCompany-New/src/main/services/provider-auth-profiles.ts @@ -0,0 +1,122 @@ +import type { JsonRecord } from "./provider-http"; +import { isRecord } from "./provider-http"; + +export type ProviderAuthProfile = + | { type: "api_key"; apiKey: string } + | { type: "access_token"; accessToken: string } + | { type: "account_token"; accessToken: string } + | { type: "azure_openai"; baseUrl: string; deployment: string; apiKey: string } + | { type: "aws_bedrock"; region: string; accessKeyId: string; secretAccessKey: string; sessionToken: string | null } + | { type: "vertex_ai"; projectId: string; region: string; serviceAccountJson: string }; + +export function parseApiKey(secret: string): ProviderAuthProfile { + return { type: "api_key", apiKey: nonEmptySecret(secret, "API key") }; +} + +export function parseAccessToken(secret: string): ProviderAuthProfile { + return { type: "access_token", accessToken: nonEmptySecret(secret, "Access token") }; +} + +export function parseAccountToken(secret: string): ProviderAuthProfile { + return { type: "account_token", accessToken: nonEmptySecret(secret, "Account token") }; +} + +export function parseAzureProfile(secret: string): ProviderAuthProfile { + const parsed = parseCloudSecret(secret); + const baseUrl = requiredString(parsed.baseUrl, "Azure base URL").replace(/\/+$/, ""); + const deployment = requiredString(parsed.deployment, "Azure deployment"); + return { type: "azure_openai", baseUrl: normalizedHttpsUrl(baseUrl, "Azure base URL"), deployment, apiKey: requiredString(parsed.secret, "Azure API key") }; +} + +export function parseBedrockProfile(secret: string): ProviderAuthProfile { + const parsed = parseCloudSecret(secret); + const credential = parseCredentialJson(requiredString(parsed.secret, "AWS credential")); + return { + type: "aws_bedrock", + region: requiredString(parsed.region, "AWS region"), + accessKeyId: requiredString(credential.accessKeyId, "AWS access key id"), + secretAccessKey: requiredString(credential.secretAccessKey, "AWS secret access key"), + sessionToken: typeof credential.sessionToken === "string" && credential.sessionToken.trim() ? credential.sessionToken.trim() : null, + }; +} + +export function parseVertexProfile(secret: string): ProviderAuthProfile { + const parsed = parseCloudSecret(secret); + const serviceAccountJson = requiredString(parsed.secret, "Vertex service account JSON"); + parseCredentialJson(serviceAccountJson); + return { + type: "vertex_ai", + projectId: requiredString(parsed.projectId, "Vertex project id"), + region: requiredString(parsed.region, "Vertex region"), + serviceAccountJson, + }; +} + +export function requireApiKeyProfile(profile: ProviderAuthProfile, label: string): Extract { + if (profile.type !== "api_key") throw new Error(`${label} API key profile is required.`); + return profile; +} + +export function requireAccessTokenProfile(profile: ProviderAuthProfile): Extract { + if (profile.type !== "access_token") throw new Error("Access token profile is required."); + return profile; +} + +export function requireAccountTokenProfile(profile: ProviderAuthProfile): Extract { + if (profile.type !== "account_token") throw new Error("Account token profile is required."); + return profile; +} + +export function requireAzureProfile(profile: ProviderAuthProfile): Extract { + if (profile.type !== "azure_openai") throw new Error("Azure OpenAI profile is required."); + return profile; +} + +export function requireBedrockProfile(profile: ProviderAuthProfile): Extract { + if (profile.type !== "aws_bedrock") throw new Error("AWS Bedrock profile is required."); + return profile; +} + +export function requireVertexProfile(profile: ProviderAuthProfile): Extract { + if (profile.type !== "vertex_ai") throw new Error("Vertex AI profile is required."); + return profile; +} + +function parseCloudSecret(secret: string): JsonRecord { + try { + const parsed = JSON.parse(secret) as unknown; + if (!isRecord(parsed)) throw new Error("Cloud credential profile is invalid."); + return parsed; + } catch (error) { + if (error instanceof SyntaxError) throw new Error("Cloud credential profile must be JSON."); + throw error; + } +} + +function parseCredentialJson(value: string): JsonRecord { + try { + const parsed = JSON.parse(value) as unknown; + if (!isRecord(parsed)) throw new Error("Credential JSON is invalid."); + return parsed; + } catch (error) { + if (error instanceof SyntaxError) throw new Error("Credential must be JSON."); + throw error; + } +} + +function requiredString(value: unknown, label: string): string { + if (typeof value !== "string" || !value.trim()) throw new Error(`${label} is required.`); + return value.trim(); +} + +function nonEmptySecret(secret: string, label: string): string { + const value = secret.trim(); + if (!value) throw new Error(`${label} is required.`); + return value; +} + +function normalizedHttpsUrl(value: string, label: string): string { + const url = new URL(value); + if (url.protocol !== "https:") throw new Error(`${label} must use HTTPS.`); + return url.toString().replace(/\/+$/, ""); +} diff --git a/AgentCompany-New/src/main/services/provider-background-service.ts b/AgentCompany-New/src/main/services/provider-background-service.ts new file mode 100644 index 00000000..468bc08e --- /dev/null +++ b/AgentCompany-New/src/main/services/provider-background-service.ts @@ -0,0 +1,322 @@ +import type { DatabaseSync } from "node:sqlite"; +import type { CreateProviderResponseInput } from "../../shared/commands"; +import { + FEATURE_FLAG_KEY, + PROVIDER_AUTH_TYPE, + PROVIDER_DATA_POLICY, + PROVIDER_EXECUTION_MODE, + PROVIDER_REQUEST_STATE, + type ProviderAuthType, + type ProviderExecutionMode, +} from "../../shared/domain"; +import { currentProviderUsageWindow, type ProviderObservedRateLimitState } from "../../shared/provider-runtime-domain"; +import { AuditService } from "./audit-service"; +import { AuthBrokerService, type ProviderSecretVault } from "./auth-broker-service"; +import { OperationsService } from "./operations-service"; +import { SecretVault } from "./secret-vault"; +import { providerAdapters, type ProviderAdapterRegistry } from "./provider-adapters"; +import { ProviderRateLimitRegistry } from "./provider-rate-limit-registry"; +import { ProviderRequestStore, type ProviderBackgroundRequestRow } from "./provider-request-store"; +import { ProviderResponseObserver } from "./provider-response-observer"; +import { hashPrompt, readProviderFailureMessage, requireBackgroundText } from "./provider-response-utils"; + +const AUTH_STATE_CONFIGURED = "configured"; +const HEALTH_STATE_HEALTHY = "healthy"; +const ACTIVE_BACKGROUND_STATUSES = new Set(["queued", "in_progress"]); +const FAILED_BACKGROUND_STATUSES = new Set(["failed", "incomplete"]); +const CANCELLED_BACKGROUND_STATUSES = new Set(["cancelled"]); + +export class ProviderBackgroundService { + private readonly audit: AuditService; + private readonly authBroker: AuthBrokerService; + private readonly operations: OperationsService; + private readonly rateLimitRegistry: ProviderRateLimitRegistry; + private readonly requestStore: ProviderRequestStore; + private readonly responseObserver: ProviderResponseObserver; + + constructor( + private readonly db: DatabaseSync, + private readonly adapters: ProviderAdapterRegistry = providerAdapters, + vault: ProviderSecretVault = new SecretVault(), + ) { + this.audit = new AuditService(db); + this.authBroker = new AuthBrokerService(db, vault); + this.operations = new OperationsService(db); + this.rateLimitRegistry = new ProviderRateLimitRegistry(db); + this.requestStore = new ProviderRequestStore(db); + this.responseObserver = new ProviderResponseObserver(db, this.audit, this.operations, this.requestStore, this.rateLimitRegistry); + } + + async create(input: CreateProviderResponseInput): Promise { + const execution = this.prepareExecution(input); + const requestId = this.requestStore.create({ + orgId: input.orgId, + provider: execution.provider, + model: execution.provider.model, + promptSha256: hashPrompt(execution.prompt), + inputCharCount: execution.prompt.length, + executionMode: PROVIDER_EXECUTION_MODE.BACKGROUND, + createdAt: new Date().toISOString(), + }); + const startedAt = Date.now(); + try { + const authProfile = execution.adapter.parseAuthProfile(execution.credential.secret); + const result = await execution.adapter.createBackgroundResponse({ + authProfile, + model: execution.provider.model, + prompt: execution.prompt, + instructions: execution.instructions, + }); + this.recordObservedRateLimit(input.orgId, execution.provider, execution.provider.model, result.rateLimitState); + this.requestStore.markBackgroundAccepted({ + orgId: input.orgId, + requestId, + rawResponseJson: result.rawResponseJson, + externalResponseId: result.responseId, + }); + this.recordProviderHealth(execution.provider.id, input.orgId, HEALTH_STATE_HEALTHY, null); + this.audit.record({ + orgId: input.orgId, + entityType: "provider_request", + entityId: requestId, + actorType: "system", + actorId: null, + action: "background_response_started", + payload: { providerKey: execution.provider.provider_key, model: execution.provider.model, status: result.status }, + }); + return requestId; + } catch (error) { + this.responseObserver.recordFailure({ + orgId: input.orgId, + provider: execution.provider, + requestId, + executionMode: PROVIDER_EXECUTION_MODE.BACKGROUND, + startedAt, + inputCharCount: execution.prompt.length, + error, + }); + throw error; + } + } + + async poll(input: { orgId: string; requestId: string }): Promise { + await this.pollRow(this.requestStore.loadBackgroundRequest(input.orgId, input.requestId)); + } + + async recoverBackgroundResponses(orgId?: string): Promise { + const requests = this.requestStore.listRecoverableBackgroundRequests(orgId); + let recovered = 0; + for (const request of requests) { + await this.pollRow(request); + recovered += 1; + } + return recovered; + } + + private async pollRow(request: ProviderBackgroundRequestRow): Promise { + let provider: ProviderRow | ProviderBackgroundRequestProvider = { + id: request.provider_id, + provider_key: request.provider_key, + model: request.model, + }; + const startedAt = Date.now(); + try { + const providerRow = this.loadConfiguredProvider(request.org_id, request.provider_id); + provider = providerRow; + const adapter = this.requireAdapter(providerRow.provider_key, providerRow.auth_type); + const credential = this.loadCredential(request.org_id, providerRow); + const result = await adapter.retrieveResponse({ + authProfile: adapter.parseAuthProfile(credential.secret), + responseId: request.external_response_id, + }); + this.recordObservedRateLimit(request.org_id, providerRow, providerRow.model, result.rateLimitState); + if (ACTIVE_BACKGROUND_STATUSES.has(result.status)) { + this.requestStore.markBackgroundProgress({ + orgId: request.org_id, + requestId: request.id, + rawResponseJson: result.rawResponseJson, + externalResponseId: result.responseId, + state: result.status === "in_progress" ? PROVIDER_REQUEST_STATE.STREAMING : PROVIDER_REQUEST_STATE.PENDING, + }); + this.recordProviderHealth(provider.id, request.org_id, HEALTH_STATE_HEALTHY, null); + return; + } + if (result.status === "completed") { + const responseText = requireBackgroundText(result.text, provider.provider_key); + this.requestStore.markSucceeded({ + orgId: request.org_id, + requestId: request.id, + responseText, + rawResponseJson: result.rawResponseJson, + outputCharCount: responseText.length, + externalResponseId: result.responseId, + completedAt: new Date().toISOString(), + }); + this.responseObserver.recordSuccess({ + orgId: request.org_id, + provider, + requestId: request.id, + executionMode: PROVIDER_EXECUTION_MODE.BACKGROUND, + startedAt, + action: "background_response_completed", + }); + this.requestStore.recordUsage(request.org_id, provider.id, request.input_char_count, responseText.length); + return; + } + if (CANCELLED_BACKGROUND_STATUSES.has(result.status)) { + this.requestStore.markCancelled({ + orgId: request.org_id, + requestId: request.id, + rawResponseJson: result.rawResponseJson, + externalResponseId: result.responseId, + cancelledAt: new Date().toISOString(), + }); + this.recordProviderHealth(provider.id, request.org_id, HEALTH_STATE_HEALTHY, null); + return; + } + if (FAILED_BACKGROUND_STATUSES.has(result.status)) { + throw new ProviderBackgroundStatusError(readProviderFailureMessage(result.rawResponseJson, result.status), result.rawResponseJson); + } + throw new Error(`Provider background response returned unsupported status ${result.status}.`); + } catch (error) { + this.responseObserver.recordFailure({ + orgId: request.org_id, + provider, + requestId: request.id, + executionMode: PROVIDER_EXECUTION_MODE.BACKGROUND, + startedAt, + inputCharCount: request.input_char_count, + error, + rawResponseJson: error instanceof ProviderBackgroundStatusError ? error.rawResponseJson : null, + }); + } + } + + private prepareExecution(input: CreateProviderResponseInput): PreparedBackgroundExecution { + const prompt = input.prompt.trim(); + if (!prompt) throw new Error("Prompt is required."); + const provider = this.loadConfiguredProvider(input.orgId, input.providerId); + this.enforcePolicy(input.orgId, provider, input.executionMode ?? PROVIDER_EXECUTION_MODE.BACKGROUND); + if (provider.auth_type !== PROVIDER_AUTH_TYPE.API_KEY) { + throw new Error("Provider background execution requires an API key adapter."); + } + if (!provider.model) throw new Error("Provider model is required."); + return { + prompt, + instructions: input.instructions?.trim() || null, + provider: { ...provider, model: provider.model }, + adapter: this.requireAdapter(provider.provider_key, provider.auth_type), + credential: this.loadCredential(input.orgId, provider), + }; + } + + private enforcePolicy(orgId: string, provider: ProviderRow, executionMode: ProviderExecutionMode): void { + if (executionMode !== PROVIDER_EXECUTION_MODE.BACKGROUND) throw new Error("Provider background execution mode is required."); + if (provider.data_policy === PROVIDER_DATA_POLICY.ZERO_DATA_RETENTION) { + throw new Error("Background execution conflicts with zero data retention policy."); + } + if (!this.operations.isFeatureEnabled(orgId, FEATURE_FLAG_KEY.PROVIDER_BACKGROUND_EXECUTION)) { + throw new Error("Provider background execution is disabled by feature flag."); + } + if (provider.background_enabled !== 1) throw new Error("Background execution is disabled for this provider policy."); + const window = currentProviderUsageWindow(); + const usage = this.db.prepare(` + select request_count + from provider_usage_windows + where org_id = ? and provider_id = ? and window_start = ? + `).get(orgId, provider.id, window.start) as { request_count: number } | undefined; + const requestCount = usage?.request_count ?? 0; + if (requestCount >= provider.monthly_request_limit) throw new Error("Provider monthly request limit has been reached."); + } + + private loadConfiguredProvider(orgId: string, providerId: string): ProviderRow { + const provider = this.db.prepare(` + select id, provider_key, auth_type, model, credential_id, enabled, auth_state, + data_policy, background_enabled, monthly_request_limit, workspace_scope + from providers + where id = ? and org_id = ? + `).get(providerId, orgId) as ProviderRow | undefined; + if (!provider) throw new Error("Provider was not found."); + if (provider.enabled !== 1 || provider.auth_state !== AUTH_STATE_CONFIGURED) throw new Error("Provider credential is required."); + if (!provider.credential_id) throw new Error("Provider credential is required."); + return provider; + } + + private loadCredential(orgId: string, provider: ProviderRow): { secret: string } { + if (!provider.credential_id) throw new Error("Provider credential is required."); + return { + secret: this.authBroker.loadProviderSecret({ + orgId, + providerId: provider.id, + providerKey: provider.provider_key, + credentialId: provider.credential_id, + }), + }; + } + + private requireAdapter(providerKey: string, authType: string): ProviderAdapterRegistry[string] { + const adapter = this.adapters[providerKey]; + if (!adapter) throw new Error(`Provider adapter is unavailable for ${providerKey}.`); + if (adapter.authType !== authType) throw new Error(`Provider adapter auth type mismatch for ${providerKey}.`); + return adapter; + } + + private recordProviderHealth(providerId: string, orgId: string, healthState: string, error: string | null): void { + this.db.prepare(` + update providers + set health_state = ?, last_error = ?, last_checked_at = ? + where id = ? and org_id = ? + `).run(healthState, error, new Date().toISOString(), providerId, orgId); + } + + private recordObservedRateLimit( + orgId: string, + provider: Pick, + model: string | null, + state: ProviderObservedRateLimitState | null, + ): void { + if (!state) return; + this.rateLimitRegistry.recordObservation({ + orgId, + providerId: provider.id, + providerKey: provider.provider_key, + model, + state, + }); + } +} + +interface PreparedBackgroundExecution { + prompt: string; + instructions: string | null; + provider: ProviderRow & { model: string }; + adapter: ProviderAdapterRegistry[string]; + credential: { secret: string }; +} + +interface ProviderRow { + id: string; + provider_key: string; + auth_type: ProviderAuthType; + model: string | null; + credential_id: string | null; + enabled: number; + auth_state: string; + data_policy: string; + background_enabled: number; + monthly_request_limit: number; + workspace_scope: string; +} + +interface ProviderBackgroundRequestProvider { + id: string; + provider_key: string; + model: string; +} + +class ProviderBackgroundStatusError extends Error { + constructor(message: string, readonly rawResponseJson: string) { + super(message); + this.name = "ProviderBackgroundStatusError"; + } +} diff --git a/AgentCompany-New/src/main/services/provider-background.ts b/AgentCompany-New/src/main/services/provider-background.ts new file mode 100644 index 00000000..f0e10ceb --- /dev/null +++ b/AgentCompany-New/src/main/services/provider-background.ts @@ -0,0 +1,28 @@ +import type { ProviderObservedRateLimitState } from "../../shared/provider-runtime-domain"; + +export interface ProviderBackgroundCreateResult { + responseId: string; + status: string; + rawResponseJson: string; + rateLimitState: ProviderObservedRateLimitState | null; +} + +export interface ProviderBackgroundRetrieveResult { + responseId: string; + status: string; + text: string | null; + rawResponseJson: string; + rateLimitState: ProviderObservedRateLimitState | null; +} + +export function backgroundUnsupported(providerLabel: string): () => Promise { + return async () => { + throw new Error(`${providerLabel} background execution is unavailable.`); + }; +} + +export function retrieveUnsupported(providerLabel: string): () => Promise { + return async () => { + throw new Error(`${providerLabel} response retrieval is unavailable.`); + }; +} diff --git a/AgentCompany-New/src/main/services/provider-request-store.ts b/AgentCompany-New/src/main/services/provider-request-store.ts index 8269f4a1..b5c80cf0 100644 --- a/AgentCompany-New/src/main/services/provider-request-store.ts +++ b/AgentCompany-New/src/main/services/provider-request-store.ts @@ -1,5 +1,5 @@ import type { DatabaseSync } from "node:sqlite"; -import { PROVIDER_REQUEST_STATE, type ProviderExecutionMode } from "../../shared/domain"; +import { PROVIDER_EXECUTION_MODE, PROVIDER_REQUEST_STATE, type ProviderExecutionMode } from "../../shared/domain"; import { createId } from "../../shared/ids"; import { currentProviderUsageWindow } from "../../shared/provider-runtime-domain"; @@ -29,6 +29,21 @@ export interface ProviderRequestProgressInput { externalResponseId: string | null; } +export interface ProviderBackgroundAcceptedInput { + orgId: string; + requestId: string; + rawResponseJson: string; + externalResponseId: string; +} + +export interface ProviderBackgroundProgressInput { + orgId: string; + requestId: string; + rawResponseJson: string; + externalResponseId: string; + state: typeof PROVIDER_REQUEST_STATE.PENDING | typeof PROVIDER_REQUEST_STATE.STREAMING; +} + export interface ProviderRequestResultInput { orgId: string; requestId: string; @@ -44,6 +59,7 @@ export interface ProviderRequestFailureInput { requestId: string; errorMessage: string; completedAt: string; + rawResponseJson?: string | null; } export interface ProviderRequestCancellationInput { @@ -64,6 +80,18 @@ export interface ProviderCancellationRow { external_response_id: string | null; } +export interface ProviderBackgroundRequestRow { + id: string; + org_id: string; + provider_id: string; + provider_key: string; + model: string; + state: string; + external_response_id: string; + input_char_count: number; + execution_mode: ProviderExecutionMode; +} + export class ProviderRequestStore { constructor(private readonly db: DatabaseSync) {} @@ -111,6 +139,28 @@ export class ProviderRequestStore { ); } + markBackgroundAccepted(input: ProviderBackgroundAcceptedInput): void { + this.db.prepare(` + update provider_requests + set state = ?, raw_response_json = ?, external_response_id = ? + where id = ? and org_id = ? + `).run( + PROVIDER_REQUEST_STATE.PENDING, + input.rawResponseJson, + input.externalResponseId, + input.requestId, + input.orgId, + ); + } + + markBackgroundProgress(input: ProviderBackgroundProgressInput): void { + this.db.prepare(` + update provider_requests + set state = ?, raw_response_json = ?, external_response_id = ? + where id = ? and org_id = ? + `).run(input.state, input.rawResponseJson, input.externalResponseId, input.requestId, input.orgId); + } + markSucceeded(input: ProviderRequestResultInput): void { this.db.prepare(` update provider_requests @@ -132,9 +182,9 @@ export class ProviderRequestStore { markFailed(input: ProviderRequestFailureInput): void { this.db.prepare(` update provider_requests - set state = ?, error_message = ?, completed_at = ? + set state = ?, error_message = ?, raw_response_json = coalesce(?, raw_response_json), completed_at = ? where id = ? and org_id = ? - `).run(PROVIDER_REQUEST_STATE.FAILED, input.errorMessage, input.completedAt, input.requestId, input.orgId); + `).run(PROVIDER_REQUEST_STATE.FAILED, input.errorMessage, input.rawResponseJson ?? null, input.completedAt, input.requestId, input.orgId); } markCancelled(input: ProviderRequestCancellationInput): void { @@ -163,6 +213,31 @@ export class ProviderRequestStore { return request; } + listRecoverableBackgroundRequests(orgId?: string): ProviderBackgroundRequestRow[] { + const sql = ` + select id, org_id, provider_id, provider_key, model, state, external_response_id, input_char_count, execution_mode + from provider_requests + where execution_mode = ? + and state in (?, ?) + and external_response_id is not null + ${orgId ? "and org_id = ?" : ""} + order by created_at asc + `; + return orgId + ? this.db.prepare(sql).all(PROVIDER_EXECUTION_MODE.BACKGROUND, PROVIDER_REQUEST_STATE.PENDING, PROVIDER_REQUEST_STATE.STREAMING, orgId) as unknown as ProviderBackgroundRequestRow[] + : this.db.prepare(sql).all(PROVIDER_EXECUTION_MODE.BACKGROUND, PROVIDER_REQUEST_STATE.PENDING, PROVIDER_REQUEST_STATE.STREAMING) as unknown as ProviderBackgroundRequestRow[]; + } + + loadBackgroundRequest(orgId: string, requestId: string): ProviderBackgroundRequestRow { + const request = this.db.prepare(` + select id, org_id, provider_id, provider_key, model, state, external_response_id, input_char_count, execution_mode + from provider_requests + where id = ? and org_id = ? and execution_mode = ? and external_response_id is not null + `).get(requestId, orgId, PROVIDER_EXECUTION_MODE.BACKGROUND) as ProviderBackgroundRequestRow | undefined; + if (!request) throw new Error("Provider background request was not found."); + return request; + } + recordUsage(orgId: string, providerId: string, inputCharCount: number, outputCharCount: number): void { const window = currentProviderUsageWindow(); const now = new Date().toISOString(); diff --git a/AgentCompany-New/src/main/services/provider-response-observer.ts b/AgentCompany-New/src/main/services/provider-response-observer.ts index 2e6b3567..768252df 100644 --- a/AgentCompany-New/src/main/services/provider-response-observer.ts +++ b/AgentCompany-New/src/main/services/provider-response-observer.ts @@ -22,7 +22,7 @@ export interface ProviderResponseSuccessInput { requestId: string; executionMode: ProviderExecutionMode; startedAt: number; - action: "response_created" | "response_streamed"; + action: "response_created" | "response_streamed" | "background_response_completed"; } export interface ProviderResponseFailureInput { @@ -33,6 +33,7 @@ export interface ProviderResponseFailureInput { startedAt: number; inputCharCount: number; error: unknown; + rawResponseJson?: string | null; } export class ProviderResponseObserver { @@ -73,6 +74,7 @@ export class ProviderResponseObserver { requestId: input.requestId, errorMessage: message, completedAt: new Date().toISOString(), + rawResponseJson: input.rawResponseJson, }); this.requestStore.recordUsage(input.orgId, input.provider.id, input.inputCharCount, 0); this.recordProviderHealth(input.provider.id, input.orgId, HEALTH_STATE_FAILED, message); diff --git a/AgentCompany-New/src/main/services/provider-response-utils.ts b/AgentCompany-New/src/main/services/provider-response-utils.ts index a5a40366..224ca85d 100644 --- a/AgentCompany-New/src/main/services/provider-response-utils.ts +++ b/AgentCompany-New/src/main/services/provider-response-utils.ts @@ -1,3 +1,9 @@ +import { createHash } from "node:crypto"; + +export function hashPrompt(prompt: string): string { + return createHash("sha256").update(prompt).digest("hex"); +} + export function readProviderResponseId(rawResponseJson: string): string | null { const parsed = JSON.parse(rawResponseJson) as unknown; if (!isRecord(parsed)) return null; @@ -5,12 +11,28 @@ export function readProviderResponseId(rawResponseJson: string): string | null { return typeof id === "string" && id.trim() ? id.trim() : null; } +export function readProviderFailureMessage(rawResponseJson: string, status: string): string { + const parsed = JSON.parse(rawResponseJson) as unknown; + if (!isRecord(parsed)) return `Provider background response ended with status ${status}.`; + const error = parsed.error; + if (typeof error === "string" && error.trim()) return error.trim(); + if (isRecord(error) && typeof error.message === "string" && error.message.trim()) return error.message.trim(); + const incomplete = parsed.incomplete_details; + if (isRecord(incomplete) && typeof incomplete.reason === "string" && incomplete.reason.trim()) return incomplete.reason.trim(); + return `Provider background response ended with status ${status}.`; +} + export function requireStreamText(parts: string[]): string { const text = parts.join(""); if (!text.trim()) throw new Error("Provider stream returned no text content."); return text; } +export function requireBackgroundText(text: string | null, providerKey: string): string { + if (!text?.trim()) throw new Error(`${providerKey} background response returned no text content.`); + return text; +} + function isRecord(value: unknown): value is Record { return typeof value === "object" && value !== null && !Array.isArray(value); } diff --git a/AgentCompany-New/src/main/services/provider-service.ts b/AgentCompany-New/src/main/services/provider-service.ts index 0a7e5708..f082f7e5 100644 --- a/AgentCompany-New/src/main/services/provider-service.ts +++ b/AgentCompany-New/src/main/services/provider-service.ts @@ -1,5 +1,4 @@ import type { DatabaseSync } from "node:sqlite"; -import { createHash } from "node:crypto"; import type { CancelProviderResponseInput, CreateProviderResponseInput, RevokeProviderCredentialInput, SaveProviderCredentialInput, StreamProviderResponseInput, UpdateProviderPolicyInput } from "../../shared/commands"; import { FEATURE_FLAG_KEY, PROVIDER_AUTH_TYPE, PROVIDER_DATA_POLICY, PROVIDER_EXECUTION_MODE, PROVIDER_REQUEST_STATE, type ProviderAuthType, type ProviderExecutionMode } from "../../shared/domain"; import type { ProviderStreamRendererEvent } from "../../shared/provider-request-domain"; @@ -10,9 +9,10 @@ import { AuthBrokerService, type ProviderSecretVault } from "./auth-broker-servi import { OperationsService } from "./operations-service"; import { SecretVault } from "./secret-vault"; import { providerAdapters, providerRateLimitStateFromError, type ProviderAdapterRegistry } from "./provider-adapters"; +import { ProviderBackgroundService } from "./provider-background-service"; import { ProviderRateLimitRegistry } from "./provider-rate-limit-registry"; import { ProviderResponseObserver } from "./provider-response-observer"; -import { readProviderResponseId, requireStreamText } from "./provider-response-utils"; +import { hashPrompt, readProviderResponseId, requireStreamText } from "./provider-response-utils"; import { ProviderRequestStore } from "./provider-request-store"; import { launchProviderSignInWindow } from "./provider-sign-in-window"; import type { ProviderObservedRateLimitState } from "../../shared/provider-runtime-domain"; @@ -31,6 +31,7 @@ export class ProviderService { private readonly rateLimitRegistry: ProviderRateLimitRegistry; private readonly responseObserver: ProviderResponseObserver; private readonly requestStore: ProviderRequestStore; + private readonly background: ProviderBackgroundService; constructor( private readonly db: DatabaseSync, @@ -43,6 +44,7 @@ export class ProviderService { this.rateLimitRegistry = new ProviderRateLimitRegistry(db); this.requestStore = new ProviderRequestStore(db); this.responseObserver = new ProviderResponseObserver(db, this.audit, this.operations, this.requestStore, this.rateLimitRegistry); + this.background = new ProviderBackgroundService(db, adapters, vault); } saveCredential(input: SaveProviderCredentialInput): void { @@ -192,6 +194,7 @@ export class ProviderService { } async createResponse(input: CreateProviderResponseInput): Promise { + if (input.executionMode === PROVIDER_EXECUTION_MODE.BACKGROUND) return this.background.create(input); const execution = this.prepareExecution(input); if (!execution.adapter.createResponse) throw new Error(`Provider response execution is unavailable for ${execution.provider.provider_key}.`); const requestId = this.createRequestRecord(execution); @@ -489,7 +492,3 @@ interface ProviderRow { monthly_request_limit: number; workspace_scope: string; } - -function hashPrompt(prompt: string): string { - return createHash("sha256").update(prompt).digest("hex"); -} diff --git a/AgentCompany-New/tests/contract/provider-contracts.test.ts b/AgentCompany-New/tests/contract/provider-contracts.test.ts index d52a02b6..3c37181b 100644 --- a/AgentCompany-New/tests/contract/provider-contracts.test.ts +++ b/AgentCompany-New/tests/contract/provider-contracts.test.ts @@ -89,6 +89,8 @@ describe("provider contracts", () => { await expect(adapter.getRateLimitState({ authProfile, model: provider.defaultModel })).resolves.toBeNull(); expect(typeof adapter.streamResponse).toBe("function"); expect(typeof adapter.cancelResponse).toBe("function"); + expect(typeof adapter.createBackgroundResponse).toBe("function"); + expect(typeof adapter.retrieveResponse).toBe("function"); } }); diff --git a/AgentCompany-New/tests/unit/provider-background-service.test.ts b/AgentCompany-New/tests/unit/provider-background-service.test.ts new file mode 100644 index 00000000..e2506746 --- /dev/null +++ b/AgentCompany-New/tests/unit/provider-background-service.test.ts @@ -0,0 +1,208 @@ +import { randomUUID } from "node:crypto"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { describe, expect, test } from "vitest"; +import { DatabaseConnection } from "../../src/main/database/connection"; +import { OrganizationService } from "../../src/main/services/organization-service"; +import { providerAdapters, type ProviderAdapterRegistry } from "../../src/main/services/provider-adapters"; +import { ProviderBackgroundService } from "../../src/main/services/provider-background-service"; +import { ProviderService } from "../../src/main/services/provider-service"; +import { PROVIDER_DATA_POLICY, PROVIDER_EXECUTION_MODE, PROVIDER_REQUEST_STATE } from "../../src/shared/domain"; + +function createConnection(): DatabaseConnection { + return new DatabaseConnection(join(tmpdir(), `agentcompany-new-provider-background-${randomUUID()}.db`)); +} + +function createOrganization(organization: OrganizationService): string { + return organization.createOrganization({ + organizationName: "Acme Operations", + employeeName: "Dana Lee", + mission: "Coordinate agents with auditable approvals.", + goal: "Ship a verified control plane.", + }); +} + +function providerId(connection: DatabaseConnection, orgId: string, key: string): string { + return (connection.db.prepare("select id from providers where org_id = ? and provider_key = ?") + .get(orgId, key) as { id: string }).id; +} + +const testVault = { + encryptSecret: (secret: string) => Buffer.from(secret, "utf8"), + decryptSecret: (ciphertext: Buffer) => ciphertext.toString("utf8"), +}; + +describe("provider background service", () => { + test("creates a background response and recovers the completed result", async () => { + const connection = createConnection(); + const organization = new OrganizationService(connection.db); + const adapters = backgroundAdapters("completed", "Background analysis complete."); + const providers = new ProviderService(connection.db, adapters, testVault); + const background = new ProviderBackgroundService(connection.db, adapters, testVault); + let orgId = ""; + let openAiProviderId = ""; + + connection.transaction(() => { + orgId = createOrganization(organization); + openAiProviderId = providerId(connection, orgId, "openai"); + providers.saveCredential({ orgId, providerId: openAiProviderId, secret: "sk-live-provider", model: "gpt-5.4" }); + providers.updatePolicy({ + orgId, + providerId: openAiProviderId, + dataPolicy: PROVIDER_DATA_POLICY.STANDARD, + backgroundEnabled: true, + monthlyRequestLimit: 10, + workspaceScope: "organization", + }); + }); + + const requestId = await background.create({ + orgId, + providerId: openAiProviderId, + prompt: "Run long research", + instructions: "Use verified citations.", + executionMode: PROVIDER_EXECUTION_MODE.BACKGROUND, + }); + await background.recoverBackgroundResponses(); + + const request = connection.db.prepare(` + select state, response_text, external_response_id, raw_response_json, input_char_count, output_char_count + from provider_requests + where id = ? + `).get(requestId) as { + state: string; + response_text: string; + external_response_id: string; + raw_response_json: string; + input_char_count: number; + output_char_count: number; + }; + const usage = connection.db.prepare(` + select request_count, input_char_count, output_char_count + from provider_usage_windows + where org_id = ? and provider_id = ? + `).get(orgId, openAiProviderId); + const signal = connection.db.prepare("select success from operational_events where source = 'provider_gateway'").get(); + + expect(request).toMatchObject({ + state: PROVIDER_REQUEST_STATE.SUCCEEDED, + response_text: "Background analysis complete.", + external_response_id: "resp_bg_1", + input_char_count: 17, + output_char_count: 29, + }); + expect(JSON.parse(request.raw_response_json)).toMatchObject({ id: "resp_bg_1", status: "completed" }); + expect(usage).toEqual({ request_count: 1, input_char_count: 17, output_char_count: 29 }); + expect(signal).toEqual({ success: 1 }); + connection.close(); + }); + + test("recovers active background requests across service instances", async () => { + const connection = createConnection(); + const organization = new OrganizationService(connection.db); + const adapters = backgroundAdapters("in_progress", null); + const providers = new ProviderService(connection.db, adapters, testVault); + let orgId = ""; + let openAiProviderId = ""; + + connection.transaction(() => { + orgId = createOrganization(organization); + openAiProviderId = providerId(connection, orgId, "openai"); + providers.saveCredential({ orgId, providerId: openAiProviderId, secret: "sk-live-provider", model: "gpt-5.4" }); + providers.updatePolicy({ + orgId, + providerId: openAiProviderId, + dataPolicy: PROVIDER_DATA_POLICY.STANDARD, + backgroundEnabled: true, + monthlyRequestLimit: 10, + workspaceScope: "organization", + }); + }); + + const requestId = await new ProviderBackgroundService(connection.db, adapters, testVault).create({ + orgId, + providerId: openAiProviderId, + prompt: "Run long research", + instructions: null, + executionMode: PROVIDER_EXECUTION_MODE.BACKGROUND, + }); + await new ProviderBackgroundService(connection.db, adapters, testVault).recoverBackgroundResponses(); + + expect(connection.db.prepare("select state from provider_requests where id = ?").get(requestId)) + .toEqual({ state: PROVIDER_REQUEST_STATE.STREAMING }); + connection.close(); + }); + + test("persists failed background response payloads", async () => { + const connection = createConnection(); + const organization = new OrganizationService(connection.db); + const adapters = backgroundAdapters("failed", null); + const providers = new ProviderService(connection.db, adapters, testVault); + const background = new ProviderBackgroundService(connection.db, adapters, testVault); + let orgId = ""; + let openAiProviderId = ""; + + connection.transaction(() => { + orgId = createOrganization(organization); + openAiProviderId = providerId(connection, orgId, "openai"); + providers.saveCredential({ orgId, providerId: openAiProviderId, secret: "sk-live-provider", model: "gpt-5.4" }); + providers.updatePolicy({ + orgId, + providerId: openAiProviderId, + dataPolicy: PROVIDER_DATA_POLICY.STANDARD, + backgroundEnabled: true, + monthlyRequestLimit: 10, + workspaceScope: "organization", + }); + }); + + const requestId = await background.create({ + orgId, + providerId: openAiProviderId, + prompt: "Run long research", + instructions: null, + executionMode: PROVIDER_EXECUTION_MODE.BACKGROUND, + }); + await background.poll({ orgId, requestId }); + + const request = connection.db.prepare(` + select state, error_message, raw_response_json + from provider_requests + where id = ? + `).get(requestId) as { state: string; error_message: string; raw_response_json: string }; + + expect(request.state).toBe(PROVIDER_REQUEST_STATE.FAILED); + expect(request.error_message).toBe("provider worker failed"); + expect(JSON.parse(request.raw_response_json)).toMatchObject({ id: "resp_bg_1", status: "failed" }); + connection.close(); + }); +}); + +function backgroundAdapters(status: string, text: string | null): ProviderAdapterRegistry { + return { + ...providerAdapters, + openai: { + ...providerAdapters.openai, + parseAuthProfile: (secret) => ({ type: "api_key", apiKey: secret.trim() }), + checkHealth: async () => ({ checkedAt: new Date().toISOString(), rateLimitState: null }), + createBackgroundResponse: async () => ({ + responseId: "resp_bg_1", + status: "queued", + rawResponseJson: JSON.stringify({ id: "resp_bg_1", status: "queued" }), + rateLimitState: null, + }), + retrieveResponse: async () => ({ + responseId: "resp_bg_1", + status, + text, + rawResponseJson: JSON.stringify({ + id: "resp_bg_1", + status, + error: status === "failed" ? { message: "provider worker failed" } : null, + output_text: text, + }), + rateLimitState: null, + }), + }, + }; +}