Add provider gateway and collaboration records

This commit is contained in:
ZacharyZhang-NY committed 2026-05-20 23:35:30 -04:00
1 parent fcf6d90731
commit c694b7fc2c
12 files changed
+556 -9

No files matched your search

@@ -54,6 +54,15 @@ export function registerIpcHandlers(deps: IpcDeps): void {
ipcMain.handle(IPC_CHANNELS.decideApproval, (_event, input) => mutate(() => approvals.decide(input)));
ipcMain.handle(IPC_CHANNELS.sendMessage, (_event, input) => mutate(() => chat.sendMessage(input)));
ipcMain.handle(IPC_CHANNELS.saveProviderCredential, (_event, input) => mutate(() => providers.saveCredential(input)));
ipcMain.handle(IPC_CHANNELS.createProviderResponse, async (_event, input) => {
try {
await providers.createResponse(input);
deps.mainWindow.webContents.send(IPC_CHANNELS.rendererEvent, { type: "snapshot-changed" });
return ok(snapshot.load());
} catch (error) {
return fail("provider_response_failed", error instanceof Error ? error.message : String(error));
}
});
ipcMain.handle(IPC_CHANNELS.importSkillDirectory, (_event, input) => mutate(() => skills.importDirectory(input)));
ipcMain.handle(IPC_CHANNELS.bindSkill, (_event, input) => mutate(() => skills.bind(input)));
ipcMain.handle(IPC_CHANNELS.createSocialAccount, (_event, input) => mutate(() => social.createAccount(input)));
+115 -6
View File
@@ -1,6 +1,6 @@
import type { DatabaseSync } from "node:sqlite";
const SCHEMA_VERSION = 1;
const SCHEMA_VERSION = 2;
export function installSchema(db: DatabaseSync): void {
db.exec(`
@@ -12,12 +12,14 @@ export function installSchema(db: DatabaseSync): void {
`);
const row = db.prepare("select coalesce(max(version), 0) as version from schema_migrations").get() as { version: number };
if (row.version > SCHEMA_VERSION) {
throw new Error(`Database schema ${row.version} is newer than supported schema ${SCHEMA_VERSION}.`);
let currentVersion = row.version;
if (currentVersion > SCHEMA_VERSION) {
throw new Error(`Database schema ${currentVersion} is newer than supported schema ${SCHEMA_VERSION}.`);
}
if (row.version === SCHEMA_VERSION) return;
if (currentVersion === SCHEMA_VERSION) return;
db.exec(`
if (currentVersion < 1) {
db.exec(`
create table if not exists organizations (
id text primary key,
name text not null,
@@ -250,6 +252,113 @@ export function installSchema(db: DatabaseSync): void {
create index if not exists idx_messages_space_created on chat_messages(space_id, created_at);
create index if not exists idx_audit_org_created on audit_events(org_id, created_at);
insert into schema_migrations(version, name, applied_at)
values (${SCHEMA_VERSION}, 'initial_enterprise_schema', datetime('now'));
values (1, 'initial_enterprise_schema', datetime('now'));
`);
currentVersion = 1;
}
if (currentVersion < 2) {
db.exec(`
create table if not exists skill_dependencies (
revision_id text not null references skill_revisions(id) on delete cascade,
depends_on_revision_id text not null references skill_revisions(id) on delete cascade,
dependency_kind text not null,
optional integer not null,
primary key (revision_id, depends_on_revision_id)
);
create table if not exists skill_runtime_snapshots (
id text primary key,
agent_id text not null references agents(id) on delete cascade,
run_id text not null,
compiled_prompt text not null,
tools_manifest text not null,
source_revision_ids text not null,
created_at text not null
);
create table if not exists skill_catalog_tags (
skill_id text not null references skill_definitions(id) on delete cascade,
tag text not null,
tag_type text not null,
primary key (skill_id, tag, tag_type)
);
create table if not exists chat_threads (
id text primary key,
org_id text not null references organizations(id) on delete cascade,
space_id text not null references chat_spaces(id) on delete cascade,
root_message_id text references chat_messages(id) on delete set null,
title text not null,
state text not null,
created_at text not null,
updated_at text not null
);
create table if not exists chat_mentions (
message_id text not null references chat_messages(id) on delete cascade,
target_type text not null,
target_id text not null,
offset_start integer not null,
offset_end integer not null
);
create table if not exists chat_attachments (
message_id text not null references chat_messages(id) on delete cascade,
file_id text not null references file_objects(id) on delete cascade,
transfer_type text not null,
checksum text not null,
preview_key text
);
create table if not exists chat_presence (
principal_type text not null,
principal_id text not null,
status text not null,
status_text text not null,
last_seen_at text not null,
primary key (principal_type, principal_id)
);
create table if not exists chat_slash_invocations (
id text primary key,
message_id text not null references chat_messages(id) on delete cascade,
command text not null,
args_json text not null,
result_ref text,
created_at text not null
);
create table if not exists provider_requests (
id text primary key,
org_id text not null references organizations(id) on delete cascade,
provider_id text not null references providers(id) on delete cascade,
provider_key text not null,
model text not null,
prompt_sha256 text not null,
response_text text,
raw_response_json text,
state text not null,
error_message text,
created_at text not null,
completed_at text
);
create table if not exists policy_rules (
id text primary key,
org_id text not null references organizations(id) on delete cascade,
subject_type text not null,
subject_id text,
action text not null,
effect text not null,
risk_ceiling text not null,
created_at text not null
);
create index if not exists idx_provider_requests_org_created on provider_requests(org_id, created_at);
create index if not exists idx_chat_threads_space_updated on chat_threads(space_id, updated_at);
insert into schema_migrations(version, name, applied_at)
values (2, 'collaboration_provider_policy_schema', datetime('now'));
`);
}
}
@@ -14,6 +14,7 @@ export class ChatService {
const body = input.bodyMd.trim();
if (!body) throw new Error("Message body is required.");
const id = createId("msg");
const threadId = input.threadId ?? createId("thread");
const now = new Date().toISOString();
this.db.prepare(`
insert into chat_messages(
@@ -23,7 +24,7 @@ export class ChatService {
id,
input.orgId,
input.spaceId,
input.threadId,
threadId,
input.senderType,
input.senderId,
body,
@@ -31,6 +32,25 @@ export class ChatService {
null,
null,
);
if (!input.threadId) {
this.db.prepare(`
insert into chat_threads(id, org_id, space_id, root_message_id, title, state, created_at, updated_at)
values (?, ?, ?, ?, ?, ?, ?, ?)
`).run(threadId, input.orgId, input.spaceId, id, createThreadTitle(body), "open", now, now);
} else {
this.db.prepare("update chat_threads set updated_at = ? where id = ? and org_id = ?").run(now, input.threadId, input.orgId);
}
this.recordMentions(id, body);
this.recordSlashInvocation(id, body, now);
this.recordAttachments(input.orgId, id, input.attachmentFileIds ?? []);
if (input.senderId) {
this.db.prepare(`
insert into chat_presence(principal_type, principal_id, status, status_text, last_seen_at)
values (?, ?, ?, ?, ?)
on conflict(principal_type, principal_id)
do update set status = excluded.status, status_text = excluded.status_text, last_seen_at = excluded.last_seen_at
`).run(input.senderType, input.senderId, "active", "", now);
}
this.audit.record({
orgId: input.orgId,
entityType: "chat_message",
@@ -38,8 +58,62 @@ export class ChatService {
actorType: input.senderType,
actorId: input.senderId,
action: "sent",
payload: { spaceId: input.spaceId, threadId: input.threadId },
payload: { spaceId: input.spaceId, threadId, attachmentCount: input.attachmentFileIds?.length ?? 0 },
});
return id;
}
private recordMentions(messageId: string, body: string): void {
const seen = new Set<string>();
for (const match of body.matchAll(/(^|\s)@([A-Za-z][A-Za-z0-9_-]{1,63})/g)) {
const target = match[2];
const offsetStart = match.index + match[1].length;
const key = `${target}:${offsetStart}`;
if (seen.has(key)) continue;
seen.add(key);
this.db.prepare(`
insert into chat_mentions(message_id, target_type, target_id, offset_start, offset_end)
values (?, ?, ?, ?, ?)
`).run(messageId, "alias", target, offsetStart, offsetStart + target.length + 1);
}
}
private recordSlashInvocation(messageId: string, body: string, now: string): void {
const match = body.match(/^\/([a-z][a-z0-9_-]*)(?:\s+([\s\S]*))?$/i);
if (!match) return;
const command = match[1].toLowerCase();
if (!ALLOWED_SLASH_COMMANDS.has(command)) throw new Error(`Slash command /${command} is unavailable.`);
this.db.prepare(`
insert into chat_slash_invocations(id, message_id, command, args_json, result_ref, created_at)
values (?, ?, ?, ?, ?, ?)
`).run(createId("slash"), messageId, command, JSON.stringify({ raw: match[2] ?? "" }), null, now);
}
private recordAttachments(orgId: string, messageId: string, fileIds: string[]): void {
for (const fileId of fileIds) {
const file = this.db.prepare("select sha256 from file_objects where id = ? and org_id = ?")
.get(fileId, orgId) as { sha256: string } | undefined;
if (!file) throw new Error("Attachment file was not found.");
this.db.prepare(`
insert into chat_attachments(message_id, file_id, transfer_type, checksum, preview_key)
values (?, ?, ?, ?, ?)
`).run(messageId, fileId, "handoff", file.sha256, null);
}
}
}
const ALLOWED_SLASH_COMMANDS = new Set([
"agent",
"handoff",
"approve",
"deny",
"skill",
"runbook",
"file",
"escalate",
]);
function createThreadTitle(body: string): string {
const firstLine = body.split(/\r?\n/, 1)[0].trim();
return firstLine.length > 80 ? `${firstLine.slice(0, 77)}...` : firstLine;
}
@@ -0,0 +1,119 @@
export interface ProviderExecutionInput {
apiKey: string;
model: string;
prompt: string;
instructions: string | null;
}
export interface ProviderExecutionResult {
text: string;
rawResponseJson: string;
}
export interface ProviderAdapter {
createResponse(input: ProviderExecutionInput): Promise<ProviderExecutionResult>;
}
type JsonRecord = Record<string, unknown>;
async function postJson(endpoint: string, headers: Record<string, string>, body: JsonRecord): Promise<JsonRecord> {
const response = await fetch(endpoint, {
method: "POST",
headers: {
"content-type": "application/json",
...headers,
},
body: JSON.stringify(body),
});
const data = await response.json() as JsonRecord;
if (!response.ok) {
const message = readProviderError(data) ?? `Provider request failed with HTTP ${response.status}.`;
throw new Error(message);
}
return data;
}
function readProviderError(data: JsonRecord): string | null {
const error = data.error;
if (typeof error === "string") return error;
if (isRecord(error) && typeof error.message === "string") return error.message;
return null;
}
function isRecord(value: unknown): value is JsonRecord {
return typeof value === "object" && value !== null && !Array.isArray(value);
}
function collectText(value: unknown): string[] {
if (typeof value === "string") return [value];
if (Array.isArray(value)) return value.flatMap((item) => collectText(item));
if (!isRecord(value)) return [];
const direct = ["text", "output_text", "content"].flatMap((key) => collectText(value[key]));
return direct.length > 0 ? direct : Object.values(value).flatMap((item) => collectText(item));
}
function requireText(parts: string[], providerKey: string): string {
const text = parts.map((part) => part.trim()).filter(Boolean).join("\n");
if (!text) throw new Error(`${providerKey} returned no text content.`);
return text;
}
export const providerAdapters: Record<string, ProviderAdapter> = {
openai: {
async createResponse(input) {
const data = await postJson("https://api.openai.com/v1/responses", {
authorization: `Bearer ${input.apiKey}`,
}, {
model: input.model,
input: input.prompt,
...(input.instructions ? { instructions: input.instructions } : {}),
});
const text = requireText(collectText(data.output_text ?? data.output), "OpenAI");
return { text, rawResponseJson: JSON.stringify(data) };
},
},
anthropic: {
async createResponse(input) {
const data = await postJson("https://api.anthropic.com/v1/messages", {
"x-api-key": input.apiKey,
"anthropic-version": "2023-06-01",
}, {
model: input.model,
max_tokens: 4096,
...(input.instructions ? { system: input.instructions } : {}),
messages: [{ role: "user", content: input.prompt }],
});
const text = requireText(collectText(data.content), "Anthropic");
return { text, rawResponseJson: JSON.stringify(data) };
},
},
xai: {
async createResponse(input) {
return createOpenAiCompatibleResponse("https://api.x.ai/v1/chat/completions", "xAI", input);
},
},
groq: {
async createResponse(input) {
return createOpenAiCompatibleResponse("https://api.groq.com/openai/v1/chat/completions", "Groq", input);
},
},
};
async function createOpenAiCompatibleResponse(
endpoint: string,
providerLabel: string,
input: ProviderExecutionInput,
): Promise<ProviderExecutionResult> {
const messages = [
...(input.instructions ? [{ role: "system", content: input.instructions }] : []),
{ role: "user", content: input.prompt },
];
const data = await postJson(endpoint, {
authorization: `Bearer ${input.apiKey}`,
}, {
model: input.model,
messages,
});
const text = requireText(collectText(data.choices), providerLabel);
return { text, rawResponseJson: JSON.stringify(data) };
}
@@ -1,8 +1,11 @@
import type { DatabaseSync } from "node:sqlite";
import type { SaveProviderCredentialInput } from "../../shared/commands";
import { createHash } from "node:crypto";
import type { CreateProviderResponseInput, SaveProviderCredentialInput } from "../../shared/commands";
import { PROVIDER_AUTH_TYPE, PROVIDER_REQUEST_STATE } from "../../shared/domain";
import { createId } from "../../shared/ids";
import { AuditService } from "./audit-service";
import { SecretVault } from "./secret-vault";
import { providerAdapters } from "./provider-adapters";
function displayHint(secret: string): string {
const clean = secret.trim();
@@ -50,4 +53,94 @@ export class ProviderService {
payload: { providerKey: provider.provider_key, authType: provider.auth_type },
});
}
async createResponse(input: CreateProviderResponseInput): Promise<void> {
const prompt = input.prompt.trim();
if (!prompt) throw new Error("Prompt is required.");
const provider = this.db.prepare(`
select id, provider_key, auth_type, model, credential_id, enabled
from providers
where id = ? and org_id = ?
`).get(input.providerId, input.orgId) as ProviderRow | undefined;
if (!provider) throw new Error("Provider was not found.");
if (provider.enabled !== 1) throw new Error("Provider credential is required.");
if (provider.auth_type !== PROVIDER_AUTH_TYPE.API_KEY) {
throw new Error("Provider response execution requires an API key adapter.");
}
if (!provider.credential_id) throw new Error("Provider credential is required.");
if (!provider.model) throw new Error("Provider model is required.");
const adapter = providerAdapters[provider.provider_key];
if (!adapter) throw new Error(`Provider adapter is unavailable for ${provider.provider_key}.`);
const credential = this.db.prepare(`
select ciphertext
from credentials
where id = ? and org_id = ? and provider_key = ?
`).get(provider.credential_id, input.orgId, provider.provider_key) as { ciphertext: Uint8Array } | undefined;
if (!credential) throw new Error("Provider credential was not found.");
const requestId = createId("preq");
const now = new Date().toISOString();
this.db.prepare(`
insert into provider_requests(
id, org_id, provider_id, provider_key, model, prompt_sha256,
response_text, raw_response_json, state, error_message, created_at, completed_at
)
values (?, ?, ?, ?, ?, ?, null, null, ?, null, ?, null)
`).run(
requestId,
input.orgId,
provider.id,
provider.provider_key,
provider.model,
hashPrompt(prompt),
PROVIDER_REQUEST_STATE.PENDING,
now,
);
try {
const apiKey = this.vault.decryptSecret(Buffer.from(credential.ciphertext));
const result = await adapter.createResponse({
apiKey,
model: provider.model,
prompt,
instructions: input.instructions?.trim() || null,
});
this.db.prepare(`
update provider_requests
set state = ?, response_text = ?, raw_response_json = ?, completed_at = ?
where id = ? and org_id = ?
`).run(PROVIDER_REQUEST_STATE.SUCCEEDED, result.text, result.rawResponseJson, new Date().toISOString(), requestId, input.orgId);
this.audit.record({
orgId: input.orgId,
entityType: "provider_request",
entityId: requestId,
actorType: "system",
actorId: null,
action: "response_created",
payload: { providerKey: provider.provider_key, model: provider.model },
});
} catch (error) {
const message = error instanceof Error ? error.message : String(error);
this.db.prepare(`
update provider_requests
set state = ?, error_message = ?, completed_at = ?
where id = ? and org_id = ?
`).run(PROVIDER_REQUEST_STATE.FAILED, message, new Date().toISOString(), requestId, input.orgId);
throw error;
}
}
}
interface ProviderRow {
id: string;
provider_key: string;
auth_type: string;
model: string | null;
credential_id: string | null;
enabled: number;
}
function hashPrompt(prompt: string): string {
return createHash("sha256").update(prompt).digest("hex");
}
@@ -44,13 +44,19 @@ export class SnapshotService {
approvals: list(this.db, "approvals", "created_at desc"),
chatSpaces: list(this.db, "chat_spaces", "created_at asc"),
chatMessages: list(this.db, "chat_messages", "created_at asc"),
chatThreads: list(this.db, "chat_threads", "updated_at desc"),
chatAttachments: list(this.db, "chat_attachments", "message_id asc"),
chatPresence: list(this.db, "chat_presence", "last_seen_at desc"),
files: list(this.db, "file_objects", "created_at desc"),
providers,
providerCatalog: PROVIDER_CATALOG,
providerRequests: list(this.db, "provider_requests", "created_at desc"),
skillDefinitions: list(this.db, "skill_definitions", "created_at asc"),
skillRevisions: list(this.db, "skill_revisions", "published_at desc"),
skillBindings,
skillRuntimeSnapshots: list(this.db, "skill_runtime_snapshots", "created_at desc"),
socialAccounts: list(this.db, "social_accounts", "created_at asc"),
policyRules: list(this.db, "policy_rules", "created_at asc"),
auditEvents: list<AppSnapshot["auditEvents"][number]>(this.db, "audit_events", "created_at desc").slice(0, 200),
};
}