chore(09-02): merge main to get plan 01 infrastructure

- Resolve merge conflicts in worker-service.ts and ipc-core-work-graph-handlers.ts
- Keep granular emitEvent calls from plan 02 while preserving session management and task validation from main
- Fix duplicate createSkillService import in service-graph.ts

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
ZacharyZhang-NYandClaude Opus 4.6 committed 2026-03-27 19:47:33 -04:00
commit ffab9888ed
1169 files changed
+457249 -237

No files matched your search

+16 -1
View File
@@ -8,6 +8,7 @@ import { canAssignTask, canControlAgent, canManageGoal, canManageOrganization, c
import { resolveApprovalDecision, shouldAutoApproveApproval } from "./approval-policy";
import type { AppDatabase } from "./database";
import { inferTaskType } from "./connector-matching";
import { validateTaskCreation } from "./task-quality-gates";
import { dispatchAgentMessage } from "./message-dispatch";
import type { DesktopEvent, SectionId } from "@shared/types";
@@ -355,7 +356,7 @@ export class AgentApiServer {
return;
}
const previousStatus = task.status;
const success = this.db.checkoutTask(params.issueId!, agentId);
const success = this.db.checkoutTask(params.issueId!, token.runId);
if (!success) {
this.json(res, 409, { error: "Task cannot be checked out — already assigned or not in valid status" });
return;
@@ -561,6 +562,12 @@ export class AgentApiServer {
this.json(res, 403, { error: "Current agent is not allowed to assign this task to the selected agent" });
return;
}
const isHeartbeat = input.title.startsWith("[Heartbeat]");
const taskValidation = validateTaskCreation({ title: input.title, description: input.description ?? "" }, isHeartbeat);
if (!taskValidation.valid) {
this.json(res, 422, { error: taskValidation.errors.join("; ") });
return;
}
let id: string;
try {
@@ -1582,6 +1589,14 @@ export class AgentApiServer {
if (actionItems.length > 0) {
for (const item of actionItems) {
if (item.title && typeof item.title === "string") {
const meetingTaskValidation = validateTaskCreation(
{ title: item.title, description: item.description ?? `Action item from meeting: ${data.title ?? existing.title}` },
false,
);
if (!meetingTaskValidation.valid) {
// Log but skip invalid meeting action items rather than failing the whole request
continue;
}
const taskId = this.db.saveTask({
companyId: token.companyId,
title: item.title,
+7
View File
@@ -8,6 +8,7 @@ import type {
RunStatus,
} from "@shared/types";
import { dispatchAgentMessage } from "./message-dispatch";
import { validateTaskCreation } from "./task-quality-gates";
import type { NotificationOptions } from "./notification-service";
/** Document types considered important enough to gate behind user review when reviewDeliverables is enabled. */
@@ -268,6 +269,12 @@ export function createAutomationService({
const description = (config.description as string) ?? "";
const priority = (config.priority as string) ?? "medium";
const assigneeAgentId = (config.assigneeAgentId as string) ?? null;
const isHeartbeat = title.startsWith("[Heartbeat]");
const taskValidation = validateTaskCreation({ title, description }, isHeartbeat);
if (!taskValidation.valid) {
logger.warn(`[automation] Skipping create_task — validation failed: ${taskValidation.errors.join("; ")}`);
break;
}
const taskId = db.saveTask({
companyId: ctx.companyId,
projectId: ctx.projectId ?? (config.projectId as string) ?? null,
+220
View File
@@ -0,0 +1,220 @@
import type { AppDatabase } from "./database";
import type { NotificationOptions } from "./notification-service";
import type { BudgetPolicyRecord, BudgetStatusSummary } from "@shared/types";
interface BudgetEnforcementDependencies {
db: AppDatabase;
logger: { info: (...args: unknown[]) => void; warn: (...args: unknown[]) => void; error: (...args: unknown[]) => void };
notify: (options: NotificationOptions) => void;
publishDomainChanged: () => void;
}
export interface BudgetGateResult {
allowed: boolean;
blockedBy: Array<{
policyId: string;
scopeType: string;
scopeId: string;
spent: number;
limit: number;
}>;
}
/**
* Returns a human-readable label for a budget policy scope.
* Looks up entity names from the database for agent and project scopes.
*/
function getScopeLabel(db: AppDatabase, policy: BudgetPolicyRecord): string {
if (policy.scopeType === "agent") {
try {
const agents = db.loadSnapshot(null).agents;
const agent = agents.find((a) => a.id === policy.scopeId);
return agent ? `Agent "${agent.name}"` : `Agent ${policy.scopeId.slice(0, 8)}`;
} catch {
return `Agent ${policy.scopeId.slice(0, 8)}`;
}
}
if (policy.scopeType === "project") {
try {
const projects = db.loadSnapshot(null).projects;
const project = projects.find((p) => p.id === policy.scopeId);
return project ? `Project "${project.name}"` : `Project ${policy.scopeId.slice(0, 8)}`;
} catch {
return `Project ${policy.scopeId.slice(0, 8)}`;
}
}
return "Company budget";
}
export function createBudgetEnforcementService({
db,
logger,
notify,
publishDomainChanged,
}: BudgetEnforcementDependencies) {
/**
* Evaluates whether a new run should be allowed based on active budget policies.
*
* Per D-01 (strictest-wins): if ANY policy scope is exhausted, the run is blocked.
* Per D-02: if no policies exist, the run is allowed (unlimited).
* Per D-04/D-05: this only blocks NEW dispatches -- in-flight runs are unaffected.
*/
function evaluateBudgetGate(
companyId: string,
agentId: string,
projectId: string | null,
): BudgetGateResult {
const policies = db.getActivePoliciesForRun(companyId, agentId, projectId);
if (policies.length === 0) {
return { allowed: true, blockedBy: [] };
}
const blockedBy: BudgetGateResult["blockedBy"] = [];
for (const policy of policies) {
if (!policy.hardStopEnabled) continue;
const spent = db.getSpentInWindow(policy);
if (spent >= policy.amount) {
blockedBy.push({
policyId: policy.id,
scopeType: policy.scopeType,
scopeId: policy.scopeId,
spent,
limit: policy.amount,
});
}
}
return { allowed: blockedBy.length === 0, blockedBy };
}
/**
* Checks all active policies for the given run context and creates/updates
* budget incidents at warn and hard-stop thresholds.
*
* Per D-07: fires both warn and hard-stop in the same code path.
* Per D-08: deduplicates incidents via upsertBudgetIncident.
*/
function checkAndCreateIncidents(
companyId: string,
agentId: string,
projectId: string | null,
): void {
const policies = db.getActivePoliciesForRun(companyId, agentId, projectId);
for (const policy of policies) {
// Guard against division by zero
if (policy.amount <= 0) continue;
const spent = db.getSpentInWindow(policy);
const utilization = spent / policy.amount;
const warnThreshold = policy.warnPercent / 100;
const scopeLabel = getScopeLabel(db, policy);
if (utilization >= warnThreshold) {
db.upsertBudgetIncident(policy, "warn_threshold", spent);
if (policy.notifyEnabled) {
notify({
title: "Budget Warning",
body: `${scopeLabel} at ${Math.round(utilization * 100)}% of budget ($${(spent / 100).toFixed(2)} / $${(policy.amount / 100).toFixed(2)})`,
urgency: "informational",
navigation: { section: "costs" },
batchKey: "budget_warning",
});
}
}
if (utilization >= 1.0 && policy.hardStopEnabled) {
db.upsertBudgetIncident(policy, "hard_stop", spent);
if (policy.notifyEnabled) {
notify({
title: "Budget Exhausted",
body: `${scopeLabel} budget exhausted - new runs blocked`,
urgency: "critical",
navigation: { section: "costs" },
batchKey: "budget_exhaustion",
});
}
}
}
publishDomainChanged();
}
/**
* Checks if a policy's limit was increased above the observed spend amount,
* and auto-resolves open incidents if so.
*
* Per D-06: fires auto_resolved_limit_increase status.
*/
function checkAutoResolve(companyId: string, policyId: string): void {
const policy = db.getBudgetPolicy(policyId);
if (!policy) return;
const spent = db.getSpentInWindow(policy);
let resolved = false;
// If spend is below hard-stop threshold, resolve all open incidents
if (spent < policy.amount) {
const count = db.resolveIncidentsForPolicy(policyId, "auto_resolved_limit_increase");
if (count > 0) resolved = true;
}
// If spend is below warn threshold, resolve warn incidents specifically
const warnAmount = Math.floor((policy.amount * policy.warnPercent) / 100);
if (spent < warnAmount) {
const count = db.resolveWarnIncidentsForPolicy(policyId, "auto_resolved_limit_increase");
if (count > 0) resolved = true;
}
if (resolved) {
logger.info(`Budget auto-resolve: policy ${policyId} incidents resolved (spent=${spent}, limit=${policy.amount})`);
publishDomainChanged();
}
}
/**
* Closes incidents whose calendar month window has ended.
* Per D-06: window rollover cleanup.
*/
function autoCloseExpiredIncidents(companyId: string): number {
const count = db.autoCloseExpiredIncidents(companyId);
if (count > 0) {
logger.info(`Budget auto-close: ${count} expired incident(s) for company ${companyId}`);
publishDomainChanged();
}
return count;
}
/**
* Returns a summary of all budget policies and their current spend status.
*/
function getBudgetStatus(companyId: string): BudgetStatusSummary {
const policies = db.listBudgetPolicies(companyId);
const openIncidents = db.listBudgetIncidents(companyId, "open");
return {
policies: policies.map((policy) => {
const spent = db.getSpentInWindow(policy);
const hasOpenIncident = openIncidents.some((i) => i.policyId === policy.id);
return {
policyId: policy.id,
scopeType: policy.scopeType,
scopeId: policy.scopeId,
amount: policy.amount,
spent,
warnPercent: policy.warnPercent,
hardStopEnabled: policy.hardStopEnabled,
hasOpenIncident,
};
}),
};
}
return {
evaluateBudgetGate,
checkAndCreateIncidents,
checkAutoResolve,
autoCloseExpiredIncidents,
getBudgetStatus,
};
}
+3
View File
@@ -57,6 +57,7 @@ const CONNECTORS: ConnectorDefinition[] = [
execute: codexExecute,
testEnvironment: codexTestEnvironment,
sessionCodec: codexSessionCodec,
nativeContextManagement: true,
models: codexModels,
},
listModels: async () => codexModels,
@@ -81,6 +82,7 @@ const CONNECTORS: ConnectorDefinition[] = [
execute: claudeExecute,
testEnvironment: claudeTestEnvironment,
sessionCodec: claudeSessionCodec,
nativeContextManagement: true,
models: claudeModels,
},
listModels: async () => claudeModels,
@@ -105,6 +107,7 @@ const CONNECTORS: ConnectorDefinition[] = [
execute: geminiExecute,
testEnvironment: geminiTestEnvironment,
sessionCodec: geminiSessionCodec,
nativeContextManagement: false,
models: geminiModels,
},
listModels: async () => geminiModels,
+1054 -47
View File
File diff suppressed because it is too large. Load diff
+99
View File
@@ -0,0 +1,99 @@
import {
IPC_CHANNELS,
budgetPolicyInputSchema,
listBudgetPoliciesSchema,
deleteBudgetPolicySchema,
listBudgetIncidentsSchema,
resolveBudgetIncidentSchema,
getBudgetStatusSchema,
} from "@shared/contracts";
import { fail, ok, type RegisterDomainHandlersDependencies } from "./ipc-domain-common";
import type { createBudgetEnforcementService } from "./budget-enforcement-service";
import type { AppDatabase } from "./database";
interface BudgetHandlerDependencies extends Pick<
RegisterDomainHandlersDependencies,
"registerHandle" | "ensureCompanyExists" | "publishDomainChanged"
> {
budgetService: ReturnType<typeof createBudgetEnforcementService> | null;
db: AppDatabase;
}
export function registerBudgetHandlers({
registerHandle,
ensureCompanyExists,
publishDomainChanged,
budgetService,
db,
}: BudgetHandlerDependencies) {
function requireBudgetService() {
if (!budgetService) {
return fail("SERVICE_NOT_READY", "Budget enforcement service is not initialized.");
}
return null;
}
// ---- Save (create/update) a budget policy ----
registerHandle(IPC_CHANNELS.saveBudgetPolicy, async (_event, payload) => {
const parsed = budgetPolicyInputSchema.parse(payload);
const companyError = ensureCompanyExists(parsed.companyId);
if (companyError) return companyError;
const id = db.saveBudgetPolicy(parsed);
// D-06: Auto-resolve incidents if limit was increased above current spend
if (budgetService) {
try {
budgetService.checkAutoResolve(parsed.companyId, id);
} catch {
// Best effort -- service may not be ready
}
}
publishDomainChanged();
return ok(id);
});
// ---- Delete a budget policy ----
registerHandle(IPC_CHANNELS.deleteBudgetPolicy, async (_event, payload) => {
const parsed = deleteBudgetPolicySchema.parse(payload);
const companyError = ensureCompanyExists(parsed.companyId);
if (companyError) return companyError;
db.deleteBudgetPolicy(parsed.companyId, parsed.policyId);
publishDomainChanged();
return ok(true);
});
// ---- List all budget policies for a company ----
registerHandle(IPC_CHANNELS.listBudgetPolicies, async (_event, payload) => {
const parsed = listBudgetPoliciesSchema.parse(payload);
const companyError = ensureCompanyExists(parsed.companyId);
if (companyError) return companyError;
return ok(db.listBudgetPolicies(parsed.companyId));
});
// ---- List budget incidents ----
registerHandle(IPC_CHANNELS.listBudgetIncidents, async (_event, payload) => {
const parsed = listBudgetIncidentsSchema.parse(payload);
const companyError = ensureCompanyExists(parsed.companyId);
if (companyError) return companyError;
return ok(db.listBudgetIncidents(parsed.companyId, parsed.status));
});
// ---- Resolve a budget incident manually ----
registerHandle(IPC_CHANNELS.resolveBudgetIncident, async (_event, payload) => {
const parsed = resolveBudgetIncidentSchema.parse(payload);
const companyError = ensureCompanyExists(parsed.companyId);
if (companyError) return companyError;
db.resolveBudgetIncident(parsed.incidentId, parsed.companyId);
publishDomainChanged();
return ok(true);
});
// ---- Get budget status summary ----
registerHandle(IPC_CHANNELS.getBudgetStatus, async (_event, payload) => {
const parsed = getBudgetStatusSchema.parse(payload);
const companyError = ensureCompanyExists(parsed.companyId);
if (companyError) return companyError;
const serviceError = requireBudgetService();
if (serviceError) return serviceError;
return ok(budgetService!.getBudgetStatus(parsed.companyId));
});
}
+67
View File
@@ -4,8 +4,10 @@ import { z } from "zod";
import {
IPC_CHANNELS,
connectorInputSchema,
entityGetterSchema,
metricsQuerySchema,
settingsSchema,
wakeupQuerySchema,
} from "@shared/contracts";
import { fail, ok, type RegisterCoreHandlersDependencies } from "./ipc-core-common";
@@ -142,4 +144,69 @@ export function registerCoreReadHandlers({
if (err) return err;
return ok(db.getStandupReport(companyId));
});
registerHandle(IPC_CHANNELS.listWakeupRequests, (_event, input) => {
const parsed = wakeupQuerySchema.parse(input);
const err = ensureCompanyExists(parsed.companyId);
if (err) return err;
return ok(db.listWakeupRequests(parsed.companyId, {
status: parsed.status,
agentId: parsed.agentId,
limit: parsed.limit,
}));
});
// --- Entity getters for granular IPC updates ---
registerHandle(IPC_CHANNELS.getTask, (_event, payload) => {
const parsed = entityGetterSchema.parse(payload);
const companyError = ensureCompanyExists(parsed.companyId);
if (companyError) return companyError;
try {
const task = db.getTask(parsed.id);
if (task.companyId !== parsed.companyId) return fail("TASK_NOT_FOUND", "Task not found in current company.");
return ok(task);
} catch {
return fail("TASK_NOT_FOUND", "Task not found.");
}
});
registerHandle(IPC_CHANNELS.getAgent, (_event, payload) => {
const parsed = entityGetterSchema.parse(payload);
const companyError = ensureCompanyExists(parsed.companyId);
if (companyError) return companyError;
try {
const agent = db.getAgent(parsed.id);
if (agent.companyId !== parsed.companyId) return fail("AGENT_NOT_FOUND", "Agent not found in current company.");
return ok(agent);
} catch {
return fail("AGENT_NOT_FOUND", "Agent not found.");
}
});
registerHandle(IPC_CHANNELS.getRun, (_event, payload) => {
const parsed = entityGetterSchema.parse(payload);
const companyError = ensureCompanyExists(parsed.companyId);
if (companyError) return companyError;
try {
const run = db.getRun(parsed.id);
if (run.companyId !== parsed.companyId) return fail("RUN_NOT_FOUND", "Run not found in current company.");
return ok(run);
} catch {
return fail("RUN_NOT_FOUND", "Run not found.");
}
});
registerHandle(IPC_CHANNELS.getApproval, (_event, payload) => {
const parsed = entityGetterSchema.parse(payload);
const companyError = ensureCompanyExists(parsed.companyId);
if (companyError) return companyError;
try {
const approval = db.getApprovalRecord(parsed.id);
if (approval.companyId !== parsed.companyId) return fail("APPROVAL_NOT_FOUND", "Approval not found in current company.");
return ok(approval);
} catch {
return fail("APPROVAL_NOT_FOUND", "Approval not found.");
}
});
}
+18 -1
View File
@@ -11,6 +11,7 @@ import {
import { getConnectorExecutionReadinessIssue } from "@shared/connector-policy";
import { resolveApprovalDecision, shouldAutoApproveApproval } from "./approval-policy";
import { fail, ok, type RegisterCoreHandlersDependencies } from "./ipc-core-common";
import { validateTaskCreation, checkCapabilityMatch } from "./task-quality-gates";
export function registerCoreWorkGraphHandlers({
registerHandle,
@@ -316,6 +317,13 @@ export function registerCoreWorkGraphHandlers({
return fail("INVALID_COMPANY_REFERENCE", `The selected ${invalidReference.label} does not belong to the current company.`);
}
const isNew = !parsed.id;
if (isNew) {
const isHeartbeat = parsed.title.startsWith("[Heartbeat]");
const validation = validateTaskCreation(parsed, isHeartbeat);
if (!validation.valid) {
return fail("TASK_VALIDATION_FAILED", validation.errors.join("; "));
}
}
let previousStatus: string | null = null;
let previousAssigneeAgentId: string | null = null;
if (!isNew) {
@@ -357,8 +365,17 @@ export function registerCoreWorkGraphHandlers({
wakeAgentIfPossible(parsed.assigneeAgentId, parsed.companyId, "assignment");
}
}
let capabilityWarning: string | undefined;
if (isNew && parsed.assigneeAgentId) {
try {
const match = checkCapabilityMatch(db, parsed.assigneeAgentId, parsed.description, parsed.companyId);
if (!match.matched) {
capabilityWarning = match.warning;
}
} catch { /* capability check is advisory — never block task creation */ }
}
emitEvent({ type: "task-updated", taskId: id, action: isNew ? "created" : "updated" });
return ok(id);
return ok({ id, capabilityWarning });
});
registerHandle(IPC_CHANNELS.deleteAgent, async (_event, payload) => {
+4
View File
@@ -3,6 +3,8 @@ import { z } from "zod";
import type { DesktopResult } from "@shared/contracts";
import type { bootstrapOnboardingSchema } from "@shared/contracts";
import type { AppDatabase } from "./database";
import type { createSkillService } from "./skill-service";
import type { createBudgetEnforcementService } from "./budget-enforcement-service";
export interface BrowserActionControl {
clearSocialAccountData: (socialAccountId: string, screenshotPaths?: string[]) => Promise<void>;
@@ -38,6 +40,8 @@ export interface RegisterDomainHandlersDependencies {
bootstrapOnboarding: (input: z.infer<typeof bootstrapOnboardingSchema>) => unknown;
getBrowserActions: () => BrowserActionControl | null;
randomFallbackCompanyId: () => string;
skillService: ReturnType<typeof createSkillService> | null;
budgetService: ReturnType<typeof createBudgetEnforcementService> | null;
}
export function ok<T>(data: T): DesktopResult<T> {
+16
View File
@@ -5,6 +5,8 @@ import { registerBrowserHandlers } from "./ipc-browser-handlers";
import { registerContentOperationsHandlers } from "./ipc-content-operations-handlers";
import { registerOnboardingHandlers } from "./ipc-onboarding-handlers";
import { registerRecoveryHandlers } from "./ipc-recovery-handlers";
import { registerSkillHandlers } from "./ipc-skill-handlers";
import { registerBudgetHandlers } from "./ipc-budget-handlers";
export function registerDomainHandlers(dependencies: RegisterDomainHandlersDependencies) {
registerOnboardingHandlers(dependencies);
@@ -13,4 +15,18 @@ export function registerDomainHandlers(dependencies: RegisterDomainHandlersDepen
registerAutomationOperationsHandlers(dependencies);
registerAgentCoordinationHandlers(dependencies);
registerRecoveryHandlers(dependencies);
registerSkillHandlers({
registerHandle: dependencies.registerHandle,
ensureCompanyExists: dependencies.ensureCompanyExists,
publishDomainChanged: dependencies.publishDomainChanged,
skillService: dependencies.skillService,
db: dependencies.db,
});
registerBudgetHandlers({
registerHandle: dependencies.registerHandle,
ensureCompanyExists: dependencies.ensureCompanyExists,
publishDomainChanged: dependencies.publishDomainChanged,
budgetService: dependencies.budgetService,
db: dependencies.db,
});
}
+140
View File
@@ -0,0 +1,140 @@
import {
IPC_CHANNELS,
importSkillSchema,
listSkillsSchema,
getSkillDetailSchema,
deleteSkillSchema,
assignSkillToAgentSchema,
removeSkillFromAgentSchema,
checkSkillUpdateSchema,
listSkillAgentsSchema,
listAgentSkillsSchema,
} from "@shared/contracts";
import { fail, ok, type RegisterDomainHandlersDependencies } from "./ipc-domain-common";
import type { createSkillService } from "./skill-service";
import type { AppDatabase } from "./database";
interface SkillHandlerDependencies extends Pick<
RegisterDomainHandlersDependencies,
"registerHandle" | "ensureCompanyExists" | "publishDomainChanged"
> {
skillService: ReturnType<typeof createSkillService> | null;
db: AppDatabase;
}
export function registerSkillHandlers({
registerHandle,
ensureCompanyExists,
publishDomainChanged,
skillService,
db,
}: SkillHandlerDependencies) {
function requireSkillService() {
if (!skillService) {
return fail("SERVICE_NOT_READY", "Skill service is not initialized.");
}
return null;
}
// ---- Import skill from source (GitHub URL, local path, raw URL, skills.sh) ----
registerHandle(IPC_CHANNELS.importSkill, async (_event, payload) => {
const parsed = importSkillSchema.parse(payload);
const companyError = ensureCompanyExists(parsed.companyId);
if (companyError) return companyError;
const serviceError = requireSkillService();
if (serviceError) return serviceError;
const result = await skillService!.importFromSource(parsed.companyId, parsed.source);
if (result.ok) {
publishDomainChanged();
}
return result;
});
// ---- List all skills for a company ----
registerHandle(IPC_CHANNELS.listSkills, async (_event, payload) => {
const parsed = listSkillsSchema.parse(payload);
const companyError = ensureCompanyExists(parsed.companyId);
if (companyError) return companyError;
const serviceError = requireSkillService();
if (serviceError) return serviceError;
return ok(skillService!.listForCompany(parsed.companyId));
});
// ---- Get skill detail ----
registerHandle(IPC_CHANNELS.getSkillDetail, async (_event, payload) => {
const parsed = getSkillDetailSchema.parse(payload);
const companyError = ensureCompanyExists(parsed.companyId);
if (companyError) return companyError;
const serviceError = requireSkillService();
if (serviceError) return serviceError;
return skillService!.getDetail(parsed.companyId, parsed.skillId);
});
// ---- Delete a skill ----
registerHandle(IPC_CHANNELS.deleteSkill, async (_event, payload) => {
const parsed = deleteSkillSchema.parse(payload);
const companyError = ensureCompanyExists(parsed.companyId);
if (companyError) return companyError;
const serviceError = requireSkillService();
if (serviceError) return serviceError;
const result = await skillService!.deleteSkill(parsed.companyId, parsed.skillId);
if (result.ok) {
publishDomainChanged();
}
return result;
});
// ---- Assign skill to agent ----
registerHandle(IPC_CHANNELS.assignSkillToAgent, async (_event, payload) => {
const parsed = assignSkillToAgentSchema.parse(payload);
const companyError = ensureCompanyExists(parsed.companyId);
if (companyError) return companyError;
const serviceError = requireSkillService();
if (serviceError) return serviceError;
const result = skillService!.assignToAgent(parsed.companyId, parsed.agentId, parsed.skillId);
if (result.ok) {
publishDomainChanged();
}
return result;
});
// ---- Remove skill from agent ----
registerHandle(IPC_CHANNELS.removeSkillFromAgent, async (_event, payload) => {
const parsed = removeSkillFromAgentSchema.parse(payload);
const companyError = ensureCompanyExists(parsed.companyId);
if (companyError) return companyError;
const serviceError = requireSkillService();
if (serviceError) return serviceError;
const result = skillService!.removeFromAgent(parsed.companyId, parsed.agentId, parsed.skillId);
if (result.ok) {
publishDomainChanged();
}
return result;
});
// ---- Check if skill has update available ----
registerHandle(IPC_CHANNELS.checkSkillUpdate, async (_event, payload) => {
const parsed = checkSkillUpdateSchema.parse(payload);
const companyError = ensureCompanyExists(parsed.companyId);
if (companyError) return companyError;
const serviceError = requireSkillService();
if (serviceError) return serviceError;
return skillService!.checkUpdateStatus(parsed.companyId, parsed.skillId);
});
// ---- List agent IDs assigned to a skill ----
registerHandle(IPC_CHANNELS.listSkillAgents, async (_event, payload) => {
const parsed = listSkillAgentsSchema.parse(payload);
const companyError = ensureCompanyExists(parsed.companyId);
if (companyError) return companyError;
return ok(db.listSkillAgentAssignments(parsed.skillId));
});
// ---- List skill IDs assigned to an agent ----
registerHandle(IPC_CHANNELS.listAgentSkills, async (_event, payload) => {
const parsed = listAgentSkillsSchema.parse(payload);
const companyError = ensureCompanyExists(parsed.companyId);
if (companyError) return companyError;
return ok(db.listAgentSkillAssignments(parsed.agentId));
});
}
+2
View File
@@ -168,6 +168,8 @@ export function createMainRuntimeSupport(state: MainRuntimeState) {
bootstrapOnboarding: (input) => requireOnboardingService().bootstrapOnboarding(input),
getBrowserActions: () => requireServiceGraph().getBrowserActionService(),
randomFallbackCompanyId: () => requireConnectorHealthService().randomFallbackCompanyId(),
skillService: requireServiceGraph().getOrCreateSkillService(),
budgetService: requireServiceGraph().getOrCreateBudgetEnforcementService(),
});
}
+162 -20
View File
@@ -1,9 +1,14 @@
import { randomUUID } from "node:crypto";
import { buildTaskPrompt } from "./prompt-builder";
import type { AppDatabase } from "./database";
import { getConnectorExecutionReadinessIssue } from "@shared/connector-policy";
import type { DesktopResult } from "@shared/contracts";
import type { ConnectorRecord } from "@shared/types";
import type { NotificationOptions } from "./notification-service";
import type { createBudgetEnforcementService } from "./budget-enforcement-service";
import { sweepStaleReviews } from "./task-quality-gates";
import { resolveApprovalDecision } from "./approval-policy";
import { dispatchAgentMessage } from "./message-dispatch";
type WorkspaceLockMode = "exclusive" | "shared" | "none";
@@ -44,6 +49,7 @@ interface EnqueueRunPayload {
sessionParams: Record<string, unknown> | null;
systemPrompt: string | null;
trigger: string | null;
managedSkills: Array<{ runtimeName: string; sourcePath: string }>;
}
interface OrchestrationDependencies {
@@ -62,6 +68,9 @@ interface OrchestrationDependencies {
hasWorkerReady: () => boolean;
enqueueWorkerRun: (payload: EnqueueRunPayload) => void;
isQuitting: () => boolean;
listRuntimeSkillEntries: (companyId: string, agentId: string, connectorId: string) => Array<{ runtimeName: string; sourcePath: string }>;
budgetService: ReturnType<typeof createBudgetEnforcementService>;
getSessionHandoff?: (taskId: string) => string | null;
}
export function createOrchestrationService({
@@ -76,6 +85,9 @@ export function createOrchestrationService({
hasWorkerReady,
enqueueWorkerRun,
isQuitting,
listRuntimeSkillEntries,
budgetService,
getSessionHandoff,
}: OrchestrationDependencies) {
let heartbeatTimer: ReturnType<typeof setInterval> | null = null;
const scheduleLastFiredMap = new Map<string, number>();
@@ -120,6 +132,7 @@ export function createOrchestrationService({
return { workspace: null, lockMode: "none", source: "none" };
}
// DEPRECATED: Superseded by BudgetEnforcementService.evaluateBudgetGate (Phase 5)
function autoPauseBudgetExceededAgent(agent: { id: string; companyId: string; name: string; spentMonthlyUsd: number; budgetMonthlyUsd: number }) {
if (!(agent.budgetMonthlyUsd > 0) || agent.spentMonthlyUsd < agent.budgetMonthlyUsd) {
return false;
@@ -154,14 +167,16 @@ export function createOrchestrationService({
function getRunEligibilityFailure(
agent: { status: string; budgetMonthlyUsd: number; spentMonthlyUsd: number; name: string; connectorId: string; id: string; companyId: string },
connector: Pick<ConnectorRecord, "id" | "label" | "status" | "capabilityMatrix">,
projectId: string | null = null,
) {
if (["terminated", "paused", "pending_approval"].includes(agent.status)) {
return "AGENT_NOT_RUNNABLE";
}
if (agent.budgetMonthlyUsd > 0 && agent.spentMonthlyUsd >= agent.budgetMonthlyUsd) {
autoPauseBudgetExceededAgent(agent);
logger.info(`[run] Agent ${agent.name} budget exhausted, skipping`);
return "AGENT_BUDGET_EXHAUSTED";
// Policy-based budget check replaces flat field check (D-03, D-04, D-05)
const gate = budgetService.evaluateBudgetGate(agent.companyId, agent.id, projectId);
if (!gate.allowed) {
logger.info(`[run] Agent ${agent.name} budget policy blocked: ${gate.blockedBy.map(b => `${b.scopeType}(${b.policyId})`).join(", ")}`);
return "BUDGET_POLICY_BLOCKED";
}
const connectorIssue = getConnectorExecutionReadinessIssue(connector);
if (connectorIssue) {
@@ -221,6 +236,16 @@ export function createOrchestrationService({
createdAt: comment.createdAt,
}));
// D-07: Load handoff from previous rotated session
let sessionHandoff: string | null = null;
if (getSessionHandoff) {
try {
sessionHandoff = getSessionHandoff(task.id);
} catch (err) {
logger.warn(`[run] Failed to load session handoff for task ${task.id}: ${err instanceof Error ? err.message : String(err)}`);
}
}
const systemPrompt = apiUrl ? buildTaskPrompt({
agent,
apiUrl,
@@ -272,6 +297,7 @@ export function createOrchestrationService({
.filter((document) => document.companyId === companyId && (document.authorAgentId === agent.id || document.reviewerAgentId === agent.id || document.status === "in_review"))
.slice(0, 5),
allConnectors: snapshot.connectors.map((c) => ({ id: c.id, label: c.label, status: c.status })),
sessionHandoff,
}) : null;
if (apiUrl) {
@@ -284,7 +310,14 @@ export function createOrchestrationService({
env.AGENT_COMPANY_WAKE_REASON = trigger;
}
const taskSession = db.getTaskSession(task.id);
// SESS-01: Prefer task_sessions table, fall back to legacy tasks.session_json
const activeSession = db.getActiveTaskSession(task.id);
const taskSession = activeSession
? {
sessionParams: activeSession.sessionParamsJson ? JSON.parse(activeSession.sessionParamsJson) as Record<string, unknown> : null,
sessionDisplayId: activeSession.sessionDisplayId,
}
: db.getTaskSession(task.id);
let thinkingEffort: string | null = null;
try {
const meta = JSON.parse(agent.metadataJson || "{}");
@@ -295,6 +328,14 @@ export function createOrchestrationService({
thinkingEffort = null;
}
// Fetch managed skills for this agent -- best-effort, never blocks run dispatch
let managedSkills: Array<{ runtimeName: string; sourcePath: string }> = [];
try {
managedSkills = listRuntimeSkillEntries(companyId, agent.id, connector.id);
} catch (err) {
logger.warn(`[run] Failed to fetch managed skills for agent ${agent.name}: ${err instanceof Error ? err.message : String(err)}`);
}
enqueueWorkerRun({
runId,
companyId,
@@ -314,6 +355,7 @@ export function createOrchestrationService({
sessionParams: taskSession.sessionParams,
systemPrompt,
trigger,
managedSkills,
});
}
@@ -331,7 +373,7 @@ export function createOrchestrationService({
return null;
}
const connector = db.getConnector(agent.connectorId);
if (getRunEligibilityFailure(agent, connector)) {
if (getRunEligibilityFailure(agent, connector, null)) {
return null;
}
if (hasActiveRun({ agentId: agent.id, companyId })) {
@@ -408,8 +450,20 @@ export function createOrchestrationService({
logger.info(`[heartbeat] Created bootstrap task for ${agent.name}: ${taskTitle}`);
}
if (task.activeRunId || hasActiveRun({ taskId: task.id, companyId })) {
logger.info(`[heartbeat] Skipping ${agent.name} because task ${task.title} already has an active run.`);
// Generate runId upfront for CAS checkout (EXEC-01, EXEC-02)
const runId = randomUUID();
const checkoutSuccess = db.checkoutTask(task.id, runId);
if (!checkoutSuccess) {
// D-04: Activity log entry only, no toast/popup
db.addActivity({
companyId,
actor: "system",
action: "execution.checkout_rejected",
entityType: "task",
entityId: task.id,
detail: `Agent ${agent.name} lost checkout race on task "${task.title}"`,
});
logger.info(`[heartbeat] Agent ${agent.name} lost checkout race on task ${task.title}`);
return null;
}
@@ -420,6 +474,7 @@ export function createOrchestrationService({
);
const run = db.createRun({
id: runId,
companyId,
taskId: task.id,
agentId: agent.id,
@@ -473,9 +528,16 @@ export function createOrchestrationService({
}
const agentBusy = hasActiveRun({ agentId, companyId });
if (agentBusy) {
// Agent is busy — enqueue wakeup with lifecycle tracking (D-05)
db.enqueueWakeupRequest(companyId, agentId, trigger, `Wake for agent: ${trigger}`);
return null;
}
const runId = startAgentHeartbeatRun(agentId, companyId, trigger);
if (!runId && agentBusy) {
db.enqueuePendingWake(companyId, agentId, trigger);
if (!runId) {
// Run failed to start but agent not busy — still enqueue for retry
db.enqueueWakeupRequest(companyId, agentId, trigger, `Wake for agent (deferred): ${trigger}`);
}
return runId;
}
@@ -493,21 +555,18 @@ export function createOrchestrationService({
if (companyStatus !== "active") {
return fail("COMPANY_NOT_ACTIVE", "Runs can only start while the company is active.");
}
if (task.activeRunId || hasActiveRun({ taskId: task.id, companyId: task.companyId })) {
return fail("TASK_ALREADY_RUNNING", "This task already has an active run.");
}
if (!task.assigneeAgentId) return fail("TASK_UNASSIGNED", "Assign an agent before starting a run.");
const agent = db.getAgent(task.assigneeAgentId);
if (agent.companyId !== task.companyId) {
return fail("AGENT_COMPANY_MISMATCH", "The assigned agent does not belong to the task company.");
}
const connector = db.getConnector(agent.connectorId);
const eligibilityFailure = getRunEligibilityFailure(agent, connector);
const eligibilityFailure = getRunEligibilityFailure(agent, connector, task.projectId ?? null);
if (eligibilityFailure === "AGENT_NOT_RUNNABLE") {
return fail("AGENT_NOT_RUNNABLE", "The assigned agent is paused, pending approval, or terminated.");
}
if (eligibilityFailure === "AGENT_BUDGET_EXHAUSTED") {
return fail("AGENT_BUDGET_EXHAUSTED", "The assigned agent has exhausted its monthly budget.");
if (eligibilityFailure === "BUDGET_POLICY_BLOCKED") {
return fail("BUDGET_POLICY_BLOCKED", "A budget policy blocks this run. Review budget settings.");
}
if (eligibilityFailure === "CONNECTOR_NOT_READY") {
return fail("CONNECTOR_NOT_READY", "The assigned agent connector is not ready.");
@@ -526,6 +585,13 @@ export function createOrchestrationService({
);
}
// CAS checkout — atomic claim (EXEC-01, EXEC-02)
const runId = randomUUID();
const checkoutSuccess = db.checkoutTask(task.id, runId);
if (!checkoutSuccess) {
return fail("CHECKOUT_RACE_LOST", "Another run claimed this task. Refresh and try again.");
}
const workspaceResolution = resolveWorkspace(task, agent, task.companyId);
const workspace = workspaceResolution.workspace;
if (workspace && workspace.companyId !== task.companyId) {
@@ -533,6 +599,7 @@ export function createOrchestrationService({
}
const run = db.createRun({
id: runId,
companyId: task.companyId,
taskId: task.id,
agentId: agent.id,
@@ -567,6 +634,27 @@ export function createOrchestrationService({
}
function resumeQueuedRuns() {
// D-02: Stale lock recovery at app startup (EXEC-03)
try {
const recovered = db.recoverStaleLocks();
for (const task of recovered) {
db.addActivity({
companyId: task.companyId,
actor: "system",
action: "execution.lock_recovered",
entityType: "task",
entityId: task.taskId,
detail: `Recovered stale lock on task "${task.title}" after 15min timeout (startup sweep)`,
});
logger.info(`[init] Recovered stale lock on task ${task.taskId}: "${task.title}"`);
}
if (recovered.length > 0) {
logger.info(`[init] Recovered ${recovered.length} stale lock(s) at startup`);
}
} catch (err) {
logger.error(`[init] Stale lock recovery at startup failed: ${err instanceof Error ? err.message : String(err)}`);
}
if (isQuitting() || !hasWorkerReady()) {
return;
}
@@ -593,7 +681,7 @@ export function createOrchestrationService({
const task = db.getTask(run.taskId);
const agent = db.getAgent(run.agentId);
const connector = db.getConnector(run.connectorId);
if (getRunEligibilityFailure(agent, connector)) {
if (getRunEligibilityFailure(agent, connector, task.projectId ?? null)) {
continue;
}
const workspaceResolution = resolveWorkspace(task, agent, run.companyId);
@@ -622,6 +710,26 @@ export function createOrchestrationService({
heartbeatTimer = setInterval(() => {
try {
if (isQuitting() || !hasWorkerReady()) return;
// D-02: Stale lock recovery piggybacked on heartbeat cycle (EXEC-03)
try {
const recovered = db.recoverStaleLocks();
for (const task of recovered) {
// D-04: Activity log entry only
db.addActivity({
companyId: task.companyId,
actor: "system",
action: "execution.lock_recovered",
entityType: "task",
entityId: task.taskId,
detail: `Recovered stale lock on task "${task.title}" after 15min timeout`,
});
logger.info(`[heartbeat] Recovered stale lock on task ${task.taskId}: "${task.title}"`);
}
} catch (err) {
logger.error(`[heartbeat] Stale lock recovery failed: ${err instanceof Error ? err.message : String(err)}`);
}
const activeCompanies = db.listActiveCompanyIds();
const now = Date.now();
@@ -680,6 +788,40 @@ export function createOrchestrationService({
db.setLastBudgetResetMonth(company.id, currentMonth);
logger.info(`[heartbeat] Reset monthly budgets for company ${company.name} (${currentMonth})`);
}
// D-06: Auto-close incidents from previous budget windows
const closedIncidents = budgetService.autoCloseExpiredIncidents(company.id);
if (closedIncidents > 0) {
logger.info(`[heartbeat] Auto-closed ${closedIncidents} expired budget incident(s) for company ${company.name}`);
}
// TASK-04: Stale review escalation sweep
try {
const escalated = sweepStaleReviews(company.id, {
db: db as unknown as import("./task-quality-gates").SweepStaleReviewsDeps["db"],
dispatchAgentMessage: dispatchAgentMessage as unknown as import("./task-quality-gates").SweepStaleReviewsDeps["dispatchAgentMessage"],
resolveApprovalDecision: resolveApprovalDecision as unknown as import("./task-quality-gates").SweepStaleReviewsDeps["resolveApprovalDecision"],
notify: notify as unknown as import("./task-quality-gates").SweepStaleReviewsDeps["notify"],
handleTaskStatusChange: (_taskId: string, _companyId: string, _prev: string, _next: string) => {
publishDomainChanged();
},
handleApprovalResolved: (_approvalId: string, _companyId: string, _decision: string) => {
publishDomainChanged();
},
emitEvent: (..._args: unknown[]) => {},
publishDomainChanged,
wakeAgentIfPossible: (agentId: string, cId: string, trigger: string) => {
wakeAgentIfPossible(agentId, cId, trigger);
},
logger,
});
if (escalated > 0) {
logger.info(`[heartbeat] Auto-escalated ${escalated} stale review(s) for company ${company.name}`);
publishDomainChanged();
}
} catch (error) {
logger.error(`[heartbeat] Stale review sweep failed: ${error instanceof Error ? error.message : String(error)}`);
}
} catch (error) {
logger.error(`[heartbeat] Error processing company ${company.id}: ${error instanceof Error ? error.message : String(error)}`);
}
@@ -691,9 +833,9 @@ export function createOrchestrationService({
logger.info(`[heartbeat] Released ${staleReleased} stale workspace lock(s)`);
publishDomainChanged();
}
const stalePendingWakes = db.purgeExpiredPendingWakes();
if (stalePendingWakes > 0) {
logger.info(`[heartbeat] Purged ${stalePendingWakes} stale pending wake(s)`);
const staleWakeups = db.purgeExpiredWakeupRequests();
if (staleWakeups > 0) {
logger.info(`[heartbeat] Purged ${staleWakeups} expired wakeup request(s)`);
}
} catch (error) {
logger.error(`[heartbeat] Stale lock cleanup failed: ${error instanceof Error ? error.message : String(error)}`);
+10
View File
@@ -42,6 +42,7 @@ interface RunJob {
sessionParams: Record<string, unknown> | null;
systemPrompt: string | null;
trigger: string | null;
managedSkills: Array<{ runtimeName: string; sourcePath: string }>;
}
type ParentMessage =
@@ -64,6 +65,8 @@ type ChildMessage =
sessionDisplayId: string | null;
costUsd: number | null;
unattributedCost: boolean;
inputTokens: number | null;
outputTokens: number | null;
};
const parentPort = process.parentPort;
@@ -95,6 +98,7 @@ function buildConnectorExecutionConfig(job: RunJob, sanitizedEnv: Record<string,
cwd: job.workspacePath ?? process.cwd(),
model: job.model,
env: sanitizedEnv,
managedSkills: job.managedSkills,
};
switch (job.connectorId) {
@@ -234,6 +238,8 @@ async function startRun(job: RunJob) {
sessionDisplayId: result.sessionDisplayId ?? null,
costUsd: typeof result.costUsd === "number" ? result.costUsd : null,
unattributedCost: !result.costUsd,
inputTokens: result.usage?.inputTokens ?? null,
outputTokens: result.usage?.outputTokens ?? null,
});
} catch (error) {
post({
@@ -249,6 +255,8 @@ async function startRun(job: RunJob) {
sessionDisplayId: null,
costUsd: null,
unattributedCost: true,
inputTokens: null,
outputTokens: null,
});
} finally {
cancelledRuns.delete(job.runId);
@@ -277,6 +285,8 @@ function cancelRun(runId: string) {
sessionDisplayId: null,
costUsd: null,
unattributedCost: true,
inputTokens: null,
outputTokens: null,
});
return;
}
+11
View File
@@ -31,6 +31,8 @@ interface PromptContext {
activeSprint?: SprintRecord | null;
recentDocuments?: DocumentRecord[];
allConnectors?: Array<{ id: string; label: string; status: string }>;
/** D-07: Handoff markdown from a rotated session, injected as context for the agent */
sessionHandoff?: string | null;
}
export function buildSystemPrompt(ctx: PromptContext): string {
@@ -195,6 +197,15 @@ ${relevantProjects.map((p) => `- [${p.id}] ${p.name}: ${p.description || ""} (st
`);
}
// D-07: Inject handoff context from previous rotated session
if (ctx.sessionHandoff) {
sections.push(`# Session Handoff
The following context was preserved from your previous session on this task:
${ctx.sessionHandoff}
`);
}
if (ctx.task) {
sections.push(`# Current Task
- ID: ${ctx.task.id}
+69 -1
View File
@@ -8,9 +8,12 @@ import { createConnectorHealthService } from "./connector-health-service";
import { createDesktopShellService } from "./desktop-shell-service";
import { createOnboardingService } from "./onboarding-service";
import { createOrchestrationService } from "./orchestration-service";
import { createSkillService } from "./skill-service";
import { createUpdaterService } from "./updater-service";
import { createWorkerService } from "./worker-service";
import { createAppLifecycleService } from "./app-lifecycle-service";
import { createBudgetEnforcementService } from "./budget-enforcement-service";
import { createSessionManagementService } from "./session-management-service";
import type { DesktopEvent, RunRecord } from "@shared/types";
import type { AutomationEventContext } from "./automation-service";
import type { NotificationOptions } from "./notification-service";
@@ -67,8 +70,44 @@ export function createServiceGraph({
let workerService: ReturnType<typeof createWorkerService> | null = null;
let updaterService: ReturnType<typeof createUpdaterService> | null = null;
let appLifecycleService: ReturnType<typeof createAppLifecycleService> | null = null;
let skillService: ReturnType<typeof createSkillService> | null = null;
let budgetEnforcementService: ReturnType<typeof createBudgetEnforcementService> | null = null;
let sessionManagementService: ReturnType<typeof createSessionManagementService> | null = null;
const wakeAgentIfPossible = (agentId: string | null | undefined, companyId: string, trigger: string) => {
function getOrCreateSkillService() {
if (!skillService) {
skillService = createSkillService({
db,
profileDir: db.profileDir,
logger,
});
}
return skillService;
}
function getOrCreateBudgetEnforcementService() {
if (!budgetEnforcementService) {
budgetEnforcementService = createBudgetEnforcementService({
db,
logger,
notify,
publishDomainChanged,
});
}
return budgetEnforcementService;
}
function getOrCreateSessionManagementService() {
if (!sessionManagementService) {
sessionManagementService = createSessionManagementService({
db,
logger,
});
}
return sessionManagementService;
}
const wakeAgentIfPossible =(agentId: string | null | undefined, companyId: string, trigger: string) => {
if (!orchestrationService) {
throw new Error("Orchestration service is not initialized.");
}
@@ -163,6 +202,8 @@ export function createServiceGraph({
dispatchRunToWorker,
processRunCompletion: (run, event) => processRunCompletion(run, event),
notify,
budgetService: getOrCreateBudgetEnforcementService(),
sessionService: getOrCreateSessionManagementService(),
});
orchestrationService = createOrchestrationService({
@@ -179,6 +220,27 @@ export function createServiceGraph({
workerService?.getWorker()?.postMessage({ type: "enqueue-run", payload });
},
isQuitting: () => isQuittingRef.current,
listRuntimeSkillEntries: (() => {
let cachedSkillService: ReturnType<typeof createSkillService> | null = null;
return (companyId: string, agentId: string, connectorId: string) => {
try {
if (!cachedSkillService) {
cachedSkillService = createSkillService({ db, profileDir: db.profileDir, logger });
}
return cachedSkillService.listRuntimeSkillEntries(companyId, agentId, connectorId);
} catch {
return [];
}
};
})(),
budgetService: getOrCreateBudgetEnforcementService(),
getSessionHandoff: (taskId: string) => {
try {
return getOrCreateSessionManagementService().getHandoffContext(taskId);
} catch {
return null;
}
},
});
workerService.createWorker();
@@ -237,6 +299,9 @@ export function createServiceGraph({
orchestrationService = null;
updaterService = null;
appLifecycleService = null;
skillService = null;
budgetEnforcementService = null;
sessionManagementService = null;
db.close();
// Force exit after grace period — covers all platforms (macOS Dock, Windows tray, Linux)
// Worker child processes (claude/codex CLI) may keep the event loop alive indefinitely
@@ -256,5 +321,8 @@ export function createServiceGraph({
getOnboardingService: () => onboardingService,
getUpdaterService: () => updaterService,
getAppLifecycleService: () => appLifecycleService,
getOrCreateSkillService,
getOrCreateBudgetEnforcementService,
getOrCreateSessionManagementService,
};
}
+207
View File
@@ -0,0 +1,207 @@
import type { AppDatabase } from "./database";
import { getConnectorDefinition } from "./connectors";
import type { ConnectorId, TaskSessionRecord } from "@shared/types";
interface SessionManagementDependencies {
db: AppDatabase;
logger: { info: (...args: unknown[]) => void; warn: (...args: unknown[]) => void; error: (...args: unknown[]) => void };
}
// D-01/D-02: Exported for testing and external override. Conservative defaults.
export const DEFAULT_THRESHOLDS = {
maxRuns: 20,
maxTokens: 500_000,
maxAgeHours: 72,
};
// D-01: Type for per-adapter or company-level threshold overrides
export type RotationThresholds = Partial<typeof DEFAULT_THRESHOLDS>;
// D-04: Handoff size cap to prevent prompt bloat
const MAX_HANDOFF_LENGTH = 2000;
const MAX_HANDOFF_RUNS = 5;
export function createSessionManagementService({ db, logger }: SessionManagementDependencies) {
// D-08: Accumulate token usage on task_sessions row (called for ALL adapters per D-05)
function accumulateTokenUsage(sessionId: string, inputTokens: number, outputTokens: number, runId: string): void {
db.incrementSessionUsage(sessionId, inputTokens, outputTokens, runId);
}
// D-01: Any-of thresholds with optional overrides. D-04: Check nativeContextManagement.
// Per D-01: "Configurable per-adapter with company-level overrides" -- accepts optional
// thresholds parameter that merges with defaults. Callers can pass per-adapter or
// company-level overrides; omitted fields fall back to DEFAULT_THRESHOLDS.
function evaluateRotation(
session: TaskSessionRecord,
connectorId: string,
thresholdsOverride?: RotationThresholds,
): { shouldRotate: boolean; reason: string | null } {
// D-04: Check adapter exemption
try {
const connector = getConnectorDefinition(connectorId as ConnectorId);
if (connector.module.nativeContextManagement === true) {
return { shouldRotate: false, reason: null };
}
} catch {
// Unknown connector - proceed with rotation check
}
// Merge overrides with defaults (per D-01 configurability)
const thresholds = { ...DEFAULT_THRESHOLDS, ...thresholdsOverride };
// D-01: Any-of thresholds - first trigger wins
if (session.runCount >= thresholds.maxRuns) {
return { shouldRotate: true, reason: "max_runs" };
}
const totalTokens = session.totalInputTokens + session.totalOutputTokens;
if (totalTokens >= thresholds.maxTokens) {
return { shouldRotate: true, reason: "max_tokens" };
}
const ageMs = Date.now() - new Date(session.createdAt).getTime();
const ageHours = ageMs / (1000 * 60 * 60);
if (ageHours >= thresholds.maxAgeHours) {
return { shouldRotate: true, reason: "max_age_hours" };
}
return { shouldRotate: false, reason: null };
}
// D-06: Build handoff markdown from structured data, no AI call
function generateHandoff(taskId: string, session: TaskSessionRecord): string {
let task: { title: string; description: string } | null = null;
try {
task = db.getTask(taskId);
} catch {
task = null;
}
const recentRuns = db.getRecentRunsForTask(taskId, MAX_HANDOFF_RUNS);
const errorRuns = recentRuns.filter(r => r.errorMessage);
const activityEntries = db.getRecentActivityForTask(taskId, 10);
const decisionEntries = activityEntries.filter(a =>
a.action.includes("decision") || a.action.includes("approval") || a.action.includes("resolved")
);
const sections: string[] = [];
sections.push("# Session Handoff");
sections.push("");
if (task) {
sections.push("## Task");
sections.push(`**${task.title}**`);
const descSnippet = task.description.length > 200 ? task.description.slice(0, 200) + "..." : task.description;
sections.push(descSnippet);
sections.push("");
}
if (recentRuns.length > 0) {
sections.push("## Recent Run History");
for (const run of recentRuns) {
const summaryText = run.summary ? run.summary.slice(0, 100) : "(no summary)";
sections.push(`- [${run.status}] ${summaryText} (${run.finishedAt ?? "in progress"})`);
}
sections.push("");
}
if (errorRuns.length > 0) {
sections.push("## Errors Encountered");
for (const run of errorRuns) {
sections.push(`- ${run.errorMessage!.slice(0, 150)} (${run.finishedAt ?? "unknown"})`);
}
sections.push("");
}
if (decisionEntries.length > 0) {
sections.push("## Key Decisions");
for (const entry of decisionEntries.slice(0, 5)) {
sections.push(`- ${entry.detail.slice(0, 120)} (${entry.createdAt})`);
}
sections.push("");
}
sections.push(`*Session stats: ${session.runCount} runs, ${session.totalInputTokens + session.totalOutputTokens} total tokens*`);
let result = sections.join("\n");
// Cap handoff size per research pitfall #5
if (result.length > MAX_HANDOFF_LENGTH) {
result = result.slice(0, MAX_HANDOFF_LENGTH - 50) + "\n\n... (earlier content omitted)";
}
return result;
}
// Archive current session, generate handoff, create new session, update task pointer
function rotateSession(session: TaskSessionRecord): string {
const reason = "threshold_exceeded";
const handoff = generateHandoff(session.taskId, session);
logger.info(`[session] Rotating session ${session.id} for task ${session.taskId}: ${reason}`);
// 1. Archive old session
db.archiveSession(session.id, reason, handoff);
// 2. Create new session row with fresh counters, carry forward session_params
const newSessionId = db.createNewSession({
taskId: session.taskId,
companyId: session.companyId,
agentId: session.agentId,
adapterType: session.adapterType,
sessionParamsJson: session.sessionParamsJson,
sessionDisplayId: session.sessionDisplayId,
});
logger.info(`[session] New session ${newSessionId} created for task ${session.taskId}`);
return newSessionId;
}
// D-07: Read handoff from most recently rotated session for a task
function getHandoffContext(taskId: string): string | null {
// Find the most recent archived session for this task (the one just before current)
const row = db.queryAll<{ handoffMarkdown: string | null }>(
`SELECT handoff_markdown as handoffMarkdown FROM task_sessions
WHERE task_id = ? AND rotated_at IS NOT NULL
ORDER BY rotated_at DESC LIMIT 1`,
taskId
);
return row[0]?.handoffMarkdown ?? null;
}
// D-09: Snapshot token totals for delta tracking
function snapshotUsage(session: TaskSessionRecord): { inputTokens: number; outputTokens: number } {
return {
inputTokens: session.totalInputTokens,
outputTokens: session.totalOutputTokens,
};
}
// D-09: Compute delta from snapshot
function getUsageDelta(
sessionId: string,
snapshot: { inputTokens: number; outputTokens: number }
): { inputTokensDelta: number; outputTokensDelta: number } {
// Re-read current totals from DB
const current = db.queryAll<{ totalInputTokens: number; totalOutputTokens: number }>(
`SELECT total_input_tokens as totalInputTokens, total_output_tokens as totalOutputTokens
FROM task_sessions WHERE id = ?`,
sessionId
);
if (!current[0]) return { inputTokensDelta: 0, outputTokensDelta: 0 };
return {
inputTokensDelta: current[0].totalInputTokens - snapshot.inputTokens,
outputTokensDelta: current[0].totalOutputTokens - snapshot.outputTokens,
};
}
return {
accumulateTokenUsage,
evaluateRotation,
generateHandoff,
rotateSession,
getHandoffContext,
snapshotUsage,
getUsageDelta,
};
}
+243
View File
@@ -0,0 +1,243 @@
// ---------------------------------------------------------------------------
// SKILL.md format parser, inventory classifier, and trust level derivation
// ---------------------------------------------------------------------------
import { extname } from "node:path";
import type { FileInventoryEntry, TrustLevel } from "./skill-types.js";
// ---------------------------------------------------------------------------
// Minimal YAML frontmatter parsing (no external YAML library)
// Follows the Paperclip reference pattern for SKILL.md files.
// ---------------------------------------------------------------------------
function isPlainRecord(value: unknown): value is Record<string, unknown> {
return typeof value === "object" && value !== null && !Array.isArray(value);
}
function parseYamlScalar(rawValue: string): unknown {
const trimmed = rawValue.trim();
if (trimmed === "") return "";
if (trimmed === "null" || trimmed === "~") return null;
if (trimmed === "true") return true;
if (trimmed === "false") return false;
if (trimmed === "[]") return [];
if (trimmed === "{}") return {};
if (/^-?\d+(\.\d+)?$/.test(trimmed)) return Number(trimmed);
if (
trimmed.startsWith('"') ||
trimmed.startsWith("[") ||
trimmed.startsWith("{")
) {
try {
return JSON.parse(trimmed);
} catch {
return trimmed;
}
}
return trimmed;
}
function prepareYamlLines(
raw: string,
): Array<{ indent: number; content: string }> {
return raw
.split("\n")
.map((line) => ({
indent: line.match(/^ */)?.[0].length ?? 0,
content: line.trim(),
}))
.filter((line) => line.content.length > 0 && !line.content.startsWith("#"));
}
function parseYamlBlock(
lines: Array<{ indent: number; content: string }>,
startIndex: number,
indentLevel: number,
): { value: unknown; nextIndex: number } {
let index = startIndex;
while (index < lines.length && lines[index]!.content.length === 0)
index += 1;
if (index >= lines.length || lines[index]!.indent < indentLevel) {
return { value: {}, nextIndex: index };
}
const isArray =
lines[index]!.indent === indentLevel &&
lines[index]!.content.startsWith("-");
if (isArray) {
const values: unknown[] = [];
while (index < lines.length) {
const line = lines[index]!;
if (line.indent < indentLevel) break;
if (line.indent !== indentLevel || !line.content.startsWith("-")) break;
const remainder = line.content.slice(1).trim();
index += 1;
if (!remainder) {
const nested = parseYamlBlock(lines, index, indentLevel + 2);
values.push(nested.value);
index = nested.nextIndex;
continue;
}
const inlineObjectSeparator = remainder.indexOf(":");
if (
inlineObjectSeparator > 0 &&
!remainder.startsWith('"') &&
!remainder.startsWith("{") &&
!remainder.startsWith("[")
) {
const key = remainder.slice(0, inlineObjectSeparator).trim();
const rawValue = remainder.slice(inlineObjectSeparator + 1).trim();
const nextObject: Record<string, unknown> = {
[key]: parseYamlScalar(rawValue),
};
if (index < lines.length && lines[index]!.indent > indentLevel) {
const nested = parseYamlBlock(lines, index, indentLevel + 2);
if (isPlainRecord(nested.value)) {
Object.assign(nextObject, nested.value);
}
index = nested.nextIndex;
}
values.push(nextObject);
continue;
}
values.push(parseYamlScalar(remainder));
}
return { value: values, nextIndex: index };
}
// Object parsing
const record: Record<string, unknown> = {};
while (index < lines.length) {
const line = lines[index]!;
if (line.indent < indentLevel) break;
const colonIndex = line.content.indexOf(":");
if (colonIndex < 0) {
index += 1;
continue;
}
const key = line.content.slice(0, colonIndex).trim();
const remainder = line.content.slice(colonIndex + 1).trim();
index += 1;
if (!remainder && index < lines.length && lines[index]!.indent > indentLevel) {
const nested = parseYamlBlock(lines, index, lines[index]!.indent);
record[key] = nested.value;
index = nested.nextIndex;
continue;
}
record[key] = parseYamlScalar(remainder);
}
return { value: record, nextIndex: index };
}
function parseYamlFrontmatter(raw: string): Record<string, unknown> {
const prepared = prepareYamlLines(raw);
if (prepared.length === 0) return {};
const parsed = parseYamlBlock(prepared, 0, prepared[0]!.indent);
return isPlainRecord(parsed.value) ? parsed.value : {};
}
// ---------------------------------------------------------------------------
// Public API
// ---------------------------------------------------------------------------
/**
* Parse a SKILL.md file with optional YAML frontmatter.
* Normalizes CRLF to LF before parsing.
*/
export function parseFrontmatterMarkdown(
raw: string,
): { frontmatter: Record<string, unknown>; body: string } {
const normalized = raw.replace(/\r\n/g, "\n");
if (!normalized.startsWith("---\n")) {
return { frontmatter: {}, body: normalized.trim() };
}
const closing = normalized.indexOf("\n---\n", 4);
if (closing < 0) {
// Check for closing delimiter at EOF (no trailing newline)
const eofClosing = normalized.indexOf("\n---", 4);
if (eofClosing >= 0 && eofClosing + 4 >= normalized.length) {
const frontmatterRaw = normalized.slice(4, eofClosing).trim();
return {
frontmatter: parseYamlFrontmatter(frontmatterRaw),
body: "",
};
}
return { frontmatter: {}, body: normalized.trim() };
}
const frontmatterRaw = normalized.slice(4, closing).trim();
const body = normalized.slice(closing + 5).trim();
return {
frontmatter: parseYamlFrontmatter(frontmatterRaw),
body,
};
}
/**
* Classify a file path within a skill package into its inventory kind.
* Uses directory prefixes and file extensions for classification.
*/
export function classifyInventoryKind(
relativePath: string,
): FileInventoryEntry["kind"] {
// Normalize path separators and lowercase for matching
const normalized = relativePath.replace(/\\/g, "/").toLowerCase();
if (normalized.endsWith("/skill.md") || normalized === "skill.md")
return "skill";
if (normalized.startsWith("references/")) return "reference";
if (normalized.startsWith("scripts/")) return "script";
if (normalized.startsWith("assets/")) return "asset";
if (normalized.endsWith(".md")) return "markdown";
const ext = extname(normalized).toLowerCase();
const scriptExtensions = new Set([
".sh",
".js",
".mjs",
".cjs",
".ts",
".py",
".rb",
".bash",
]);
if (scriptExtensions.has(ext)) return "script";
const assetExtensions = new Set([
".png",
".jpg",
".jpeg",
".gif",
".svg",
".webp",
".pdf",
]);
if (assetExtensions.has(ext)) return "asset";
return "other";
}
/**
* Derive the trust level for a skill based on its file inventory.
* Scripts/executables -> highest risk; assets -> medium; markdown only -> lowest.
*/
export function deriveTrustLevel(inventory: FileInventoryEntry[]): TrustLevel {
if (inventory.some((entry) => entry.kind === "script"))
return "scripts_executables";
if (
inventory.some(
(entry) => entry.kind === "asset" || entry.kind === "other",
)
)
return "assets";
return "markdown_only";
}
/**
* Estimate the token count for a text string.
* Uses a rough 1:4 character-to-token ratio.
*/
export function estimateTokens(text: string): number {
if (text.length === 0) return 0;
return Math.ceil(text.length / 4);
}
+162
View File
@@ -0,0 +1,162 @@
// ---------------------------------------------------------------------------
// GitHub REST API skill fetcher -- imports skills without git clone (D-02)
// Public repos only (D-03). Uses raw.githubusercontent.com for content
// downloads (does not count against API rate limits per research).
// ---------------------------------------------------------------------------
/**
* Data returned for each skill found in a GitHub repository.
*/
export interface GitHubSkillData {
skillMdPath: string;
markdown: string;
inventory: string[];
owner: string;
repo: string;
}
interface Logger {
info: (...args: unknown[]) => void;
warn: (...args: unknown[]) => void;
error: (...args: unknown[]) => void;
}
// ---------------------------------------------------------------------------
// Commit SHA resolution
// ---------------------------------------------------------------------------
/**
* Resolve the latest commit SHA for a given ref (branch, tag, or HEAD).
* Returns the 40-character hex SHA on success, null on error.
*/
export async function resolveGitHubCommitSha(
owner: string,
repo: string,
ref?: string,
logger?: Logger,
): Promise<string | null> {
const target = ref || "HEAD";
const url = `https://api.github.com/repos/${owner}/${repo}/commits/${encodeURIComponent(target)}`;
try {
const response = await fetch(url, {
headers: { Accept: "application/vnd.github.sha" },
});
if (!response.ok) {
if (response.status === 403) {
const resetHeader = response.headers.get("X-RateLimit-Reset");
logger?.warn(
`GitHub API rate limit hit for ${owner}/${repo}`,
resetHeader ? `resets at ${new Date(Number(resetHeader) * 1000).toISOString()}` : "",
);
} else {
logger?.warn(`GitHub commit resolve failed: ${response.status} for ${owner}/${repo}@${target}`);
}
return null;
}
const sha = (await response.text()).trim();
return /^[0-9a-f]{40}$/i.test(sha) ? sha : null;
} catch (err) {
logger?.error("GitHub commit resolve network error:", err);
return null;
}
}
// ---------------------------------------------------------------------------
// GitHub tree + SKILL.md fetcher
// ---------------------------------------------------------------------------
/**
* Fetch all SKILL.md files from a GitHub repository via the REST API.
*
* 1. GET the recursive git tree to find all SKILL.md paths
* 2. Optionally filter to a specific skill slug directory
* 3. Download each SKILL.md via raw.githubusercontent.com
* 4. Collect sibling files as file inventory
*/
export async function fetchGitHubSkills(
owner: string,
repo: string,
ref: string,
requestedSlug: string | null,
logger?: Logger,
): Promise<{ skills: GitHubSkillData[]; commitSha: string }> {
const treeUrl = `https://api.github.com/repos/${owner}/${repo}/git/trees/${encodeURIComponent(ref)}?recursive=1`;
try {
const treeResponse = await fetch(treeUrl, {
headers: { Accept: "application/json" },
});
if (!treeResponse.ok) {
logger?.warn(`GitHub tree fetch failed: ${treeResponse.status} for ${owner}/${repo}@${ref}`);
return { skills: [], commitSha: ref };
}
const treeData = (await treeResponse.json()) as {
tree?: Array<{ path: string; type: string }>;
sha?: string;
};
const treeSha = typeof treeData.sha === "string" ? treeData.sha : ref;
const allEntries = (treeData.tree ?? []).filter(
(entry) => entry.type === "blob" && typeof entry.path === "string",
);
const allPaths = allEntries.map((entry) => entry.path);
// Find SKILL.md files (at root or nested)
let skillMdPaths = allPaths.filter(
(p) => p.endsWith("/SKILL.md") || p === "SKILL.md",
);
// Filter to requested slug directory if provided
if (requestedSlug) {
skillMdPaths = skillMdPaths.filter((p) => {
const dir = p === "SKILL.md" ? "" : p.slice(0, p.lastIndexOf("/"));
const dirName = dir.split("/").pop() ?? "";
return dirName.toLowerCase() === requestedSlug.toLowerCase();
});
}
if (skillMdPaths.length === 0) {
logger?.info(`No SKILL.md files found in ${owner}/${repo}@${ref}`);
return { skills: [], commitSha: treeSha };
}
const skills: GitHubSkillData[] = [];
for (const skillMdPath of skillMdPaths) {
try {
// Download content via raw.githubusercontent.com (no API rate limit)
const rawUrl = `https://raw.githubusercontent.com/${owner}/${repo}/${ref}/${skillMdPath}`;
const contentResponse = await fetch(rawUrl);
if (!contentResponse.ok) {
logger?.warn(`Failed to download ${rawUrl}: ${contentResponse.status}`);
continue;
}
const markdown = await contentResponse.text();
// Collect sibling files in the same directory (or subdirectories)
const skillDir =
skillMdPath === "SKILL.md"
? ""
: skillMdPath.slice(0, skillMdPath.lastIndexOf("/"));
const prefix = skillDir ? `${skillDir}/` : "";
const inventory = allPaths.filter(
(p) => p !== skillMdPath && (prefix ? p.startsWith(prefix) : !p.includes("/")),
);
skills.push({
skillMdPath,
markdown,
inventory,
owner,
repo,
});
} catch (err) {
logger?.error(`Error downloading skill ${skillMdPath}:`, err);
continue;
}
}
return { skills, commitSha: treeSha };
} catch (err) {
logger?.error(`GitHub tree fetch network error for ${owner}/${repo}:`, err);
return { skills: [], commitSha: ref };
}
}
+194
View File
@@ -0,0 +1,194 @@
// ---------------------------------------------------------------------------
// Canonical skill key derivation and skill import source resolution
// ---------------------------------------------------------------------------
import { createHash } from "node:crypto";
import { resolve } from "node:path";
import type { SkillSourceType } from "./skill-types.js";
// ---------------------------------------------------------------------------
// Slug normalization helpers
// ---------------------------------------------------------------------------
const SLUG_DELIM_RE = /[^a-z0-9]+/g;
const SLUG_TRIM_RE = /^-+|-+$/g;
/**
* Normalize a value into a URL-safe slug.
* Returns null if the value is falsy or results in an empty string.
*/
export function normalizeSkillSlug(
value: string | undefined | null,
): string | null {
if (!value) return null;
const trimmed = value.trim();
if (!trimmed) return null;
const normalized = trimmed
.toLowerCase()
.replace(SLUG_DELIM_RE, "-")
.replace(SLUG_TRIM_RE, "");
return normalized.length > 0 ? normalized : null;
}
// ---------------------------------------------------------------------------
// Hashing
// ---------------------------------------------------------------------------
/**
* Compute the first 10 characters of a SHA-256 hex digest.
* Used for hash segments in canonical keys.
*/
export function hashSkillValue(value: string): string {
return createHash("sha256").update(value).digest("hex").slice(0, 10);
}
// ---------------------------------------------------------------------------
// Canonical key derivation
// ---------------------------------------------------------------------------
function asString(value: unknown): string | null {
return typeof value === "string" && value.length > 0 ? value : null;
}
function isPlainRecord(value: unknown): value is Record<string, unknown> {
return typeof value === "object" && value !== null && !Array.isArray(value);
}
/**
* Derive a deterministic canonical key for a skill.
*
* Key formats by source type:
* - GitHub/skills_sh with owner+repo: `{owner}/{repo}/{slug}`
* - URL with sourceLocator: `url/{normalizedHost}/{hash10}/{slug}`
* - local_path with sourceLocator: `local/{hash10 of resolved path}/{slug}`
* - Fallback (managed or unknown): `company/{companyId}/{slug}`
*/
export function deriveCanonicalSkillKey(
companyId: string,
input: {
slug: string;
sourceType: SkillSourceType;
sourceLocator: string | null;
metadata: Record<string, unknown> | null;
},
): string {
const slug = normalizeSkillSlug(input.slug) ?? "skill";
const metadata = isPlainRecord(input.metadata) ? input.metadata : null;
const owner = normalizeSkillSlug(asString(metadata?.owner));
const repo = normalizeSkillSlug(asString(metadata?.repo));
// GitHub or skills.sh with owner/repo metadata
if (
(input.sourceType === "github" || input.sourceType === "skills_sh") &&
owner &&
repo
) {
return `${owner}/${repo}/${slug}`;
}
// URL source with locator
if (input.sourceType === "url") {
const locator = asString(input.sourceLocator);
if (locator) {
try {
const url = new URL(locator);
const host = normalizeSkillSlug(url.host) ?? "url";
return `url/${host}/${hashSkillValue(locator)}/${slug}`;
} catch {
return `url/unknown/${hashSkillValue(locator)}/${slug}`;
}
}
}
// Local path source with locator
if (input.sourceType === "local_path") {
const locator = asString(input.sourceLocator);
if (locator) {
return `local/${hashSkillValue(resolve(locator))}/${slug}`;
}
}
// Fallback: company-scoped key
return `company/${companyId}/${slug}`;
}
// ---------------------------------------------------------------------------
// Skill import source resolution
// ---------------------------------------------------------------------------
// Three-part key: org/repo/skill
const THREE_PART_KEY_RE =
/^[A-Za-z0-9_.-]+\/[A-Za-z0-9_.-]+\/[A-Za-z0-9_.-]+$/;
// Two-part key: org/repo
const TWO_PART_KEY_RE = /^[A-Za-z0-9_.-]+\/[A-Za-z0-9_.-]+$/;
// skills.sh URL pattern
const SKILLS_SH_RE =
/^https?:\/\/(?:www\.)?skills\.sh\/([A-Za-z0-9_.-]+)\/([A-Za-z0-9_.-]+)(?:\/([A-Za-z0-9_.-]+))?(?:[?#].*)?$/i;
export interface ParsedSkillImportSource {
resolvedSource: string;
requestedSkillSlug: string | null;
originalSkillsShUrl: string | null;
}
/**
* Parse a raw skill import source string and resolve it to a normalized form.
*
* Supported input formats:
* - `owner/repo/skill` -> GitHub URL with skills.sh origin tracking
* - `owner/repo` -> GitHub URL
* - `https://skills.sh/owner/repo/skill` -> GitHub URL with skills.sh origin
* - `https://github.com/owner/repo` -> pass through
* - `/local/path` -> pass through
*/
export function parseSkillImportSource(
rawInput: string,
): ParsedSkillImportSource {
const trimmed = rawInput.trim();
// Three-part key (org/repo/skill) -- originates from skills.sh registry
if (
!/^https?:\/\//i.test(trimmed) &&
THREE_PART_KEY_RE.test(trimmed)
) {
const [owner, repo, skillSlugRaw] = trimmed.split("/");
return {
resolvedSource: `https://github.com/${owner}/${repo}`,
requestedSkillSlug: normalizeSkillSlug(skillSlugRaw),
originalSkillsShUrl: `https://skills.sh/${owner}/${repo}/${skillSlugRaw}`,
};
}
// Two-part key (org/repo) -- resolve to GitHub
if (
!/^https?:\/\//i.test(trimmed) &&
TWO_PART_KEY_RE.test(trimmed)
) {
return {
resolvedSource: `https://github.com/${trimmed}`,
requestedSkillSlug: null,
originalSkillsShUrl: null,
};
}
// skills.sh URL -> resolve to GitHub
const skillsShMatch = trimmed.match(SKILLS_SH_RE);
if (skillsShMatch) {
const [, owner, repo, skillSlugRaw] = skillsShMatch;
return {
resolvedSource: `https://github.com/${owner}/${repo}`,
requestedSkillSlug: skillSlugRaw
? normalizeSkillSlug(skillSlugRaw)
: null,
originalSkillsShUrl: trimmed,
};
}
// Everything else -- pass through as resolvedSource
return {
resolvedSource: trimmed,
requestedSkillSlug: null,
originalSkillsShUrl: null,
};
}
+719
View File
@@ -0,0 +1,719 @@
// ---------------------------------------------------------------------------
// SkillService -- full skill lifecycle: import, CRUD, assignment, token budget,
// update detection. Factory pattern following createOrchestrationService.
// ---------------------------------------------------------------------------
import { randomUUID } from "node:crypto";
import { existsSync, mkdirSync, readdirSync, readFileSync, statSync } from "node:fs";
import { mkdir, writeFile, rm } from "node:fs/promises";
import { join, resolve, basename, dirname, relative, posix } from "node:path";
import type { AppDatabase } from "./database";
import type { DesktopResult } from "@shared/contracts";
import type {
FileInventoryEntry,
SkillRecord,
SkillImportResult,
SkillSourceType,
SkillUpdateStatus,
RuntimeSkillEntry,
} from "./skill-types.js";
import {
parseFrontmatterMarkdown,
classifyInventoryKind,
deriveTrustLevel,
estimateTokens,
} from "./skill-format.js";
import {
deriveCanonicalSkillKey,
normalizeSkillSlug,
hashSkillValue,
parseSkillImportSource,
} from "./skill-keys.js";
import {
resolveGitHubCommitSha,
fetchGitHubSkills,
} from "./skill-github.js";
import { TOKEN_BUDGETS } from "./skill-types.js";
// ---------------------------------------------------------------------------
// Types
// ---------------------------------------------------------------------------
interface SkillServiceDependencies {
db: AppDatabase;
profileDir: string;
logger: {
info: (...args: unknown[]) => void;
warn: (...args: unknown[]) => void;
error: (...args: unknown[]) => void;
};
}
// ---------------------------------------------------------------------------
// Factory
// ---------------------------------------------------------------------------
export function createSkillService({ db, profileDir, logger }: SkillServiceDependencies) {
// ---- Result helpers ----
function ok<T>(data: T): DesktopResult<T> {
return { ok: true, data };
}
function fail(code: string, message: string): DesktopResult<never> {
return { ok: false, error: { code, message } };
}
// ---- Path helpers (D-10: skills stored in profileDir/skills/{companyId}/) ----
function skillsRoot(companyId: string): string {
return join(profileDir, "skills", companyId);
}
function skillFilesDir(companyId: string, slug: string): string {
return join(skillsRoot(companyId), slug);
}
function runtimeDir(companyId: string): string {
return join(skillsRoot(companyId), "__runtime__");
}
// ---- File materialization helpers ----
/**
* Write SKILL.md (and tracked inventory files from GitHub downloads)
* to the local skill directory.
*/
async function materializeSkillFiles(
companyId: string,
slug: string,
markdown: string,
_inventory: FileInventoryEntry[],
): Promise<string> {
const dir = skillFilesDir(companyId, slug);
await mkdir(dir, { recursive: true });
await writeFile(join(dir, "SKILL.md"), markdown, "utf-8");
return dir;
}
/** Remove a skill's file directory from disk. */
async function cleanupSkillFiles(companyId: string, slug: string): Promise<void> {
const dir = skillFilesDir(companyId, slug);
if (existsSync(dir)) {
await rm(dir, { recursive: true, force: true });
}
}
// ---- GitHub URL parser ----
function parseGitHubUrl(url: string): { owner: string; repo: string } | null {
try {
const parsed = new URL(url);
if (parsed.hostname !== "github.com") return null;
const parts = parsed.pathname.split("/").filter(Boolean);
if (parts.length < 2) return null;
return {
owner: parts[0]!,
repo: parts[1]!.replace(/\.git$/i, ""),
};
} catch {
return null;
}
}
// ---- Import routing ----
/**
* Import skills from any supported source.
* Routes to the correct handler based on parseSkillImportSource.
*/
async function importFromSource(
companyId: string,
rawSource: string,
): Promise<DesktopResult<SkillImportResult[]>> {
try {
const parsed = parseSkillImportSource(rawSource);
const source = parsed.resolvedSource;
// skills.sh resolved sources are always GitHub URLs
if (parsed.originalSkillsShUrl) {
const results = await importFromGitHub(
companyId,
source,
parsed.requestedSkillSlug,
parsed.originalSkillsShUrl,
);
return ok(results);
}
// GitHub URL
if (source.includes("github.com")) {
const results = await importFromGitHub(
companyId,
source,
parsed.requestedSkillSlug,
null,
);
return ok(results);
}
// Local path (absolute path, drive letter on Windows, or ~ home dir)
if (
source.startsWith("/") ||
/^[A-Za-z]:/.test(source) ||
source.startsWith("~")
) {
const results = await importFromLocal(companyId, source);
return ok(results);
}
// HTTP/HTTPS URL
if (/^https?:\/\//i.test(source)) {
const results = await importFromUrl(companyId, source);
return ok(results);
}
return fail("invalid_source", `Unrecognized skill source format: ${rawSource}`);
} catch (err) {
const message = err instanceof Error ? err.message : String(err);
logger.error("Skill import failed:", message);
return fail("import_error", message);
}
}
// ---- GitHub import (SKILL-01) ----
async function importFromGitHub(
companyId: string,
url: string,
requestedSlug: string | null,
originalSkillsShUrl: string | null,
): Promise<SkillImportResult[]> {
const ghParsed = parseGitHubUrl(url);
if (!ghParsed) {
throw new Error(`Invalid GitHub URL: ${url}`);
}
const { owner, repo } = ghParsed;
// Resolve latest commit SHA
const sha = await resolveGitHubCommitSha(owner, repo, undefined, logger);
if (!sha) {
throw new Error(`Failed to resolve commit SHA for ${owner}/${repo}`);
}
// Fetch SKILL.md files from repo
const { skills: githubSkills, commitSha } = await fetchGitHubSkills(
owner, repo, sha, requestedSlug, logger,
);
if (githubSkills.length === 0) {
throw new Error(`No SKILL.md files found in ${owner}/${repo}`);
}
const results: SkillImportResult[] = [];
const sourceType: SkillSourceType = originalSkillsShUrl ? "skills_sh" : "github";
const nowTs = new Date().toISOString();
for (const ghSkill of githubSkills) {
const parsed = parseFrontmatterMarkdown(ghSkill.markdown);
const rawSlug =
normalizeSkillSlug(parsed.frontmatter.slug as string | undefined) ??
normalizeSkillSlug(
ghSkill.skillMdPath === "SKILL.md"
? repo
: basename(dirname(ghSkill.skillMdPath)),
) ??
repo;
// Build file inventory from the tree paths
const skillDir =
ghSkill.skillMdPath === "SKILL.md"
? ""
: ghSkill.skillMdPath.slice(0, ghSkill.skillMdPath.lastIndexOf("/"));
const prefix = skillDir ? `${skillDir}/` : "";
const inventory: FileInventoryEntry[] = [
{ path: "SKILL.md", kind: "skill" as const },
...ghSkill.inventory.map((filePath) => {
const relativePath = prefix ? filePath.slice(prefix.length) : filePath;
return {
path: relativePath,
kind: classifyInventoryKind(relativePath),
};
}),
];
const metadata: Record<string, unknown> = {
sourceKind: "github",
owner,
repo,
ref: commitSha,
trackingRef: "HEAD",
...(originalSkillsShUrl ? { originalSkillsShUrl } : {}),
};
const key = deriveCanonicalSkillKey(companyId, {
slug: rawSlug,
sourceType,
sourceLocator: url,
metadata,
});
const trustLevel = deriveTrustLevel(inventory);
const name =
(typeof parsed.frontmatter.name === "string" ? parsed.frontmatter.name : null) ??
rawSlug;
const description =
typeof parsed.frontmatter.description === "string"
? parsed.frontmatter.description
: null;
const skillId = randomUUID();
// Materialize files to disk
await materializeSkillFiles(companyId, rawSlug, ghSkill.markdown, inventory);
// Upsert to database
const { isUpdate } = db.upsertCompanySkill({
id: skillId,
companyId,
key,
slug: rawSlug,
name,
description,
markdown: ghSkill.markdown,
sourceType,
sourceLocator: url,
sourceRef: commitSha,
trustLevel,
compatibility: "compatible",
fileInventory: JSON.stringify(inventory),
metadata: JSON.stringify(metadata),
createdAt: nowTs,
updatedAt: nowTs,
});
results.push({ skillId, key, name, isUpdate });
logger.info(`Skill ${isUpdate ? "updated" : "imported"}: ${key} from ${owner}/${repo}`);
}
return results;
}
// ---- Local path import (SKILL-02, D-04) ----
async function importFromLocal(
companyId: string,
localPath: string,
): Promise<SkillImportResult[]> {
const resolvedPath = resolve(localPath.replace(/^~/, process.env.HOME ?? process.env.USERPROFILE ?? ""));
if (!existsSync(resolvedPath)) {
throw new Error(`Local path does not exist: ${resolvedPath}`);
}
const stat = statSync(resolvedPath);
const results: SkillImportResult[] = [];
const nowTs = new Date().toISOString();
if (stat.isFile()) {
// Single file treated as a skill
const markdown = readFileSync(resolvedPath, "utf-8");
const parsed = parseFrontmatterMarkdown(markdown);
const rawSlug =
normalizeSkillSlug(parsed.frontmatter.slug as string | undefined) ??
normalizeSkillSlug(basename(resolvedPath, ".md")) ??
"skill";
const inventory: FileInventoryEntry[] = [
{ path: "SKILL.md", kind: "skill" },
];
const trustLevel = deriveTrustLevel(inventory);
const contentHash = hashSkillValue(markdown);
const key = deriveCanonicalSkillKey(companyId, {
slug: rawSlug,
sourceType: "local_path",
sourceLocator: resolvedPath,
metadata: null,
});
const skillId = randomUUID();
await materializeSkillFiles(companyId, rawSlug, markdown, inventory);
const { isUpdate } = db.upsertCompanySkill({
id: skillId,
companyId,
key,
slug: rawSlug,
name: (typeof parsed.frontmatter.name === "string" ? parsed.frontmatter.name : null) ?? rawSlug,
description: typeof parsed.frontmatter.description === "string" ? parsed.frontmatter.description : null,
markdown,
sourceType: "local_path",
sourceLocator: resolvedPath,
sourceRef: contentHash,
trustLevel,
compatibility: "compatible",
fileInventory: JSON.stringify(inventory),
metadata: null,
createdAt: nowTs,
updatedAt: nowTs,
});
results.push({ skillId, key, name: rawSlug, isUpdate });
logger.info(`Skill ${isUpdate ? "updated" : "imported"} from local file: ${key}`);
return results;
}
// Directory: walk to find SKILL.md files
const skillMdFiles = findSkillMdFiles(resolvedPath);
if (skillMdFiles.length === 0) {
throw new Error(`No SKILL.md files found in ${resolvedPath}`);
}
for (const skillMdPath of skillMdFiles) {
const skillDir = dirname(skillMdPath);
const markdown = readFileSync(skillMdPath, "utf-8");
const parsed = parseFrontmatterMarkdown(markdown);
const rawSlug =
normalizeSkillSlug(parsed.frontmatter.slug as string | undefined) ??
normalizeSkillSlug(basename(skillDir)) ??
"skill";
// Build inventory from directory listing
const inventory = buildLocalInventory(skillDir, skillMdPath);
const trustLevel = deriveTrustLevel(inventory);
const contentHash = hashSkillValue(markdown);
const key = deriveCanonicalSkillKey(companyId, {
slug: rawSlug,
sourceType: "local_path",
sourceLocator: skillDir,
metadata: null,
});
const skillId = randomUUID();
await materializeSkillFiles(companyId, rawSlug, markdown, inventory);
const { isUpdate } = db.upsertCompanySkill({
id: skillId,
companyId,
key,
slug: rawSlug,
name: (typeof parsed.frontmatter.name === "string" ? parsed.frontmatter.name : null) ?? rawSlug,
description: typeof parsed.frontmatter.description === "string" ? parsed.frontmatter.description : null,
markdown,
sourceType: "local_path",
sourceLocator: skillDir,
sourceRef: contentHash,
trustLevel,
compatibility: "compatible",
fileInventory: JSON.stringify(inventory),
metadata: null,
createdAt: nowTs,
updatedAt: nowTs,
});
results.push({ skillId, key, name: rawSlug, isUpdate });
logger.info(`Skill ${isUpdate ? "updated" : "imported"} from local path: ${key}`);
}
return results;
}
// ---- URL import (SKILL-03, D-05) ----
async function importFromUrl(
companyId: string,
url: string,
): Promise<SkillImportResult[]> {
const response = await fetch(url);
if (!response.ok) {
throw new Error(`Failed to fetch skill from URL: ${response.status} ${url}`);
}
const markdown = await response.text();
const parsed = parseFrontmatterMarkdown(markdown);
const nowTs = new Date().toISOString();
// Derive slug from frontmatter or URL path
let urlSlug: string | null = null;
try {
const urlObj = new URL(url);
const pathParts = urlObj.pathname.split("/").filter(Boolean);
const lastPart = pathParts[pathParts.length - 1] ?? "";
urlSlug = normalizeSkillSlug(lastPart.replace(/\.md$/i, ""));
} catch {
// ignore URL parsing errors
}
const rawSlug =
normalizeSkillSlug(parsed.frontmatter.slug as string | undefined) ??
urlSlug ??
"skill";
// Single-file skill: only SKILL.md
const inventory: FileInventoryEntry[] = [
{ path: "SKILL.md", kind: "skill" },
];
const trustLevel: "markdown_only" = "markdown_only";
const key = deriveCanonicalSkillKey(companyId, {
slug: rawSlug,
sourceType: "url",
sourceLocator: url,
metadata: null,
});
const skillId = randomUUID();
await materializeSkillFiles(companyId, rawSlug, markdown, inventory);
const { isUpdate } = db.upsertCompanySkill({
id: skillId,
companyId,
key,
slug: rawSlug,
name: (typeof parsed.frontmatter.name === "string" ? parsed.frontmatter.name : null) ?? rawSlug,
description: typeof parsed.frontmatter.description === "string" ? parsed.frontmatter.description : null,
markdown,
sourceType: "url",
sourceLocator: url,
sourceRef: null,
trustLevel,
compatibility: "compatible",
fileInventory: JSON.stringify(inventory),
metadata: null,
createdAt: nowTs,
updatedAt: nowTs,
});
const results: SkillImportResult[] = [
{ skillId, key, name: rawSlug, isUpdate },
];
logger.info(`Skill ${isUpdate ? "updated" : "imported"} from URL: ${key}`);
return results;
}
// ---- CRUD ----
function listForCompany(companyId: string): SkillRecord[] {
return db.listCompanySkills(companyId);
}
function getDetail(companyId: string, skillId: string): DesktopResult<SkillRecord> {
const skill = db.getCompanySkill(companyId, skillId);
if (!skill) return fail("not_found", `Skill ${skillId} not found`);
return ok(skill);
}
/** Delete a skill: (1) assignments cascade via FK, (2) files from disk, (3) DB record per D-12. */
async function deleteSkill(
companyId: string,
skillId: string,
): Promise<DesktopResult<void>> {
const skill = db.getCompanySkill(companyId, skillId);
if (!skill) return fail("not_found", `Skill ${skillId} not found`);
try {
// D-12 order: (1) assignments cascade via FK on delete,
// (2) delete files from disk
await cleanupSkillFiles(companyId, skill.slug);
// (3) delete DB record (cascades agent_skill_assignments via FK)
db.deleteCompanySkill(companyId, skillId);
logger.info(`Skill deleted: ${skill.key}`);
return ok(undefined);
} catch (err) {
const message = err instanceof Error ? err.message : String(err);
logger.error("Skill deletion failed:", message);
return fail("delete_error", message);
}
}
// ---- Agent assignment (SKILL-07) ----
function assignToAgent(
companyId: string,
agentId: string,
skillId: string,
): DesktopResult<void> {
const skill = db.getCompanySkill(companyId, skillId);
if (!skill) return fail("not_found", `Skill ${skillId} not found for company`);
db.assignSkillToAgent(agentId, skillId);
return ok(undefined);
}
function removeFromAgent(
_companyId: string,
agentId: string,
skillId: string,
): DesktopResult<void> {
db.removeSkillFromAgent(agentId, skillId);
return ok(undefined);
}
// ---- Update detection (SKILL-11, D-11) ----
async function checkUpdateStatus(
companyId: string,
skillId: string,
): Promise<DesktopResult<SkillUpdateStatus>> {
const skill = db.getCompanySkill(companyId, skillId);
if (!skill) return fail("not_found", `Skill ${skillId} not found`);
// Only GitHub/skills_sh sources support update checks
if (skill.sourceType !== "github" && skill.sourceType !== "skills_sh") {
return ok({
skillId: skill.id,
key: skill.key,
currentRef: skill.sourceRef,
latestRef: null,
hasUpdate: false,
});
}
// Parse owner/repo from metadata
const metadata = skill.metadata;
const owner = typeof metadata?.owner === "string" ? metadata.owner : null;
const repo = typeof metadata?.repo === "string" ? metadata.repo : null;
if (!owner || !repo) {
// Try to parse from sourceLocator URL
const ghParsed = skill.sourceLocator ? parseGitHubUrl(skill.sourceLocator) : null;
if (!ghParsed) {
return ok({
skillId: skill.id,
key: skill.key,
currentRef: skill.sourceRef,
latestRef: null,
hasUpdate: false,
});
}
const latestRef = await resolveGitHubCommitSha(ghParsed.owner, ghParsed.repo, undefined, logger);
return ok({
skillId: skill.id,
key: skill.key,
currentRef: skill.sourceRef,
latestRef,
hasUpdate: latestRef != null && latestRef !== skill.sourceRef,
});
}
const trackingRef = typeof metadata?.trackingRef === "string" ? metadata.trackingRef : undefined;
const latestRef = await resolveGitHubCommitSha(owner, repo, trackingRef, logger);
return ok({
skillId: skill.id,
key: skill.key,
currentRef: skill.sourceRef,
latestRef,
hasUpdate: latestRef != null && latestRef !== skill.sourceRef,
});
}
// ---- Runtime skill entries (SKILL-09 token budget) ----
/**
* Build the list of RuntimeSkillEntry for adapter injection.
* Applies token budget enforcement: skills sorted by creation date,
* accumulated until budget exceeded, then skipped with warning.
*/
function listRuntimeSkillEntries(
companyId: string,
agentId: string,
adapterType: string,
): RuntimeSkillEntry[] {
const allSkills = db.listCompanySkills(companyId);
const agentAssignments = db.listAgentSkillAssignments(agentId);
// If agent has specific assignments, filter to those; otherwise use all
const skills =
agentAssignments.length > 0
? allSkills.filter((s) => agentAssignments.includes(s.id))
: allSkills;
// Sort by creation date (oldest first -- first imported gets priority)
const sorted = [...skills].sort(
(a, b) => a.createdAt.localeCompare(b.createdAt),
);
const budget = TOKEN_BUDGETS[adapterType] ?? 30000;
let accumulated = 0;
const entries: RuntimeSkillEntry[] = [];
for (const skill of sorted) {
const tokenEstimate = estimateTokens(skill.markdown);
if (accumulated + tokenEstimate > budget) {
logger.warn(
`Token budget exceeded for ${adapterType}: skipping skill "${skill.name}" (${tokenEstimate} tokens, budget ${budget}, used ${accumulated})`,
);
continue;
}
accumulated += tokenEstimate;
entries.push({
runtimeName: skill.slug,
sourcePath: skillFilesDir(companyId, skill.slug),
name: skill.name,
tokenEstimate,
});
}
return entries;
}
// ---- Local directory walker ----
function findSkillMdFiles(rootDir: string): string[] {
const found: string[] = [];
function walk(dir: string) {
try {
const entries = readdirSync(dir, { withFileTypes: true });
for (const entry of entries) {
const fullPath = join(dir, entry.name);
if (entry.isDirectory()) {
walk(fullPath);
} else if (entry.name === "SKILL.md") {
found.push(fullPath);
}
}
} catch {
// Skip unreadable directories
}
}
walk(rootDir);
return found;
}
function buildLocalInventory(skillDir: string, skillMdPath: string): FileInventoryEntry[] {
const inventory: FileInventoryEntry[] = [
{ path: "SKILL.md", kind: "skill" },
];
try {
const entries = readdirSync(skillDir, { withFileTypes: true, recursive: true });
for (const entry of entries) {
if (!entry.isFile()) continue;
const entryPath = entry.parentPath
? join(entry.parentPath, entry.name)
: join(skillDir, entry.name);
const fullPath = resolve(entryPath);
if (fullPath === skillMdPath) continue;
const relativePath = relative(skillDir, fullPath).replace(/\\/g, "/");
inventory.push({
path: relativePath,
kind: classifyInventoryKind(relativePath),
});
}
} catch {
// If directory listing fails, return just the SKILL.md entry
}
return inventory;
}
// ---- Public API ----
return {
importFromSource,
importFromGitHub,
importFromLocal,
importFromUrl,
listForCompany,
getDetail,
deleteSkill,
assignToAgent,
removeFromAgent,
checkUpdateStatus,
listRuntimeSkillEntries,
};
}
+114
View File
@@ -0,0 +1,114 @@
// ---------------------------------------------------------------------------
// Skill system type definitions and constants
// ---------------------------------------------------------------------------
/**
* Source type for a skill -- matches the company_skills.source_type column.
*/
export type SkillSourceType =
| "github"
| "local_path"
| "url"
| "skills_sh"
| "managed";
/**
* Trust level derived from the file inventory of a skill package.
* Determines what content is injected and how cautiously.
*/
export type TrustLevel =
| "markdown_only"
| "assets"
| "scripts_executables";
/**
* A single file entry within a skill package inventory.
*/
export interface FileInventoryEntry {
path: string;
kind: "skill" | "reference" | "script" | "asset" | "markdown" | "other";
}
/**
* Full row interface matching the company_skills table.
* Property names use camelCase (JS convention) even though DB columns are snake_case.
*/
export interface SkillRecord {
id: string;
companyId: string;
key: string;
slug: string;
name: string;
description: string | null;
markdown: string;
sourceType: SkillSourceType;
sourceLocator: string | null;
sourceRef: string | null;
trustLevel: TrustLevel;
compatibility: string;
fileInventory: FileInventoryEntry[];
metadata: Record<string, unknown> | null;
createdAt: string;
updatedAt: string;
}
/**
* Result returned after importing or updating a skill.
*/
export interface SkillImportResult {
skillId: string;
key: string;
name: string;
isUpdate: boolean;
}
/**
* Runtime representation of a skill entry for adapter injection.
*/
export interface RuntimeSkillEntry {
runtimeName: string;
sourcePath: string;
name: string;
tokenEstimate: number;
}
/**
* Minimal skill reference passed to worker threads.
* Subset of RuntimeSkillEntry to minimize serialization overhead.
*/
export interface ManagedSkillRef {
runtimeName: string;
sourcePath: string;
}
/**
* Skill-to-agent assignment record.
*/
export interface SkillAssignment {
agentId: string;
skillId: string;
}
/**
* Update status for a skill -- used to check if a newer version is available.
*/
export interface SkillUpdateStatus {
skillId: string;
key: string;
currentRef: string | null;
latestRef: string | null;
hasUpdate: boolean;
}
/**
* Per-adapter token budgets for skill content injection.
* Values represent the maximum token count of skill content that can be
* included in a single prompt for each adapter type.
*/
export const TOKEN_BUDGETS: Readonly<Record<string, number>> = {
claude_local: 50000,
codex_local: 30000,
gemini_local: 50000,
opencode_local: 30000,
kimi_local: 20000,
};
+319
View File
@@ -0,0 +1,319 @@
/**
* Task Quality Gates — pure validation functions for task creation.
*
* Called by IPC handlers, API server, and automation service to enforce
* minimum description quality, acceptance criteria, and capability matching.
*/
// ---------------------------------------------------------------------------
// Types
// ---------------------------------------------------------------------------
export interface TaskValidationResult {
valid: boolean;
errors: string[];
warnings: string[];
}
export interface CapabilityMatchResult {
matched: boolean;
warning?: string;
}
/** Minimal DB interface for capability matching — keeps the function testable with mocks.
* Skill methods are optional since the skill system may not be implemented yet (graceful degradation per D-07). */
export interface CapabilityMatchDb {
getAgent(id: string): { name: string; capabilities: string };
listAgentSkillAssignments?(id: string): string[];
getCompanySkill?(companyId: string, skillId: string): { name: string; description: string | null; slug: string } | null;
}
// ---------------------------------------------------------------------------
// Constants
// ---------------------------------------------------------------------------
const MIN_DESCRIPTION_LENGTH = 50;
export const ACCEPTANCE_CRITERIA_KEYWORDS: readonly string[] = [
"acceptance criteria",
"done when",
"success criteria",
"definition of done",
"requirements:",
"must:",
"should:",
"- [ ]",
"- [x]",
"deliverable:",
"expected outcome",
"exit criteria",
];
// ---------------------------------------------------------------------------
// checkAcceptanceCriteria
// ---------------------------------------------------------------------------
export function checkAcceptanceCriteria(description: string): { hasCriteria: boolean; suggestion: string } {
const lower = description.toLowerCase();
const found = ACCEPTANCE_CRITERIA_KEYWORDS.some((kw) => lower.includes(kw));
if (found) {
return { hasCriteria: true, suggestion: "" };
}
return {
hasCriteria: false,
suggestion:
'Task description should include acceptance criteria. Add a "Done when:" or "Acceptance criteria:" section, or use checkbox items (- [ ]) to define success conditions.',
};
}
// ---------------------------------------------------------------------------
// validateTaskCreation
// ---------------------------------------------------------------------------
export function validateTaskCreation(
input: { title: string; description: string },
isHeartbeat: boolean,
): TaskValidationResult {
// Heartbeat tasks bypass all validation
if (isHeartbeat) {
return { valid: true, errors: [], warnings: [] };
}
const errors: string[] = [];
const warnings: string[] = [];
// TASK-01: Minimum description length
if (input.description.length < MIN_DESCRIPTION_LENGTH) {
errors.push(
`Task description must be at least ${MIN_DESCRIPTION_LENGTH} characters (currently ${input.description.length}). Add details about what needs to be done, expected behavior, and any constraints.`,
);
}
// TASK-02: Acceptance criteria detection (only check if length passes)
if (input.description.length >= MIN_DESCRIPTION_LENGTH) {
const criteria = checkAcceptanceCriteria(input.description);
if (!criteria.hasCriteria) {
errors.push(criteria.suggestion);
}
}
return { valid: errors.length === 0, errors, warnings };
}
// ---------------------------------------------------------------------------
// checkCapabilityMatch
// ---------------------------------------------------------------------------
/** Tokenize text into words >3 chars, same pattern as database.ts autoAssignTask */
function tokenize(text: string): string[] {
return text.toLowerCase().split(/\W+/).filter((w) => w.length > 3);
}
export function checkCapabilityMatch(
db: CapabilityMatchDb,
agentId: string,
taskDescription: string,
companyId: string,
): CapabilityMatchResult {
const taskWords = tokenize(taskDescription);
// Nothing to match against — task description too short/vague
if (taskWords.length === 0) {
return { matched: true };
}
const agent = db.getAgent(agentId);
const capWords = tokenize(agent.capabilities);
const capabilityOverlap = taskWords.filter((w) => capWords.includes(w)).length;
// Check skill assignments for additional matching (graceful degradation)
let skillOverlap = 0;
try {
if (db.listAgentSkillAssignments && db.getCompanySkill) {
const skillIds = db.listAgentSkillAssignments(agentId);
for (const skillId of skillIds) {
const skill = db.getCompanySkill(companyId, skillId);
if (skill) {
const skillWords = tokenize(`${skill.name} ${skill.description ?? ""} ${skill.slug}`);
skillOverlap += taskWords.filter((w) => skillWords.includes(w)).length;
}
}
}
} catch {
// Skill system may not be available — graceful degradation per D-07
}
// Direct capability or skill overlap found
if (capabilityOverlap > 0 || skillOverlap > 0) {
return { matched: true };
}
// No capabilities defined at all — graceful degradation (Pitfall 4)
if (capWords.length === 0 && skillOverlap === 0) {
return { matched: true };
}
// Agent has capabilities but no overlap — warn
return {
matched: false,
warning: `Agent "${agent.name}" may lack skills for this task. Their capabilities (${agent.capabilities.slice(0, 80)}...) have no overlap with the task description. Consider reassigning.`,
};
}
// ---------------------------------------------------------------------------
// TASK-04: Stale Review Escalation
// ---------------------------------------------------------------------------
export const REVIEW_TIMEOUT_MS = 30 * 60 * 1000; // 30 minutes (D-08)
export interface SweepStaleReviewsDeps {
db: {
queryAll: <T>(sql: string, ...params: unknown[]) => T[];
getManagerForAgent: (agentId: string) => { id: string; name: string; companyId: string } | null;
getChainOfCommand: (agentId: string) => Array<{ id: string; name: string }>;
advanceTaskStatus: (taskId: string, newStatus: string, comment?: string) => { previousStatus: string };
};
dispatchAgentMessage: (
deps: {
db: SweepStaleReviewsDeps["db"];
emitEvent: (...args: unknown[]) => void;
publishDomainChanged: () => void;
wakeAgentIfPossible: (agentId: string, companyId: string, trigger: string) => void;
},
message: {
companyId: string;
fromAgentId: string;
toAgentId: string | null;
channel: string;
subject: string;
body: string;
priority: string;
},
wakeTrigger: string,
) => { ok: boolean };
resolveApprovalDecision: (params: {
db: SweepStaleReviewsDeps["db"];
handleApprovalResolved: (approvalId: string, companyId: string, decision: string) => void;
companyId: string;
approvalId: string;
state: "approved" | "rejected" | "revision_requested";
decisionNote?: string;
detail?: string | null;
actor?: string;
}) => void;
notify: (options: {
title: string;
body: string;
urgency?: string;
navigation?: { section: string; entityId?: string };
batchKey?: string;
}) => void;
handleTaskStatusChange: (taskId: string, companyId: string, previousStatus: string, newStatus: string) => void;
handleApprovalResolved: (approvalId: string, companyId: string, decision: string) => void;
emitEvent: (...args: unknown[]) => void;
publishDomainChanged: () => void;
wakeAgentIfPossible: (agentId: string, companyId: string, trigger: string) => void;
logger: { info: (...args: unknown[]) => void; warn: (...args: unknown[]) => void };
}
export function sweepStaleReviews(
companyId: string,
deps: SweepStaleReviewsDeps,
now: number = Date.now(),
): number {
let escalatedCount = 0;
const reviewTasks = deps.db.queryAll<{
id: string;
assigneeAgentId: string | null;
updatedAt: string;
title: string;
}>(
`SELECT id, assignee_agent_id as assigneeAgentId, updated_at as updatedAt, title
FROM tasks
WHERE company_id = ? AND status = 'in_review'`,
companyId,
);
for (const task of reviewTasks) {
const reviewAge = now - new Date(task.updatedAt).getTime();
if (reviewAge < REVIEW_TIMEOUT_MS) continue;
if (!task.assigneeAgentId) continue;
// The reviewer is the assignee's direct manager
const reviewer = deps.db.getManagerForAgent(task.assigneeAgentId);
if (!reviewer) continue; // No manager = CEO-assigned task, skip escalation
// D-09: Escalate to the reviewer's manager (manager's manager)
const chain = deps.db.getChainOfCommand(reviewer.id);
const escalationTarget = chain.length > 0 ? chain[0] : null;
if (escalationTarget) {
deps.dispatchAgentMessage(
{
db: deps.db,
emitEvent: deps.emitEvent,
publishDomainChanged: deps.publishDomainChanged,
wakeAgentIfPossible: deps.wakeAgentIfPossible,
},
{
companyId,
fromAgentId: reviewer.id,
toAgentId: escalationTarget.id,
channel: "direct",
subject: `Review timeout: ${task.title}`,
body: `Review by ${reviewer.name} on "${task.title}" has been pending for over 30 minutes. Auto-escalating.`,
priority: "urgent",
},
"subtask_completed",
);
} else {
// D-09: No higher manager (reviewer is CEO) -- notify human
deps.notify({
title: "Review Timeout",
body: `"${task.title}" review has been pending over 30 minutes. Auto-approving.`,
urgency: "critical",
navigation: { section: "tasks", entityId: task.id },
batchKey: "review_escalation",
});
}
// D-10: Auto-approve any pending approval linked to this task
const pendingApprovals = deps.db.queryAll<{ id: string }>(
`SELECT id FROM approvals
WHERE company_id = ? AND related_task_id = ? AND state = 'pending'
LIMIT 1`,
companyId,
task.id,
);
for (const approval of pendingApprovals) {
try {
deps.resolveApprovalDecision({
db: deps.db,
handleApprovalResolved: deps.handleApprovalResolved,
companyId,
approvalId: approval.id,
state: "approved",
decisionNote: "Auto-approved: review timeout exceeded 30 minutes.",
detail: "Auto-approved by review timeout escalation",
actor: "system",
});
} catch {
// Already decided (race condition with manual approval) -- skip silently (Pitfall 3)
}
}
// D-10: Advance task status from in_review to done
try {
deps.db.advanceTaskStatus(task.id, "done", "Auto-advanced: review timeout exceeded 30 minutes.");
deps.handleTaskStatusChange(task.id, companyId, "in_review", "done");
} catch {
// Task may have been updated by manual action
}
escalatedCount++;
}
return escalatedCount;
}
+63 -8
View File
@@ -4,6 +4,8 @@ import { getConnectorDefinition } from "./connectors";
import type { AppDatabase } from "./database";
import type { RunRecord } from "@shared/types";
import type { NotificationOptions } from "./notification-service";
import type { createBudgetEnforcementService } from "./budget-enforcement-service";
import type { createSessionManagementService } from "./session-management-service";
const runStatusSchema = z.enum(["queued", "running", "succeeded", "failed", "cancelled", "timed_out", "interrupted"]);
export const workerMessageSchema = z.union([
@@ -32,6 +34,8 @@ export const workerMessageSchema = z.union([
sessionDisplayId: z.string().nullable(),
costUsd: z.number().nullable(),
unattributedCost: z.boolean(),
inputTokens: z.number().nullable(),
outputTokens: z.number().nullable(),
}),
]);
type WorkerUiEvent = Extract<z.infer<typeof workerMessageSchema>, { type: "run-log" | "run-status" }>;
@@ -63,6 +67,8 @@ interface WorkerServiceDependencies {
}) => void;
processRunCompletion: (run: RunRecord, event: z.infer<typeof workerMessageSchema> & { type: "run-finished" }) => void;
notify: (options: NotificationOptions) => void;
budgetService?: ReturnType<typeof createBudgetEnforcementService>;
sessionService?: ReturnType<typeof createSessionManagementService>;
}
export function createWorkerService({
@@ -76,6 +82,8 @@ export function createWorkerService({
dispatchRunToWorker,
processRunCompletion,
notify,
budgetService,
sessionService,
}: WorkerServiceDependencies) {
const MAX_RETRY_ATTEMPTS = 2;
const RETRY_COOLDOWN_MS = 30_000; // 30 seconds
@@ -107,6 +115,7 @@ export function createWorkerService({
const snapshot = db.listSnapshot();
const affectedAgentIds = new Set<string>();
const affectedCompanyIds = new Map<string, string>();
const requeueCandidates: Array<{ id: string; taskId: string; agentId: string; companyId: string; workspaceId: string | null; connectorId: string }> = [];
for (const run of snapshot.runs) {
@@ -126,7 +135,10 @@ export function createWorkerService({
if (run.workspaceId) {
db.releaseWorkspaceLock(run.workspaceId, run.id);
}
// EXEC-02: Release execution lock on crash recovery
db.releaseTaskLock(run.taskId, run.id);
affectedAgentIds.add(run.agentId);
affectedCompanyIds.set(run.agentId, run.companyId);
const chain = db.getChainOfCommand(run.agentId);
if (chain.length > 0) {
@@ -145,10 +157,13 @@ export function createWorkerService({
}
for (const agentId of affectedAgentIds) {
const pendingWakes = db.consumePendingWakes(agentId);
for (const wake of pendingWakes) {
wakeAgentIfPossible(agentId, wake.companyId, wake.trigger);
// EXEC-05: Promote deferred wakes after crash recovery
const cid = affectedCompanyIds.get(agentId) ?? "";
const promoted = db.promoteWakeupRequests(agentId, cid);
if (promoted > 0) {
logger.info(`[worker] Promoted ${promoted} deferred wake(s) for agent ${agentId} after crash recovery`);
}
// Let heartbeat scheduler pick up promoted wakes on next tick
}
publishDomainChanged();
@@ -280,6 +295,31 @@ export function createWorkerService({
entityId: run.id,
detail: event.summary ?? event.errorMessage ?? "",
});
// SESS-01/SESS-02/SESS-03/SESS-05: Token accumulation and rotation evaluation
if (task && sessionService) {
const activeSession = db.getActiveTaskSession(task.id);
if (activeSession) {
// D-08/D-05: Accumulate token usage for ALL adapters (including exempt ones)
sessionService.accumulateTokenUsage(
activeSession.id,
event.inputTokens ?? 0,
event.outputTokens ?? 0,
event.runId,
);
// D-03: Evaluate rotation after accumulation (after each run completes)
const rotationResult = sessionService.evaluateRotation(activeSession, run.connectorId);
if (rotationResult.shouldRotate) {
// Re-read to get updated counts after accumulation
const updatedSession = db.getActiveTaskSession(task.id);
if (updatedSession) {
sessionService.rotateSession(updatedSession);
logger.info(`[session] Rotated session for task ${task.id}: ${rotationResult.reason}`);
}
}
}
}
emitEvent({ type: "run-updated", runId: run.id, action: "completed" });
emitEvent({
type: "run-status",
@@ -323,6 +363,16 @@ export function createWorkerService({
emitEvent({ type: "agent-updated", agentId: updatedAgent.id, action: "paused" });
}
// Policy-based incident check after run completion (Phase 5)
if (budgetService && typeof event.costUsd === "number" && Number.isFinite(event.costUsd)) {
try {
const task = db.getTask(run.taskId);
budgetService.checkAndCreateIncidents(run.companyId, run.agentId, task?.projectId ?? null);
} catch (err) {
logger.error(`[budget] Incident check failed: ${err instanceof Error ? err.message : String(err)}`);
}
}
if ((event.status === "failed" || event.status === "timed_out") && !isQuitting() && !destroyed) {
const retryCount = db.getTaskRetryCount(run.taskId);
@@ -381,10 +431,15 @@ export function createWorkerService({
}
}
const pendingWakes = db.consumePendingWakes(run.agentId);
if (pendingWakes.length > 0 && !isQuitting() && !destroyed) {
for (const wake of pendingWakes) {
wakeAgentIfPossible(run.agentId, wake.companyId, wake.trigger);
// EXEC-05: Promote deferred wakes after run completion
let hadDeferredWakes = false;
if (!isQuitting() && !destroyed) {
const promoted = db.promoteWakeupRequests(run.agentId, run.companyId);
if (promoted > 0) {
hadDeferredWakes = true;
logger.info(`[worker] Promoted ${promoted} deferred wake(s) for agent ${run.agentId}`);
// Trigger processing of newly-promoted wakes
wakeAgentIfPossible(run.agentId, run.companyId, "deferred_wake_promoted");
}
}
@@ -404,7 +459,7 @@ export function createWorkerService({
}
// Self-wake with circuit breaker to prevent infinite loops
if (pendingWakes.length === 0 && !isQuitting() && !destroyed) {
if (!hadDeferredWakes && !isQuitting() && !destroyed) {
try {
const actionableTasks = db.listAgentTasks(run.agentId, run.companyId, ["todo", "in_progress"]);
if (actionableTasks.length > 0) {