43-data-security-pipeline.ts

/**
 * 43 - Data Security Pipeline
 *
 * Demonstrates a sequential pipeline with data flow control where
 * sensitive information is collected, redacted, and then presented safely:
 *
 *   collector >> validator >> responder
 *
 * - collector: Fetches raw user data using tools (includes PII).
 * - validator: Redacts sensitive fields (SSN, balances, email) using tools.
 * - responder: Presents the safe, redacted data to the user.
 *
 * This pattern enforces a security boundary between data access and
 * user-facing responses, ensuring PII never reaches the final output.
 *
 * Requirements:
 *   - Conductor server with LLM support
 *   - AGENTSPAN_SERVER_URL=http://localhost:8080/api as environment variable
 *   - AGENTSPAN_LLM_MODEL=openai/gpt-4o-mini as environment variable
 */

import { Agent, AgentRuntime, tool } from '@io-orkes/conductor-javascript/agents';
import { llmModel } from './settings';

// -- Data tools ---------------------------------------------------------------

const fetchUserData = tool(
  async (args: { userId: string }) => {
    const users: Record<string, Record<string, unknown>> = {
      U001: {
        name: 'Alice Johnson',
        email: 'alice@example.com',
        role: 'admin',
        ssn_last4: '1234',
        account_balance: 15000.0,
      },
      U002: {
        name: 'Bob Smith',
        email: 'bob@example.com',
        role: 'user',
        ssn_last4: '5678',
        account_balance: 3200.0,
      },
    };
    return users[args.userId] ?? { error: `User ${args.userId} not found` };
  },
  {
    name: 'fetch_user_data',
    description: 'Fetch user data from the database.',
    inputSchema: {
      type: 'object',
      properties: {
        userId: { type: 'string', description: 'The user\'s identifier' },
      },
      required: ['userId'],
    },
  },
);

// -- Redaction tools ----------------------------------------------------------

const redactSensitiveFields = tool(
  async (args: { data: string }) => {
    let parsed: Record<string, unknown>;
    try {
      parsed = JSON.parse(args.data) as Record<string, unknown>;
    } catch {
      return { error: 'Could not parse data for redaction' };
    }

    const sensitiveKeys = new Set(['ssn_last4', 'account_balance', 'email']);
    const redacted: Record<string, unknown> = {};
    for (const [k, v] of Object.entries(parsed)) {
      redacted[k] = sensitiveKeys.has(k) ? '***REDACTED***' : v;
    }
    return { redacted_data: redacted };
  },
  {
    name: 'redact_sensitive_fields',
    description: 'Redact sensitive fields from data before responding to users.',
    inputSchema: {
      type: 'object',
      properties: {
        data: { type: 'string', description: 'JSON string of user data to redact' },
      },
      required: ['data'],
    },
  },
);

// -- Pipeline agents ----------------------------------------------------------

// Data collector fetches raw user data
export const collector = new Agent({
  name: 'data_collector',
  model: llmModel,
  instructions:
    'You are a data collection agent. When asked about a user, ' +
    'call fetch_user_data with their ID. Pass the raw data along ' +
    'to the next agent for security review.',
  tools: [fetchUserData],
});

// Validator enforces data security policy
export const validator = new Agent({
  name: 'security_validator',
  model: llmModel,
  instructions:
    'You are a security validator. Review data for sensitive information ' +
    '(SSN, account balances, email addresses). Use the redact_sensitive_fields ' +
    'tool to redact any sensitive data before passing it along. ' +
    'Only pass redacted data to the next agent.',
  tools: [redactSensitiveFields],
});

// Responder formats the final answer
export const responder = new Agent({
  name: 'responder',
  model: llmModel,
  instructions:
    'You are a customer service agent. Use the validated, redacted data ' +
    'to answer the user\'s question. NEVER reveal redacted information. ' +
    'If data shows ***REDACTED***, explain that the information is ' +
    'restricted for security reasons.',
});

// Sequential pipeline enforces data flow: collect -> validate -> respond
const pipeline = collector.pipe(validator).pipe(responder);

async function main() {
  const runtime = new AgentRuntime();
  try {
    const result = await runtime.run(
    pipeline,
    'Tell me everything about user U001 including their financial details.',
    );
    result.printResult();

    // Production pattern:
    // 1. Deploy once during CI/CD (optional -- serve() below also deploys):
    // await runtime.deploy(pipeline);
    // CLI alternative:
    // agentspan deploy --package sdk/typescript/examples --agents data_collector
    //
    // 2. In a separate long-lived worker process (deploys + registers workers + starts polling):
    // await runtime.serve(pipeline);
  } finally {
    await runtime.shutdown();
  }
}

main().catch(console.error);