Governance & Guardrails

┌──────────────────────────────────────────────────────────┐
│  Layer 1: Permission Whitelist (agent config)            │
│  Which tools can this agent even call?                   │
└──────────────────────────────┬───────────────────────────┘
                               │ passes
┌──────────────────────────────▼───────────────────────────┐
│  Layer 2: Output Validators (post-generation)            │
│  Character limits, brand voice, banned words             │
└──────────────────────────────┬───────────────────────────┘
                               │ passes
┌──────────────────────────────▼───────────────────────────┐
│  Layer 3: Approvals Queue (human-in-the-loop)            │
│  Human signs off before ANY client-facing action         │
└──────────────────────────────┬───────────────────────────┘
                               │ approved
┌──────────────────────────────▼───────────────────────────┐
│  Layer 4: Rate Limiting & Budget Halt                    │
│  Prevent runaway loops and cost overruns                 │
└──────────────────────────────────────────────────────────┘

async function dispatchToolCall(
  toolName:    string,
  method:      string,
  input:       unknown,
  agentConfig: AgentConfig,
  taskRun:     TaskRun,
): Promise<unknown> {
  // 1. Permission check
  if (!agentConfig.toolNames.includes(toolName)) {
    throw new PermissionDeniedError(
      `Agent "${agentConfig.role}" does not have permission to use tool "${toolName}"`
    );
  }
 
  // 2. Write-action approval check
  if (isWriteAction(toolName, method)) {
    await assertApprovalExists(taskRun.taskId, toolName, method);
  }
 
  // 3. Execute
  return integrations[toolName][method](input);
}

interface CharacterLimitRule {
  field:    string;
  platform: string;
  maxChars: number;
  severity: 'hard_fail' | 'warn';
}
 
const PLATFORM_LIMITS: CharacterLimitRule[] = [
  { field: 'headline',    platform: 'google_ads',  maxChars: 30,  severity: 'hard_fail' },
  { field: 'description', platform: 'google_ads',  maxChars: 90,  severity: 'hard_fail' },
  { field: 'primary_text',platform: 'meta_ads',    maxChars: 125, severity: 'hard_fail' },
  { field: 'headline',    platform: 'meta_ads',    maxChars: 40,  severity: 'hard_fail' },
  { field: 'subject_line',platform: 'email',       maxChars: 60,  severity: 'warn'      },
  { field: 'preview_text',platform: 'email',       maxChars: 100, severity: 'warn'      },
  { field: 'tweet',       platform: 'x_twitter',   maxChars: 280, severity: 'hard_fail' },
  { field: 'linkedin_post',platform: 'linkedin',   maxChars: 3000,severity: 'warn'      },
];
 
function validateCharacterLimits(content: DeliverableContent): ValidationResult[] {
  // Parse structured content, check each field against limits
  // Return array of { field, platform, actual, limit, severity }
}

async function validateBrandVoice(
  content: string,
  brandVoiceSkill: string,
  clientId: string
): Promise<BrandVoiceScore> {
  const prompt = `
You are a brand voice reviewer. Given the brand voice guide and the content below,
score the content from 0-100 for brand alignment and list any violations.
 
BRAND VOICE GUIDE:
${brandVoiceSkill}
 
CONTENT TO REVIEW:
${content}
 
Respond with JSON: { "score": number, "violations": string[], "recommendations": string[] }
  `;
 
  const result = await ollamaExecute('gemma3:4b', prompt);
  return JSON.parse(result.text);
}

const GLOBAL_BANNED_WORDS = ['guaranteed', 'best in class', 'world-class', /* ... */];
 
async function validateBannedWords(content: string, clientId: string): Promise<string[]> {
  const clientBannedWords = await getClientBannedWords(clientId);  // from client settings
  const allBanned = [...GLOBAL_BANNED_WORDS, ...clientBannedWords];
 
  return allBanned.filter(word =>
    content.toLowerCase().includes(word.toLowerCase())
  );
}

async function injectDisclaimers(
  content: string,
  clientId: string,
  deliverableType: string
): Promise<string> {
  const disclaimers = await getRequiredDisclaimers(clientId, deliverableType);
  if (!disclaimers.length) return content;
  return content + '\n\n---\n\n' + disclaimers.join('\n\n');
}

CREATE TABLE approvals (
  id                  UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  tenant_id           UUID NOT NULL REFERENCES tenants(id),
  type                VARCHAR(50) NOT NULL,       -- ApprovalType enum
  title               TEXT NOT NULL,              -- short label in DM Portal inbox
  description         TEXT,                       -- context for the reviewer
  status              VARCHAR(20) NOT NULL DEFAULT 'pending',
                                                  -- pending | approved | rejected | expired
  risk_level          VARCHAR(10) NOT NULL,        -- low | medium | high
  created_by_type     VARCHAR(10) NOT NULL,        -- 'system' | 'agent'
  created_by_agent_role VARCHAR(100),             -- set when created_by_type = 'agent'
  reviewed_by_user_id UUID REFERENCES users(id),
  reviewer_notes      TEXT,
  validation_results  JSONB,                      -- output validator results (content_review only)
  options             TEXT[],                     -- reviewer choice options (if applicable)
  expires_at          TIMESTAMPTZ,
  created_at          TIMESTAMPTZ DEFAULT NOW(),
  resolved_at         TIMESTAMPTZ
);
 
-- Links one approval to one or more activities that are blocked on it
CREATE TABLE approval_linked_activities (
  id           UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  approval_id  UUID NOT NULL REFERENCES approvals(id) ON DELETE CASCADE,
  activity_id  UUID NOT NULL REFERENCES activities(id),
  linked_at    TIMESTAMPTZ DEFAULT NOW(),
  UNIQUE (approval_id, activity_id)
);
 
CREATE INDEX ON approvals(tenant_id, status);
CREATE INDEX ON approvals(expires_at) WHERE status = 'pending';
CREATE INDEX ON approval_linked_activities(approval_id);
CREATE INDEX ON approval_linked_activities(activity_id);

// After agent deliverable saved (content_review)
const approval = await db.insert(approvals).values({
  tenantId:         tenant.id,
  type:             'content_review',
  title:            `Review: ${deliverable.name}`,
  description:      `Agent-produced deliverable ready for review.`,
  status:           'pending',
  riskLevel:        deriveRiskLevel(deliverable.type),
  createdByType:    'system',
  validationResults: validatorOutput,
  expiresAt:        addHours(new Date(), expiryHoursForRisk(riskLevel)),
}).returning();
 
// Link to the producing activity
await db.insert(approvalLinkedActivities).values({
  approvalId:  approval.id,
  activityId:  activity.id,
});

// Tool handler for create_approval
async function handleCreateApproval(
  input:      CreateApprovalInput,
  context:    ToolCallContext,
): Promise<CreateApprovalResult> {
  const approval = await db.insert(approvals).values({
    tenantId:            context.tenantId,
    type:                input.type,
    title:               input.title,
    description:         input.description,
    status:              'pending',
    riskLevel:           input.riskLevel ?? defaultRiskForType(input.type),
    createdByType:       'agent',
    createdByAgentRole:  context.agentRole,
    options:             input.options ?? null,
    expiresAt:           addHours(new Date(), input.expiresInHours ?? expiryHoursForRisk(riskLevel)),
  }).returning();
 
  // Block the calling activity if requested (default: true)
  const activityIdsToBlock = [...(input.linkedActivityIds ?? [])];
  if (input.blockCurrentActivity !== false) {
    activityIdsToBlock.push(context.activityId);
  }
 
  // Link and block all target activities
  for (const activityId of activityIdsToBlock) {
    await db.insert(approvalLinkedActivities).values({ approvalId: approval.id, activityId });
    await db.update(activities)
      .set({ status: 'awaiting_approval', approvalId: approval.id })
      .where(eq(activities.id, activityId));
  }
 
  return {
    status:     'approval_created',
    approvalId: approval.id,
    message:    `Approval created. ${activityIdsToBlock.length} activities suspended until resolved.`,
  };
}

async function onApprovalResolved(
  approvalId:    string,
  tenantId:      string,
  approved:      boolean,
  reviewerNotes: string,
  userId:        string,
  chosenOption?: string,   // if the approval had options[], which one the reviewer selected
): Promise<void> {
  // 1. Update the approval record
  await db.update(approvals)
    .set({
      status:           approved ? 'approved' : 'rejected',
      reviewedByUserId: userId,
      reviewerNotes,
      resolvedAt:       new Date(),
    })
    .where(eq(approvals.id, approvalId));
 
  // 2. Fetch all linked activities
  const links = await db.query.approvalLinkedActivities.findMany({
    where: eq(approvalLinkedActivities.approvalId, approvalId),
  });
 
  // 3. Re-enqueue each linked activity with the resolution context
  for (const link of links) {
    const activity = await getActivity(link.activityId, tenantId);
 
    await db.update(activities)
      .set({ status: 'created', approvalId: null })
      .where(eq(activities.id, link.activityId));
 
    await enqueueActivity(activity, tenantId, activity.assigneeAgentRole, {
      wakeReason:       approved ? 'review_approved' : 'review_feedback',
      reviewApproved:   approved,
      reviewerFeedback: reviewerNotes
        + (chosenOption ? `\n\nSelected option: ${chosenOption}` : ''),
    });
  }
}

// Runs every hour
async function checkApprovalExpiry(): Promise<void> {
  const now = new Date();
 
  // 1. Warning: 24h before expiry — send in-app notification to reviewers
  const approachingExpiry = await db.query.approvals.findMany({
    where: and(
      eq(approvals.status, 'pending'),
      lte(approvals.expiresAt, addHours(now, 24)),
      gt(approvals.expiresAt, now),
    ),
  });
  for (const approval of approachingExpiry) {
    await notifyReviewers(approval, 'expiry_warning');
  }
 
  // 2. Expired: mark as expired, block all linked activities
  const expired = await db.query.approvals.findMany({
    where: and(
      eq(approvals.status, 'pending'),
      lte(approvals.expiresAt, now),
    ),
  });
  for (const approval of expired) {
    await db.update(approvals)
      .set({ status: 'expired', resolvedAt: now })
      .where(eq(approvals.id, approval.id));
 
    // Escalate: create a high-priority human task in DM Portal
    await createEscalationActivity({
      tenantId:    approval.tenantId,
      name:        `ESCALATED: Approval expired — ${approval.title}`,
      description: `Approval "${approval.title}" expired without a decision. `
                 + `${linkedCount} activities are blocked. Resolve immediately.`,
      priority:    'urgent',
    });
  }
}

new Worker(queue, processor, {
  connection: redis,
  concurrency: 3,
  limiter: {
    max:      10,    // max 10 jobs
    duration: 60000, // per 60 seconds
  }
});

async function runWatchdog(): Promise<void> {
  // Agent running more than maxTimeoutMs
  const stuckRuns = await db.query.taskRuns.findMany({
    where: and(
      eq(taskRuns.status, 'running'),
      lt(taskRuns.startedAt, new Date(Date.now() - MAX_TASK_DURATION_MS))
    )
  });
 
  for (const run of stuckRuns) {
    await killTaskRun(run);
    await createEscalation(run.taskId, 'watchdog_timeout');
  }
 
  // Campaign cost exceeded cap
  const overBudget = await db.query.campaigns.findMany({
    where: and(
      isNotNull(campaigns.budgetCapUsd),
      gt(campaigns.totalCostUsd, campaigns.budgetCapUsd)
    )
  });
 
  for (const campaign of overBudget) {
    await pauseCampaign(campaign.id);
    await drainCampaignJobs(campaign.id);
    await notifyBudgetBreached(campaign);
  }
}

apps/api/src/
├── middleware/                   # Rate limiting middleware
└── routes/
    └── approvals.ts              # HITL approval endpoints (create, resolve, query)

packages/agent-engine/src/
├── validators/                   # Character limits, brand voice, banned words, disclaimers
├── budget.ts                     # Budget halt logic
└── watchdog.ts                   # Runaway detection

packages/queue/src/
├── workers.ts                    # BullMQ rate limiter config
└── approval.ts                   # enqueueApprovalLinkedActivities() — fan-out on resolve

packages/integrations/src/
└── control-plane/
    └── create-approval.ts        # create_approval tool handler

// Called by the LLM adapter after every streaming chunk that includes usage metadata
async function trackAndCheckBudget(
  tenantId:      string,
  campaignId:    string,
  activityRunId: string,
  newCostUsd:    number,
): Promise<'continue' | 'halt'> {
  // Atomic increment + check in a single DB transaction
  const [updatedRun] = await db
    .update(activityRuns)
    .set({ costUsd: sql`cost_usd + ${newCostUsd}` })
    .where(eq(activityRuns.id, activityRunId))
    .returning({ costUsd: activityRuns.costUsd });
 
  // Check all three caps in order (cheapest check first)
  const agentConfig = await getCachedAgentConfig(tenantId);
  if (agentConfig.maxCostUsdPerActivity && updatedRun.costUsd > agentConfig.maxCostUsdPerActivity) {
    return 'halt';
  }
 
  const campaign = await getCachedCampaign(campaignId);
  if (campaign.budgetCapUsd) {
    const campaignSpend = await getCampaignTotalSpend(campaignId);  // Redis-cached, 5s TTL
    if (campaignSpend > campaign.budgetCapUsd) return 'halt';
  }
 
  const tenant = await getCachedTenant(tenantId);
  if (tenant.monthlySpendCapUsd) {
    const tenantSpend = await getTenantMonthlySpend(tenantId);  // Redis-cached, 5s TTL
    if (tenantSpend > tenant.monthlySpendCapUsd) return 'halt';
  }
 
  return 'continue';
}