# MySQL Question Answering Implementation Plan

> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.

**Goal:** Build a MySQL-only, read-only intelligent data questioning backend path for Neta Agents, with safe data-source configuration, schema/sample/query tools, cross-table JOIN support, audit, and a prompt skill.

**Architecture:** Add a `mysql` toolset backed by `netaclaw_data_source`, `SecretCryptoService`, `MysqlPoolManager`, schema/query services, and guarded Tool implementations. Security-critical rules live in services and tools; `data-analyst-mysql` only guides Agent behavior.

**Tech Stack:** Midway.js, TypeORM, mysql2/promise, TypeBox tool schemas, Jest, NetaClaw Tool Catalog/Resolver/Runtime Policy, prompt Skill.

---

## File Structure

Create:

- `packages/backend/src/modules/netaclaw/entity/data_source.ts` — MySQL data-source configuration entity.
- `packages/backend/src/modules/netaclaw/entity/data_source_query_audit.ts` — query audit entity.
- `packages/backend/src/modules/netaclaw/service/secret_crypto.ts` — shared AES-256-GCM helper for new encrypted secrets.
- `packages/backend/src/modules/netaclaw/service/data_source.ts` — admin/runtime data-source lookup, authorization, save, and test connection service.
- `packages/backend/src/modules/netaclaw/service/mysql_pool.ts` — cached mysql2 pool manager.
- `packages/backend/src/modules/netaclaw/service/mysql_schema.ts` — information_schema mapping and schema/sample helpers.
- `packages/backend/src/modules/netaclaw/service/mysql_query.ts` — read-only SQL execution, masked-column rejection, row limits, audit.
- `packages/backend/src/modules/netaclaw/tools/builtin/mysql.ts` — `mysql_list_sources`, `mysql_schema`, `mysql_table_sample`, `mysql_query`.
- `packages/backend/src/modules/netaclaw/controller/admin/data_source.ts` — admin APIs for data-source list/save/test/delete.
- `packages/backend/skills/data-analyst-mysql/SKILL.md` — prompt skill for MySQL data analysis.
- `packages/backend/test/secret_crypto.test.ts`
- `packages/backend/test/mysql_sql_guard.test.ts`
- `packages/backend/test/mysql_schema_mapper.test.ts`
- `packages/backend/test/mysql_query_service.test.ts`
- `packages/backend/test/mysql_pool_manager.test.ts`
- `packages/backend/test/data_source_projection.test.ts`

Modify:

- `packages/backend/src/entities.ts` — add data source and query audit entity imports to the generated entity list. This file is marked generated, but existing tests assert it directly, so update it for this change.
- `packages/backend/src/modules/netaclaw/tools/catalog.ts` — add `TOOLSET_MYSQL` and import `./builtin/mysql.js`.
- `packages/backend/src/modules/netaclaw/tools/manifest.ts` — route `mysql_*` tools as main-process proxy, classify as custom/text/parallel.
- `packages/backend/src/modules/netaclaw/service/tool_resolver.ts` — inject MySQL tool dependencies into runtime tool map.
- `packages/backend/test/entity_exports.test.ts` — assert new entities are exported.
- `packages/backend/test/tool_resolver.test.ts` — assert mysql toolset resolution and subagent route.

Do not implement frontend UI in this pass. The admin controller is enough for Phase 1/2 backend configuration.

---

## Task 1: Entities And Secret Crypto

**Files:**

- Create: `packages/backend/src/modules/netaclaw/entity/data_source.ts`
- Create: `packages/backend/src/modules/netaclaw/entity/data_source_query_audit.ts`
- Create: `packages/backend/src/modules/netaclaw/service/secret_crypto.ts`
- Modify: `packages/backend/src/entities.ts`
- Modify: `packages/backend/test/entity_exports.test.ts`
- Test: `packages/backend/test/secret_crypto.test.ts`
- Test: `packages/backend/test/entity_exports.test.ts`
- Test: `packages/backend/test/data_source_projection.test.ts`

- [ ] **Step 1: Write the failing secret crypto test**

Add `packages/backend/test/secret_crypto.test.ts`:

```ts
import { SecretCryptoService } from '../src/modules/netaclaw/service/secret_crypto.js';

describe('SecretCryptoService', () => {
  const originalNetaSecret = process.env.NETA_SECRET_KEY;
  const originalAppSecret = process.env.APP_SECRET;

  beforeEach(() => {
    process.env.NETA_SECRET_KEY = 'unit-test-secret-key';
    delete process.env.APP_SECRET;
  });

  afterEach(() => {
    if (originalNetaSecret === undefined) delete process.env.NETA_SECRET_KEY;
    else process.env.NETA_SECRET_KEY = originalNetaSecret;
    if (originalAppSecret === undefined) delete process.env.APP_SECRET;
    else process.env.APP_SECRET = originalAppSecret;
  });

  it('encrypts text without embedding the plaintext', () => {
    const service = new SecretCryptoService();

    const encrypted = service.encryptText('mysql-password-123');
    const envelope = JSON.parse(Buffer.from(encrypted, 'base64').toString('utf8'));

    expect(encrypted).not.toContain('mysql-password-123');
    expect(envelope).toEqual(expect.objectContaining({ v: 1, alg: 'aes-256-gcm' }));
    expect(envelope.iv).toEqual(expect.any(String));
    expect(envelope.tag).toEqual(expect.any(String));
    expect(envelope.ct).toEqual(expect.any(String));
    expect(service.decryptText(encrypted)).toBe('mysql-password-123');
  });

  it('encrypts and decrypts JSON values', () => {
    const service = new SecretCryptoService();

    const encrypted = service.encryptJson({ password: 'secret', token: 'abc' });

    expect(encrypted).not.toContain('secret');
    expect(service.decryptJson(encrypted)).toEqual({ password: 'secret', token: 'abc' });
  });

  it('rejects tampered ciphertext', () => {
    const service = new SecretCryptoService();
    const encrypted = service.encryptText('secret');
    const tampered = encrypted.slice(0, -2) + 'xx';

    expect(() => service.decryptText(tampered)).toThrow();
  });
});
```

- [ ] **Step 2: Run test to verify RED**

Run:

```bash
pnpm --filter @neta/backend test -- --runInBand test/secret_crypto.test.ts
```

Expected: FAIL because `secret_crypto.ts` does not exist.

- [ ] **Step 3: Implement `SecretCryptoService`**

Create `packages/backend/src/modules/netaclaw/service/secret_crypto.ts`:

```ts
import { Provide, Scope, ScopeEnum } from '@midwayjs/core';
import * as crypto from 'crypto';

@Provide()
@Scope(ScopeEnum.Singleton)
export class SecretCryptoService {
  private readonly algorithm = 'aes-256-gcm' as const;
  private readonly ivLength = 12;

  private deriveKey(): Buffer {
    const raw = process.env.NETA_SECRET_KEY || process.env.APP_SECRET;
    if (!raw) {
      throw new Error('NETA_SECRET_KEY or APP_SECRET is required for secret encryption');
    }
    return crypto.createHash('sha256').update(raw).digest();
  }

  encryptText(value: string): string {
    const iv = crypto.randomBytes(this.ivLength);
    const cipher = crypto.createCipheriv(this.algorithm, this.deriveKey(), iv);
    const encrypted = Buffer.concat([cipher.update(value, 'utf8'), cipher.final()]);
    const tag = cipher.getAuthTag();
    return Buffer.from(JSON.stringify({
      v: 1,
      alg: this.algorithm,
      iv: iv.toString('base64'),
      tag: tag.toString('base64'),
      ct: encrypted.toString('base64'),
    }), 'utf8').toString('base64');
  }

  decryptText(ciphertext: string): string {
    const envelope = JSON.parse(Buffer.from(ciphertext, 'base64').toString('utf8'));
    if (!envelope || envelope.v !== 1 || envelope.alg !== this.algorithm) {
      throw new Error('Unsupported ciphertext envelope');
    }
    const iv = Buffer.from(envelope.iv, 'base64');
    const tag = Buffer.from(envelope.tag, 'base64');
    const encrypted = Buffer.from(envelope.ct, 'base64');
    const decipher = crypto.createDecipheriv(this.algorithm, this.deriveKey(), iv);
    decipher.setAuthTag(tag);
    return Buffer.concat([decipher.update(encrypted), decipher.final()]).toString('utf8');
  }

  encryptJson(value: Record<string, string>): string {
    return this.encryptText(JSON.stringify(value));
  }

  decryptJson(ciphertext: string): Record<string, string> {
    const parsed = JSON.parse(this.decryptText(ciphertext));
    if (!parsed || typeof parsed !== 'object' || Array.isArray(parsed)) {
      throw new Error('Invalid encrypted JSON payload');
    }
    return Object.fromEntries(Object.entries(parsed).map(([key, value]) => [key, String(value)]));
  }
}
```

- [ ] **Step 4: Run test to verify GREEN**

Run:

```bash
pnpm --filter @neta/backend test -- --runInBand test/secret_crypto.test.ts
```

Expected: PASS.

- [ ] **Step 5: Write failing entity export test**

Modify `packages/backend/test/entity_exports.test.ts`:

```ts
import { entities } from '../src/entities.js';
import { NetaClawAgentSessionEntity } from '../src/modules/netaclaw/entity/agent_session.js';
import { NetaClawAgentSessionEntryEntity } from '../src/modules/netaclaw/entity/agent_session_entry.js';
import { NetaClawDataSourceEntity } from '../src/modules/netaclaw/entity/data_source.js';
import { NetaClawDataSourceQueryAuditEntity } from '../src/modules/netaclaw/entity/data_source_query_audit.js';

describe('entities exports', () => {
  it('includes SubagentSession entity', () => {
    const entityNames = entities
      .map(e => (typeof e === 'function' ? e.name : ''))
      .filter(Boolean);

    expect(entityNames.some(n => n.includes('SubagentSession'))).toBe(true);
  });

  it('includes session-tree MySQL entities', () => {
    expect(entities).toContain(NetaClawAgentSessionEntity);
    expect(entities).toContain(NetaClawAgentSessionEntryEntity);
  });

  it('includes MySQL question answering entities', () => {
    expect(entities).toContain(NetaClawDataSourceEntity);
    expect(entities).toContain(NetaClawDataSourceQueryAuditEntity);
  });
});
```

- [ ] **Step 6: Run test to verify RED**

Run:

```bash
pnpm --filter @neta/backend test -- --runInBand test/entity_exports.test.ts
```

Expected: FAIL because entity files do not exist.

- [ ] **Step 7: Implement data source entities**

Create `packages/backend/src/modules/netaclaw/entity/data_source.ts`:

```ts
import { BaseEntity } from '../../base/entity/base.js';
import { Column, Entity, Index } from 'typeorm';

export interface NetaClawDataSourceExtra {
  ssl?: boolean | Record<string, unknown>;
  connectTimeout?: number;
  queryTimeoutMs?: number;
  maxRows?: number;
  allowedTables?: string[];
  blockedTables?: string[];
  maskedColumns?: Record<string, 'hash' | 'partial' | 'redact'>;
  schemaVisibility?: 'allowed-only' | 'all-names-only';
  maxJoinTables?: number;
  poolConnectionLimit?: number;
}

@Entity('netaclaw_data_source')
export class NetaClawDataSourceEntity extends BaseEntity {
  @Index({ unique: true })
  @Column({ comment: '数据源唯一名', length: 100 })
  name: string;

  @Column({ comment: '显示名称', length: 200, nullable: true })
  label: string;

  @Index()
  @Column({ comment: '数据源类型', length: 20, default: 'mysql' })
  type: string;

  @Column({ comment: 'MySQL host', length: 255 })
  host: string;

  @Column({ comment: 'MySQL port', default: 3306 })
  port: number;

  @Column({ comment: '默认数据库', length: 128 })
  database: string;

  @Column({ comment: '连接用户名', length: 255 })
  username: string;

  @Column({ comment: '加密后的连接密码', type: 'text', nullable: true })
  passwordEncrypted: string;

  @Column({ comment: '是否只读', default: 1 })
  readonly: number;

  @Index()
  @Column({ comment: '状态 0=禁用 1=启用', default: 1 })
  status: number;

  @Column({ comment: '授权 Agent ID 列表', type: 'json', nullable: true })
  allowedAgentIds: number[];

  @Column({ comment: '扩展配置', type: 'json', nullable: true })
  extra: NetaClawDataSourceExtra;
}
```

Create `packages/backend/src/modules/netaclaw/entity/data_source_query_audit.ts`:

```ts
import { BaseEntity } from '../../base/entity/base.js';
import { Column, Entity, Index } from 'typeorm';

@Entity('netaclaw_data_source_query_audit')
export class NetaClawDataSourceQueryAuditEntity extends BaseEntity {
  @Index()
  @Column({ comment: '数据源 ID' })
  dataSourceId: number;

  @Index()
  @Column({ comment: 'Agent ID', nullable: true })
  agentId: number;

  @Index()
  @Column({ comment: '用户 ID', nullable: true })
  userId: number;

  @Column({ comment: '工具调用 ID', length: 100, nullable: true })
  toolCallId: string;

  @Column({ comment: 'SQL SHA-256', length: 64 })
  sqlHash: string;

  @Column({ comment: 'SQL 摘要', type: 'text', nullable: true })
  sqlPreview: string;

  @Index()
  @Column({ comment: '状态 success/rejected/failed', length: 20 })
  status: string;

  @Column({ comment: '拒绝原因', length: 100, nullable: true })
  rejectReason: string;

  @Column({ comment: '耗时 ms', nullable: true })
  elapsedMs: number;

  @Column({ comment: '返回行数', nullable: true })
  rowCount: number;

  @Column({ comment: '错误代码', length: 50, nullable: true })
  errorCode: string;
}
```

Modify `packages/backend/src/entities.ts` by adding imports after existing netaclaw imports:

```ts
import * as entity51 from './modules/netaclaw/entity/data_source';
import * as entity52 from './modules/netaclaw/entity/data_source_query_audit';
```

and append:

```ts
  ...Object.values(entity51),
  ...Object.values(entity52),
```

- [ ] **Step 8: Run entity and crypto tests to verify GREEN**

Run:

```bash
pnpm --filter @neta/backend test -- --runInBand test/secret_crypto.test.ts test/entity_exports.test.ts
```

Expected: PASS.

- [ ] **Step 9: Write failing data source projection test**

Add `packages/backend/test/data_source_projection.test.ts`:

```ts
import { NetaClawDataSourceService } from '../src/modules/netaclaw/service/data_source.js';

describe('NetaClawDataSourceService projections', () => {
  const source = {
    id: 1,
    name: 'sales_prod',
    label: '销售库',
    type: 'mysql',
    host: '127.0.0.1',
    port: 3306,
    database: 'sales',
    username: 'root',
    passwordEncrypted: 'ciphertext',
    readonly: 1,
    status: 1,
    allowedAgentIds: [9],
    extra: { allowedTables: ['orders'] },
  } as any;

  it('admin projection hides passwords but keeps connection metadata', () => {
    const service = new NetaClawDataSourceService();

    expect(service.toAdminSafe(source)).toEqual(expect.objectContaining({
      name: 'sales_prod',
      host: '127.0.0.1',
      username: 'root',
      hasPassword: true,
    }));
    expect(service.toAdminSafe(source)).not.toHaveProperty('passwordEncrypted');
  });

  it('agent projection hides host username password and permission details', () => {
    const service = new NetaClawDataSourceService();

    expect(service.toAgentSummary(source)).toEqual({
      name: 'sales_prod',
      label: '销售库',
      database: 'sales',
      status: 1,
    });
  });
});
```

- [ ] **Step 10: Run projection test to verify RED**

Run:

```bash
pnpm --filter @neta/backend test -- --runInBand test/data_source_projection.test.ts
```

Expected: FAIL because `toAdminSafe` and `toAgentSummary` do not exist.

- [ ] **Step 11: Implement safe projections**

Modify the service code in this task:

```ts
export type AdminSafeDataSource = Omit<NetaClawDataSourceEntity, 'passwordEncrypted'> & {
  hasPassword: boolean;
};

export type AgentDataSourceSummary = Pick<NetaClawDataSourceEntity, 'name' | 'label' | 'database' | 'status'>;
```

Replace any existing unsafe projection helper with:

```ts
toAdminSafe(entity: NetaClawDataSourceEntity): AdminSafeDataSource {
  const { passwordEncrypted, ...rest } = entity as any;
  return { ...rest, hasPassword: !!passwordEncrypted };
}

toAgentSummary(entity: NetaClawDataSourceEntity): AgentDataSourceSummary {
  return {
    name: entity.name,
    label: entity.label,
    database: entity.database,
    status: entity.status,
  };
}
```

Update `listAdminSafe()` and `listForAgent()` to return the right projections:

```ts
async listAdminSafe(): Promise<AdminSafeDataSource[]> {
  const rows = await this.repo.find({ order: { id: 'DESC' } as any });
  return rows.map(row => this.toAdminSafe(row));
}

async listForAgent(agentId: number): Promise<AgentDataSourceSummary[]> {
  const rows = await this.repo.find({ where: { type: 'mysql', status: 1 } as any, order: { id: 'DESC' } as any });
  return rows
    .filter(row => Array.isArray(row.allowedAgentIds) && row.allowedAgentIds.includes(agentId))
    .map(row => this.toAgentSummary(row));
}
```

Update `saveConfig()` to return `toAdminSafe(saved)`.

- [ ] **Step 12: Run tests to verify GREEN**

Run:

```bash
pnpm --filter @neta/backend test -- --runInBand test/secret_crypto.test.ts test/entity_exports.test.ts test/data_source_projection.test.ts
```

Expected: PASS.

- [ ] **Step 13: Commit**

```bash
git add packages/backend/src/modules/netaclaw/entity/data_source.ts packages/backend/src/modules/netaclaw/entity/data_source_query_audit.ts packages/backend/src/modules/netaclaw/service/secret_crypto.ts packages/backend/src/entities.ts packages/backend/test/secret_crypto.test.ts packages/backend/test/entity_exports.test.ts packages/backend/test/data_source_projection.test.ts
git commit -m "feat: add mysql data source entities and secret crypto"
```

---

## Task 2: SQL Guard

**Files:**

- Create: `packages/backend/src/modules/netaclaw/service/mysql_query.ts`
- Test: `packages/backend/test/mysql_sql_guard.test.ts`

- [ ] **Step 1: Write failing SQL guard tests**

Add `packages/backend/test/mysql_sql_guard.test.ts`:

```ts
import { validateMysqlReadOnlySql } from '../src/modules/netaclaw/service/mysql_query.js';

describe('validateMysqlReadOnlySql', () => {
  const baseOptions = {
    allowedTables: ['orders', 'customers'],
    blockedTables: [],
    maxJoinTables: 4,
    maxRows: 200,
    maskedColumns: {},
  };

  it('accepts a SELECT with an explicit JOIN condition', () => {
    expect(validateMysqlReadOnlySql(
      'SELECT c.name, COUNT(*) FROM customers c JOIN orders o ON o.customer_id = c.id GROUP BY c.name LIMIT 20',
      baseOptions
    )).toEqual(expect.objectContaining({
      ok: true,
      tables: expect.arrayContaining(['customers', 'orders']),
    }));
  });

  it.each([
    'SHOW TABLES',
    'DESCRIBE orders',
    'EXPLAIN SELECT * FROM orders',
    'UPDATE orders SET amount = 1',
    'DELETE FROM orders',
    'DROP TABLE orders',
    'SELECT * FROM orders; SELECT * FROM customers',
    'SELECT * FROM orders -- comment',
    'SELECT * FROM orders # comment',
    'SELECT * FROM orders /* comment */',
    'WITH recent AS (SELECT * FROM orders) SELECT * FROM recent',
    'SELECT * FROM orders UNION SELECT * FROM customers',
    'SELECT @x := id FROM orders',
    'SELECT SLEEP(1)',
    'SELECT * INTO OUTFILE "/tmp/a" FROM orders',
  ])('rejects unsafe SQL: %s', sql => {
    const result = validateMysqlReadOnlySql(sql, baseOptions);
    expect(result.ok).toBe(false);
  });

  it('rejects unallowed tables', () => {
    const result = validateMysqlReadOnlySql('SELECT * FROM payments LIMIT 10', baseOptions);

    expect(result).toEqual(expect.objectContaining({
      ok: false,
      reason: 'table_not_allowed',
    }));
  });

  it('rejects blocked tables even when allowed', () => {
    const result = validateMysqlReadOnlySql('SELECT * FROM orders LIMIT 10', {
      allowedTables: ['orders'],
      blockedTables: ['orders'],
      maxJoinTables: 4,
    });

    expect(result).toEqual(expect.objectContaining({
      ok: false,
      reason: 'table_blocked',
    }));
  });

  it('rejects explicit joins without ON or USING', () => {
    const result = validateMysqlReadOnlySql(
      'SELECT * FROM customers JOIN orders LIMIT 10',
      baseOptions
    );

    expect(result).toEqual(expect.objectContaining({
      ok: false,
      reason: 'join_condition_required',
    }));
  });

  it('rejects more joined tables than allowed', () => {
    const result = validateMysqlReadOnlySql(
      'SELECT * FROM a JOIN b ON b.a_id = a.id JOIN c ON c.b_id = b.id LIMIT 10',
      { allowedTables: ['a', 'b', 'c'], blockedTables: [], maxJoinTables: 2, maxRows: 200, maskedColumns: {} }
    );

    expect(result).toEqual(expect.objectContaining({
      ok: false,
      reason: 'too_many_join_tables',
    }));
  });

  it('rejects a later JOIN that lacks its own ON or USING clause', () => {
    const result = validateMysqlReadOnlySql(
      'SELECT * FROM customers c JOIN orders o ON o.customer_id = c.id JOIN invoices i LIMIT 10',
      { allowedTables: ['customers', 'orders', 'invoices'], blockedTables: [], maxJoinTables: 4, maxRows: 200, maskedColumns: {} }
    );

    expect(result).toEqual(expect.objectContaining({
      ok: false,
      reason: 'join_condition_required',
    }));
  });

  it('rejects explicit LIMIT values above maxRows', () => {
    const result = validateMysqlReadOnlySql('SELECT * FROM orders LIMIT 10000', baseOptions);

    expect(result).toEqual(expect.objectContaining({
      ok: false,
      reason: 'limit_exceeds_max_rows',
    }));
  });

  it('rejects SQL that references masked columns', () => {
    const result = validateMysqlReadOnlySql('SELECT email AS e FROM customers LIMIT 10', {
      allowedTables: ['customers'],
      blockedTables: [],
      maxJoinTables: 4,
      maxRows: 200,
      maskedColumns: { 'customers.email': 'redact' },
    });

    expect(result).toEqual(expect.objectContaining({
      ok: false,
      reason: 'masked_column_denied',
    }));
  });
});
```

- [ ] **Step 2: Run test to verify RED**

Run:

```bash
pnpm --filter @neta/backend test -- --runInBand test/mysql_sql_guard.test.ts
```

Expected: FAIL because `mysql_query.ts` does not export the guard.

- [ ] **Step 3: Implement SQL guard**

Create the initial `packages/backend/src/modules/netaclaw/service/mysql_query.ts` with these exports:

```ts
export type MysqlSqlGuardOptions = {
  allowedTables?: string[];
  blockedTables?: string[];
  maxJoinTables?: number;
  maxRows?: number;
  maskedColumns?: Record<string, 'hash' | 'partial' | 'redact'>;
};

export type MysqlSqlGuardResult =
  | { ok: true; tables: string[]; limitedSql: string }
  | { ok: false; reason: string; tables?: string[] };

const DENIED_PATTERNS: Array<[RegExp, string]> = [
  [/;/, 'multiple_statements_denied'],
  [/(--|#|\/\*|\*\/)/, 'comments_denied'],
  [/^\s*(show|describe|desc|explain)\b/i, 'metadata_sql_denied'],
  [/^\s*(insert|update|delete|replace|create|alter|drop|truncate|call|grant|revoke|start|begin|commit|rollback)\b/i, 'non_select_denied'],
  [/\bunion\b/i, 'union_denied'],
  [/^\s*with\b/i, 'cte_denied'],
  [/@[a-zA-Z0-9_]+/, 'user_variable_denied'],
  [/\btemporary\b/i, 'temporary_table_denied'],
  [/\b(load_file|sleep|benchmark)\s*\(/i, 'dangerous_function_denied'],
  [/\binto\s+(out|dump)?file\b/i, 'file_output_denied'],
  [/\bfrom\s+[^,\s]+(?:\s+\w+)?\s*,\s*[^,\s]+/i, 'implicit_join_denied'],
];

function normalizeIdentifier(value: string): string {
  return value.replace(/`/g, '').split('.').pop()!.toLowerCase();
}

function stripBackticks(value: string): string {
  return value.replace(/`/g, '').toLowerCase();
}

export function extractMysqlTables(sql: string): string[] {
  const tables = new Set<string>();
  const tablePattern = /\b(?:from|join)\s+(`?[a-zA-Z0-9_]+`?(?:\.`?[a-zA-Z0-9_]+`?)?)/gi;
  let match: RegExpExecArray | null;
  while ((match = tablePattern.exec(sql))) {
    tables.add(normalizeIdentifier(match[1]));
  }
  return [...tables];
}

export function validateMysqlReadOnlySql(sql: string, options: MysqlSqlGuardOptions = {}): MysqlSqlGuardResult {
  const trimmed = String(sql || '').trim();
  if (!trimmed) return { ok: false, reason: 'sql_empty' };
  if (!/^\s*select\b/i.test(trimmed)) return { ok: false, reason: 'only_select_allowed' };

  for (const [pattern, reason] of DENIED_PATTERNS) {
    if (pattern.test(trimmed)) return { ok: false, reason };
  }

  if (!everyJoinHasCondition(trimmed)) {
    return { ok: false, reason: 'join_condition_required' };
  }

  const tables = extractMysqlTables(trimmed);
  const allowed = new Set((options.allowedTables || []).map(item => normalizeIdentifier(item)));
  const blocked = new Set((options.blockedTables || []).map(item => normalizeIdentifier(item)));

  for (const table of tables) {
    if (blocked.has(table)) return { ok: false, reason: 'table_blocked', tables };
    if (allowed.size === 0 || !allowed.has(table)) return { ok: false, reason: 'table_not_allowed', tables };
  }

  const maxJoinTables = Math.max(1, Math.min(options.maxJoinTables ?? 4, 6));
  if (tables.length > maxJoinTables) {
    return { ok: false, reason: 'too_many_join_tables', tables };
  }

  const maxRows = Math.max(1, Math.min(options.maxRows ?? 200, 1000));
  const explicitLimit = /\blimit\s+(\d+)/i.exec(trimmed);
  if (explicitLimit && Number(explicitLimit[1]) > maxRows) {
    return { ok: false, reason: 'limit_exceeds_max_rows', tables };
  }

  const normalizedSql = stripBackticks(trimmed);
  for (const key of Object.keys(options.maskedColumns || {})) {
    const [table, column] = key.toLowerCase().split('.');
    if (tables.includes(table) && new RegExp(`\\b${column}\\b`, 'i').test(normalizedSql)) {
      return { ok: false, reason: 'masked_column_denied', tables };
    }
  }

  const limitedSql = /\blimit\s+\d+/i.test(trimmed)
    ? trimmed
    : `SELECT * FROM (${trimmed}) AS neta_limited_query LIMIT ?`;

  return { ok: true, tables, limitedSql };
}

function everyJoinHasCondition(sql: string): boolean {
  const joinPattern = /\bjoin\b/gi;
  let match: RegExpExecArray | null;
  while ((match = joinPattern.exec(sql))) {
    const rest = sql.slice(match.index + match[0].length);
    const boundary = /\b(join|where|group\s+by|order\s+by|limit)\b/i.exec(rest);
    const segment = boundary ? rest.slice(0, boundary.index) : rest;
    if (!/\b(on|using)\b/i.test(segment)) return false;
  }
  return true;
}
```

- [ ] **Step 4: Run test to verify GREEN**

Run:

```bash
pnpm --filter @neta/backend test -- --runInBand test/mysql_sql_guard.test.ts
```

Expected: PASS.

- [ ] **Step 5: Commit**

```bash
git add packages/backend/src/modules/netaclaw/service/mysql_query.ts packages/backend/test/mysql_sql_guard.test.ts
git commit -m "feat: add mysql read only sql guard"
```

---

## Task 3: Data Source Service And Pool Manager

**Files:**

- Create: `packages/backend/src/modules/netaclaw/service/data_source.ts`
- Create: `packages/backend/src/modules/netaclaw/service/mysql_pool.ts`
- Modify: `packages/backend/src/modules/netaclaw/service/mysql_query.ts`
- Test: `packages/backend/test/mysql_pool_manager.test.ts`

- [ ] **Step 1: Write failing pool manager tests**

Add `packages/backend/test/mysql_pool_manager.test.ts`:

```ts
import { MysqlPoolManager } from '../src/modules/netaclaw/service/mysql_pool.js';

describe('MysqlPoolManager', () => {
  it('reuses pools by data source id and clamps connectionLimit', async () => {
    const created: any[] = [];
    const createPool = jest.fn((config: any) => {
      const pool = { config, end: jest.fn().mockResolvedValue(undefined) };
      created.push(pool);
      return pool;
    });
    const manager = new MysqlPoolManager(createPool as any);
    const source: any = {
      id: 7,
      host: 'localhost',
      port: 3306,
      database: 'sales',
      username: 'root',
      passwordEncrypted: 'encrypted',
      extra: { poolConnectionLimit: 20, connectTimeout: 1000 },
    };

    manager.secretCrypto = { decryptText: jest.fn().mockReturnValue('plain') } as any;

    const first = await manager.getPool(source);
    const second = await manager.getPool(source);

    expect(first).toBe(second);
    expect(createPool).toHaveBeenCalledTimes(1);
    expect(created[0].config.connectionLimit).toBe(10);
    expect(created[0].config.password).toBe('plain');
  });

  it('closes and forgets a cached pool', async () => {
    const end = jest.fn().mockResolvedValue(undefined);
    const createPool = jest.fn(() => ({ end }));
    const manager = new MysqlPoolManager(createPool as any);
    manager.secretCrypto = { decryptText: jest.fn().mockReturnValue('plain') } as any;

    await manager.getPool({ id: 1, host: 'h', port: 3306, database: 'd', username: 'u', passwordEncrypted: 'p', extra: {} } as any);
    await manager.closePool(1);

    expect(end).toHaveBeenCalledTimes(1);
  });
});
```

- [ ] **Step 2: Run test to verify RED**

Run:

```bash
pnpm --filter @neta/backend test -- --runInBand test/mysql_pool_manager.test.ts
```

Expected: FAIL because `mysql_pool.ts` does not exist.

- [ ] **Step 3: Implement pool manager**

Create `packages/backend/src/modules/netaclaw/service/mysql_pool.ts`:

```ts
import { Inject, Provide, Scope, ScopeEnum } from '@midwayjs/core';
import mysql from 'mysql2/promise';
import type { Pool } from 'mysql2/promise';
import { NetaClawDataSourceEntity } from '../entity/data_source.js';
import { SecretCryptoService } from './secret_crypto.js';

type CreatePool = typeof mysql.createPool;

@Provide()
@Scope(ScopeEnum.Singleton)
export class MysqlPoolManager {
  @Inject()
  secretCrypto: SecretCryptoService;

  private pools = new Map<number, Pool>();
  private readonly createPool: CreatePool;

  constructor(createPool: CreatePool = mysql.createPool) {
    this.createPool = createPool;
  }

  async getPool(source: NetaClawDataSourceEntity): Promise<Pool> {
    const existing = this.pools.get(source.id);
    if (existing) return existing;

    const limit = Math.max(1, Math.min(Number(source.extra?.poolConnectionLimit ?? 3), 10));
    const pool = this.createPool({
      host: source.host,
      port: source.port || 3306,
      database: source.database,
      user: source.username,
      password: source.passwordEncrypted ? this.secretCrypto.decryptText(source.passwordEncrypted) : undefined,
      waitForConnections: true,
      connectionLimit: limit,
      connectTimeout: Math.max(1000, Number(source.extra?.connectTimeout ?? 10000)),
      ssl: source.extra?.ssl as any,
    });
    this.pools.set(source.id, pool);
    return pool;
  }

  async closePool(sourceId: number): Promise<void> {
    const pool = this.pools.get(sourceId);
    if (!pool) return;
    this.pools.delete(sourceId);
    await pool.end();
  }

  createTransientPool(source: NetaClawDataSourceEntity): Pool {
    const limit = Math.max(1, Math.min(Number(source.extra?.poolConnectionLimit ?? 1), 3));
    return this.createPool({
      host: source.host,
      port: source.port || 3306,
      database: source.database,
      user: source.username,
      password: source.passwordEncrypted ? this.secretCrypto.decryptText(source.passwordEncrypted) : undefined,
      waitForConnections: true,
      connectionLimit: limit,
      connectTimeout: Math.max(1000, Number(source.extra?.connectTimeout ?? 10000)),
      ssl: source.extra?.ssl as any,
    });
  }
}
```

- [ ] **Step 4: Run pool tests to verify GREEN**

Run:

```bash
pnpm --filter @neta/backend test -- --runInBand test/mysql_pool_manager.test.ts
```

Expected: PASS.

- [ ] **Step 5: Implement `DataSourceService`**

Create `packages/backend/src/modules/netaclaw/service/data_source.ts`:

```ts
import { Inject, Provide, Scope, ScopeEnum } from '@midwayjs/core';
import { InjectEntityModel } from '@midwayjs/typeorm';
import { Repository } from 'typeorm';
import { NetaClawDataSourceEntity, NetaClawDataSourceExtra } from '../entity/data_source.js';
import { SecretCryptoService } from './secret_crypto.js';
import { MysqlPoolManager } from './mysql_pool.js';

export type AdminSafeDataSource = Omit<NetaClawDataSourceEntity, 'passwordEncrypted'> & {
  hasPassword: boolean;
};

export type AgentDataSourceSummary = Pick<NetaClawDataSourceEntity, 'name' | 'label' | 'database' | 'status'>;

export type SaveDataSourceInput = Partial<NetaClawDataSourceEntity> & {
  password?: string;
};

@Provide()
@Scope(ScopeEnum.Singleton)
export class NetaClawDataSourceService {
  @InjectEntityModel(NetaClawDataSourceEntity)
  repo: Repository<NetaClawDataSourceEntity>;

  @Inject()
  secretCrypto: SecretCryptoService;

  @Inject()
  mysqlPoolManager: MysqlPoolManager;

  toAdminSafe(entity: NetaClawDataSourceEntity): AdminSafeDataSource {
    const { passwordEncrypted, ...rest } = entity as any;
    return { ...rest, hasPassword: !!passwordEncrypted };
  }

  toAgentSummary(entity: NetaClawDataSourceEntity): AgentDataSourceSummary {
    return {
      name: entity.name,
      label: entity.label,
      database: entity.database,
      status: entity.status,
    };
  }

  async listAdminSafe(): Promise<AdminSafeDataSource[]> {
    const rows = await this.repo.find({ order: { id: 'DESC' } as any });
    return rows.map(row => this.toAdminSafe(row));
  }

  async listForAgent(agentId: number): Promise<AgentDataSourceSummary[]> {
    const rows = await this.repo.find({ where: { type: 'mysql', status: 1 } as any, order: { id: 'DESC' } as any });
    return rows
      .filter(row => Array.isArray(row.allowedAgentIds) && row.allowedAgentIds.includes(agentId))
      .map(row => this.toAgentSummary(row));
  }

  async getAuthorizedSource(name: string, agentId: number): Promise<NetaClawDataSourceEntity> {
    const source = await this.repo.findOne({ where: { name, type: 'mysql', status: 1 } as any });
    if (!source) throw new Error('data_source_not_found');
    if (!Array.isArray(source.allowedAgentIds) || !source.allowedAgentIds.includes(agentId)) {
      throw new Error('data_source_not_authorized');
    }
    return source;
  }

  normalizeExtra(extra?: NetaClawDataSourceExtra | null): NetaClawDataSourceExtra {
    return {
      allowedTables: extra?.allowedTables ?? [],
      blockedTables: extra?.blockedTables ?? [],
      maskedColumns: extra?.maskedColumns ?? {},
      schemaVisibility: extra?.schemaVisibility ?? 'allowed-only',
      queryTimeoutMs: Math.max(1000, Math.min(Number(extra?.queryTimeoutMs ?? 10000), 30000)),
      maxRows: Math.max(1, Math.min(Number(extra?.maxRows ?? 200), 1000)),
      maxJoinTables: Math.max(1, Math.min(Number(extra?.maxJoinTables ?? 4), 6)),
      poolConnectionLimit: Math.max(1, Math.min(Number(extra?.poolConnectionLimit ?? 3), 10)),
      ssl: extra?.ssl,
      connectTimeout: extra?.connectTimeout,
    };
  }

  async saveConfig(input: SaveDataSourceInput): Promise<AdminSafeDataSource> {
    const existing = input.id ? await this.repo.findOneBy({ id: input.id }) : null;
    const entity = this.repo.create({
      ...(existing || {}),
      ...input,
      type: 'mysql',
      readonly: 1,
      status: input.status ?? existing?.status ?? 1,
      port: Number(input.port ?? existing?.port ?? 3306),
      allowedAgentIds: Array.isArray(input.allowedAgentIds) ? input.allowedAgentIds : existing?.allowedAgentIds ?? [],
      extra: this.normalizeExtra(input.extra),
    });
    if (input.password) {
      entity.passwordEncrypted = this.secretCrypto.encryptText(input.password);
    }
    const saved = await this.repo.save(entity);
    await this.mysqlPoolManager.closePool(saved.id);
    return this.toAdminSafe(saved);
  }

  async delete(id: number): Promise<void> {
    await this.repo.delete(id);
    await this.mysqlPoolManager.closePool(id);
  }

  async testConnection(input: SaveDataSourceInput): Promise<{ ok: boolean; error?: string }> {
    const existing = input.id ? await this.repo.findOneBy({ id: input.id }) : null;
    const source = this.repo.create({
      ...(existing || {}),
      ...input,
      id: input.id ?? -1,
      type: 'mysql',
      readonly: 1,
      port: Number(input.port ?? 3306),
      allowedAgentIds: Array.isArray(input.allowedAgentIds) ? input.allowedAgentIds : [],
      extra: this.normalizeExtra(input.extra),
      passwordEncrypted: input.password
        ? this.secretCrypto.encryptText(input.password)
        : input.passwordEncrypted ?? existing?.passwordEncrypted,
    } as any);
    let pool: any;
    try {
      pool = this.mysqlPoolManager.createTransientPool(source);
      await pool.query('SELECT 1 AS ok');
      return { ok: true };
    } catch (err: any) {
      return { ok: false, error: sanitizeMysqlConnectionError(err) };
    } finally {
      if (pool) await pool.end();
    }
  }
}

function sanitizeMysqlConnectionError(err: any): string {
  return err?.code || err?.sqlState || err?.errno ? `mysql_connection_failed:${err.code || err.sqlState || err.errno}` : 'mysql_connection_failed';
}
```

- [ ] **Step 6: Run focused tests**

Run:

```bash
pnpm --filter @neta/backend test -- --runInBand test/mysql_pool_manager.test.ts test/secret_crypto.test.ts
```

Expected: PASS.

- [ ] **Step 7: Commit**

```bash
git add packages/backend/src/modules/netaclaw/service/data_source.ts packages/backend/src/modules/netaclaw/service/mysql_pool.ts packages/backend/test/mysql_pool_manager.test.ts
git commit -m "feat: add mysql data source services"
```

---

## Task 4: Schema Mapping And Query Execution

**Files:**

- Create: `packages/backend/src/modules/netaclaw/service/mysql_schema.ts`
- Modify: `packages/backend/src/modules/netaclaw/service/mysql_query.ts`
- Test: `packages/backend/test/mysql_schema_mapper.test.ts`
- Test: `packages/backend/test/mysql_query_service.test.ts`

- [ ] **Step 1: Write failing schema mapper test**

Add `packages/backend/test/mysql_schema_mapper.test.ts`:

```ts
import { mapMysqlSchemaRows } from '../src/modules/netaclaw/service/mysql_schema.js';

describe('mapMysqlSchemaRows', () => {
  it('maps columns, primary keys, indexes and foreign keys', () => {
    const result = mapMysqlSchemaRows({
      columns: [
        { tableName: 'orders', tableComment: '订单', columnName: 'id', columnType: 'int', isNullable: 'NO', columnKey: 'PRI', columnComment: 'ID' },
        { tableName: 'orders', tableComment: '订单', columnName: 'customer_id', columnType: 'int', isNullable: 'NO', columnKey: 'MUL', columnComment: '客户' },
      ],
      indexes: [
        { tableName: 'orders', indexName: 'PRIMARY', columnName: 'id', nonUnique: 0, seqInIndex: 1 },
        { tableName: 'orders', indexName: 'idx_customer', columnName: 'customer_id', nonUnique: 1, seqInIndex: 1 },
      ],
      foreignKeys: [
        { tableName: 'orders', columnName: 'customer_id', constraintName: 'fk_orders_customer', referencedTableName: 'customers', referencedColumnName: 'id' },
      ],
      maskedColumns: { 'orders.customer_id': 'redact' },
    });

    expect(result.tables).toEqual([
      expect.objectContaining({
        name: 'orders',
        comment: '订单',
        primaryKey: ['id'],
        columns: [
          expect.objectContaining({ name: 'id', type: 'int', nullable: false, key: 'PRI', masked: false }),
          expect.objectContaining({ name: 'customer_id', type: 'int', nullable: false, key: 'MUL', masked: true, maskMode: 'redact' }),
        ],
        indexes: expect.arrayContaining([
          expect.objectContaining({ name: 'idx_customer', columns: ['customer_id'], unique: false }),
        ]),
        foreignKeys: [
          expect.objectContaining({ column: 'customer_id', referencedTable: 'customers', referencedColumn: 'id' }),
        ],
      }),
    ]);
  });
});
```

- [ ] **Step 2: Run schema mapper test to verify RED**

Run:

```bash
pnpm --filter @neta/backend test -- --runInBand test/mysql_schema_mapper.test.ts
```

Expected: FAIL because `mysql_schema.ts` does not exist.

- [ ] **Step 3: Implement schema mapper and service skeleton**

Create `packages/backend/src/modules/netaclaw/service/mysql_schema.ts` with:

```ts
import { Inject, Provide, Scope, ScopeEnum } from '@midwayjs/core';
import { NetaClawDataSourceEntity } from '../entity/data_source.js';
import { MysqlPoolManager } from './mysql_pool.js';

type MaskMode = 'hash' | 'partial' | 'redact';

export interface MysqlSchemaRows {
  columns: Array<{ tableName: string; tableComment?: string; columnName: string; columnType: string; isNullable: string; columnKey?: string; columnComment?: string }>;
  indexes: Array<{ tableName: string; indexName: string; columnName: string; nonUnique: number; seqInIndex: number }>;
  foreignKeys: Array<{ tableName: string; columnName: string; constraintName: string; referencedTableName: string; referencedColumnName: string }>;
  maskedColumns?: Record<string, MaskMode>;
}

export function mapMysqlSchemaRows(rows: MysqlSchemaRows) {
  const tableMap = new Map<string, any>();
  const getTable = (name: string, comment?: string) => {
    if (!tableMap.has(name)) {
      tableMap.set(name, { name, comment: comment || '', columns: [], primaryKey: [], indexes: [], foreignKeys: [] });
    }
    return tableMap.get(name);
  };

  for (const column of rows.columns) {
    const table = getTable(column.tableName, column.tableComment);
    const maskKey = `${column.tableName}.${column.columnName}`;
    const maskMode = rows.maskedColumns?.[maskKey];
    if (column.columnKey === 'PRI') table.primaryKey.push(column.columnName);
    table.columns.push({
      name: column.columnName,
      type: column.columnType,
      nullable: column.isNullable === 'YES',
      key: column.columnKey || '',
      comment: column.columnComment || '',
      masked: !!maskMode,
      maskMode: maskMode || null,
    });
  }

  const indexGroups = new Map<string, any>();
  for (const index of rows.indexes) {
    const key = `${index.tableName}.${index.indexName}`;
    if (!indexGroups.has(key)) {
      indexGroups.set(key, { tableName: index.tableName, name: index.indexName, unique: index.nonUnique === 0, columns: [] as string[] });
    }
    indexGroups.get(key).columns[index.seqInIndex - 1] = index.columnName;
  }
  for (const index of indexGroups.values()) {
    getTable(index.tableName).indexes.push({ name: index.name, unique: index.unique, columns: index.columns.filter(Boolean) });
  }

  for (const fk of rows.foreignKeys) {
    getTable(fk.tableName).foreignKeys.push({
      name: fk.constraintName,
      column: fk.columnName,
      referencedTable: fk.referencedTableName,
      referencedColumn: fk.referencedColumnName,
    });
  }

  return { tables: [...tableMap.values()] };
}

@Provide()
@Scope(ScopeEnum.Singleton)
export class MysqlIntrospectionService {
  @Inject()
  poolManager: MysqlPoolManager;

  async listSchema(source: NetaClawDataSourceEntity, options: { tables?: string[] } = {}) {
    const allowedTables = source.extra?.allowedTables || [];
    const requested = options.tables?.length ? options.tables : allowedTables;
    const visibleTables = requested.filter(table => allowedTables.includes(table) && !(source.extra?.blockedTables || []).includes(table));
    if (visibleTables.length === 0) return { tables: [] };

    const pool = await this.poolManager.getPool(source);
    const placeholders = visibleTables.map(() => '?').join(',');
    const [columns] = await pool.query(
      `SELECT TABLE_NAME AS tableName, '' AS tableComment, COLUMN_NAME AS columnName, COLUMN_TYPE AS columnType, IS_NULLABLE AS isNullable, COLUMN_KEY AS columnKey, COLUMN_COMMENT AS columnComment
       FROM information_schema.COLUMNS
       WHERE TABLE_SCHEMA = ? AND TABLE_NAME IN (${placeholders})
       ORDER BY TABLE_NAME, ORDINAL_POSITION`,
      [source.database, ...visibleTables]
    );
    const [indexes] = await pool.query(
      `SELECT TABLE_NAME AS tableName, INDEX_NAME AS indexName, COLUMN_NAME AS columnName, NON_UNIQUE AS nonUnique, SEQ_IN_INDEX AS seqInIndex
       FROM information_schema.STATISTICS
       WHERE TABLE_SCHEMA = ? AND TABLE_NAME IN (${placeholders})
       ORDER BY TABLE_NAME, INDEX_NAME, SEQ_IN_INDEX`,
      [source.database, ...visibleTables]
    );
    const [foreignKeys] = await pool.query(
      `SELECT TABLE_NAME AS tableName, COLUMN_NAME AS columnName, CONSTRAINT_NAME AS constraintName, REFERENCED_TABLE_NAME AS referencedTableName, REFERENCED_COLUMN_NAME AS referencedColumnName
       FROM information_schema.KEY_COLUMN_USAGE
       WHERE TABLE_SCHEMA = ? AND TABLE_NAME IN (${placeholders}) AND REFERENCED_TABLE_NAME IS NOT NULL`,
      [source.database, ...visibleTables]
    );
    return mapMysqlSchemaRows({
      columns: columns as MysqlSchemaRows['columns'],
      indexes: indexes as MysqlSchemaRows['indexes'],
      foreignKeys: foreignKeys as MysqlSchemaRows['foreignKeys'],
      maskedColumns: source.extra?.maskedColumns,
    });
  }

  async sampleTable(source: NetaClawDataSourceEntity, options: { table: string; columns?: string[]; limit?: number }) {
    const allowedTables = source.extra?.allowedTables || [];
    const blockedTables = source.extra?.blockedTables || [];
    if (!allowedTables.includes(options.table) || blockedTables.includes(options.table)) {
      throw new Error('table_not_allowed');
    }
    const schema = await this.listSchema(source, { tables: [options.table] });
    const table = schema.tables[0];
    if (!table) throw new Error('table_not_found');
    const visibleColumns = new Set(table.columns.map((column: any) => column.name));
    const requestedColumns = options.columns?.length ? options.columns : table.columns.map((column: any) => column.name);
    for (const column of requestedColumns) {
      if (!/^[A-Za-z0-9_]+$/.test(column) || !visibleColumns.has(column)) {
        throw new Error('column_not_allowed');
      }
      if (source.extra?.maskedColumns?.[`${options.table}.${column}`]) {
        throw new Error('masked_column_denied');
      }
    }
    const safeTable = `\`${options.table.replace(/`/g, '')}\``;
    const safeColumns = requestedColumns.map(column => `\`${column.replace(/`/g, '')}\``).join(', ');
    const limit = Math.max(1, Math.min(Number(options.limit ?? 5), 20));
    const pool = await this.poolManager.getPool(source);
    const [rows, fields] = await pool.query(`SELECT ${safeColumns} FROM ${safeTable} LIMIT ?`, [limit]);
    return {
      columns: Array.isArray(fields) ? fields.map((field: any) => field.name) : requestedColumns,
      rows: Array.isArray(rows) ? rows : [],
      rowCount: Array.isArray(rows) ? rows.length : 0,
      truncated: false,
    };
  }
}
```

- [ ] **Step 4: Run schema mapper test to verify GREEN**

Run:

```bash
pnpm --filter @neta/backend test -- --runInBand test/mysql_schema_mapper.test.ts
```

Expected: PASS.

- [ ] **Step 5: Write failing query service test**

Add `packages/backend/test/mysql_query_service.test.ts`:

```ts
import { MysqlQueryService } from '../src/modules/netaclaw/service/mysql_query.js';

describe('MysqlQueryService', () => {
  it('executes a guarded query and writes audit', async () => {
    const query = jest.fn().mockResolvedValue([
      [
        { id: 1, total: 99 },
      ],
      [{ name: 'id' }, { name: 'total' }],
    ]);
    const service = new MysqlQueryService();
    service.poolManager = { getPool: jest.fn().mockResolvedValue({ query }) } as any;
    service.auditRepo = { save: jest.fn().mockResolvedValue({}) } as any;

    const result = await service.executeReadOnly({
      source: {
        id: 1,
        database: 'sales',
        extra: {
          allowedTables: ['orders'],
          blockedTables: [],
          maskedColumns: {},
          maxRows: 200,
          maxJoinTables: 4,
          queryTimeoutMs: 10000,
        },
      } as any,
      sql: 'SELECT id, total FROM orders LIMIT 10',
      agentId: 3,
      userId: 5,
      toolCallId: 'tool-1',
    });

    expect(result.rows).toEqual([{ id: 1, total: 99 }]);
    expect(result.rowCount).toBe(1);
    expect(service.auditRepo.save).toHaveBeenCalledWith(expect.objectContaining({
      dataSourceId: 1,
      agentId: 3,
      userId: 5,
      toolCallId: 'tool-1',
      status: 'success',
      rowCount: 1,
    }));
  });

  it('rejects masked column references before querying mysql', async () => {
    const service = new MysqlQueryService();
    service.poolManager = { getPool: jest.fn() } as any;
    service.auditRepo = { save: jest.fn().mockResolvedValue({}) } as any;

    await expect(service.executeReadOnly({
      source: {
        id: 1,
        extra: {
          allowedTables: ['orders'],
          blockedTables: [],
          maskedColumns: { 'orders.email': 'redact' },
          maxRows: 200,
          maxJoinTables: 4,
        },
      } as any,
      sql: 'SELECT email AS e FROM orders LIMIT 10',
      agentId: 3,
      userId: 5,
      toolCallId: 'tool-masked',
    })).rejects.toThrow('mysql_sql_rejected');

    expect(service.poolManager.getPool).not.toHaveBeenCalled();
    expect(service.auditRepo.save).toHaveBeenCalledWith(expect.objectContaining({
      status: 'rejected',
      rejectReason: 'masked_column_denied',
    }));
  });

  it('audits rejected SQL', async () => {
    const service = new MysqlQueryService();
    service.poolManager = { getPool: jest.fn() } as any;
    service.auditRepo = { save: jest.fn().mockResolvedValue({}) } as any;

    await expect(service.executeReadOnly({
      source: { id: 1, extra: { allowedTables: ['orders'], blockedTables: [] } } as any,
      sql: 'DROP TABLE orders',
      agentId: 3,
      userId: 5,
      toolCallId: 'tool-2',
    })).rejects.toThrow('mysql_sql_rejected');

    expect(service.poolManager.getPool).not.toHaveBeenCalled();
    expect(service.auditRepo.save).toHaveBeenCalledWith(expect.objectContaining({
      status: 'rejected',
    }));
  });
});
```

- [ ] **Step 6: Run query service test to verify RED**

Run:

```bash
pnpm --filter @neta/backend test -- --runInBand test/mysql_query_service.test.ts
```

Expected: FAIL because `MysqlQueryService` is not implemented.

- [ ] **Step 7: Implement query service**

Extend `packages/backend/src/modules/netaclaw/service/mysql_query.ts` with:

```ts
import { Inject, Provide, Scope, ScopeEnum } from '@midwayjs/core';
import { InjectEntityModel } from '@midwayjs/typeorm';
import { Repository } from 'typeorm';
import * as crypto from 'crypto';
import { NetaClawDataSourceEntity } from '../entity/data_source.js';
import { NetaClawDataSourceQueryAuditEntity } from '../entity/data_source_query_audit.js';
import { MysqlPoolManager } from './mysql_pool.js';

export interface ExecuteReadOnlyParams {
  source: NetaClawDataSourceEntity;
  sql: string;
  agentId?: number;
  userId?: number;
  toolCallId?: string;
}

function hashSql(sql: string): string {
  return crypto.createHash('sha256').update(sql).digest('hex');
}

function sanitizeMysqlError(err: any): string {
  const code = err?.code || err?.sqlState || err?.errno;
  return code ? `mysql_query_failed:${code}` : 'mysql_query_failed';
}

@Provide()
@Scope(ScopeEnum.Singleton)
export class MysqlQueryService {
  @Inject()
  poolManager: MysqlPoolManager;

  @InjectEntityModel(NetaClawDataSourceQueryAuditEntity)
  auditRepo: Repository<NetaClawDataSourceQueryAuditEntity>;

  async executeReadOnly(params: ExecuteReadOnlyParams) {
    const start = Date.now();
    const extra = params.source.extra || {};
    const guard = validateMysqlReadOnlySql(params.sql, {
      allowedTables: extra.allowedTables || [],
      blockedTables: extra.blockedTables || [],
      maxJoinTables: extra.maxJoinTables || 4,
      maxRows: extra.maxRows || 200,
      maskedColumns: extra.maskedColumns || {},
    });

    if (!guard.ok) {
      await this.writeAudit(params, 'rejected', start, 0, guard.reason);
      throw new Error(`mysql_sql_rejected: ${guard.reason}`);
    }

    try {
      const pool = await this.poolManager.getPool(params.source);
      const maxRows = Math.max(1, Math.min(Number(extra.maxRows ?? 200), 1000));
      const appendedLimit = guard.limitedSql.includes('?');
      const sql = guard.limitedSql.includes('?') ? guard.limitedSql : guard.limitedSql;
      const sqlParams = appendedLimit ? [maxRows + 1] : [];
      const [rows, fields] = await pool.query({ sql, timeout: Math.max(1000, Math.min(Number(extra.queryTimeoutMs ?? 10000), 30000)) } as any, sqlParams);
      const fetchedRows = Array.isArray(rows) ? rows as any[] : [];
      const truncated = appendedLimit && fetchedRows.length > maxRows;
      const resultRows = truncated ? fetchedRows.slice(0, Math.min(fetchedRows.length, maxRows)) : fetchedRows;
      await this.writeAudit(params, 'success', start, resultRows.length, null);
      return {
        columns: Array.isArray(fields) ? fields.map((field: any) => field.name) : Object.keys(resultRows[0] || {}),
        rows: resultRows,
        rowCount: resultRows.length,
        truncated,
        elapsedMs: Date.now() - start,
        sql: params.sql,
      };
    } catch (err: any) {
      await this.writeAudit(params, 'failed', start, 0, null, err?.code || err?.errno || err?.sqlState);
      throw new Error(sanitizeMysqlError(err));
    }
  }

  private async writeAudit(
    params: ExecuteReadOnlyParams,
    status: 'success' | 'rejected' | 'failed',
    start: number,
    rowCount: number,
    rejectReason?: string | null,
    errorCode?: string | number | null
  ) {
    await this.auditRepo.save({
      dataSourceId: params.source.id,
      agentId: params.agentId ?? null,
      userId: params.userId ?? null,
      toolCallId: params.toolCallId ?? null,
      sqlHash: hashSql(params.sql),
      sqlPreview: params.sql.slice(0, 1000),
      status,
      rejectReason: rejectReason ?? null,
      elapsedMs: Date.now() - start,
      rowCount,
      errorCode: errorCode ? String(errorCode) : null,
    } as any);
  }
}
```

- [ ] **Step 8: Run schema/query tests**

Run:

```bash
pnpm --filter @neta/backend test -- --runInBand test/mysql_schema_mapper.test.ts test/mysql_query_service.test.ts test/mysql_sql_guard.test.ts
```

Expected: PASS.

- [ ] **Step 9: Commit**

```bash
git add packages/backend/src/modules/netaclaw/service/mysql_schema.ts packages/backend/src/modules/netaclaw/service/mysql_query.ts packages/backend/test/mysql_schema_mapper.test.ts packages/backend/test/mysql_query_service.test.ts
git commit -m "feat: add mysql schema and query services"
```

---

## Task 5: MySQL Agent Tools And Resolver Wiring

**Files:**

- Create: `packages/backend/src/modules/netaclaw/tools/builtin/mysql.ts`
- Modify: `packages/backend/src/modules/netaclaw/tools/catalog.ts`
- Modify: `packages/backend/src/modules/netaclaw/tools/manifest.ts`
- Modify: `packages/backend/src/modules/netaclaw/service/tool_resolver.ts`
- Test: `packages/backend/test/tool_resolver.test.ts`

- [ ] **Step 1: Write failing resolver tests**

Append to `packages/backend/test/tool_resolver.test.ts`:

```ts
  it('resolves mysql toolset tools when enabled for an agent', async () => {
    service.toolRegistry = {
      all: jest.fn().mockResolvedValue([
        { name: 'mysql_list_sources', status: 1, capability: 'text', promptHint: null, isCore: 0, toolset: 'mysql' },
        { name: 'mysql_schema', status: 1, capability: 'text', promptHint: null, isCore: 0, toolset: 'mysql' },
        { name: 'mysql_table_sample', status: 1, capability: 'text', promptHint: null, isCore: 0, toolset: 'mysql' },
        { name: 'mysql_query', status: 1, capability: 'text', promptHint: null, isCore: 0, toolset: 'mysql' },
      ]),
    } as any;
    service.skillLoader = {} as any;
    service.dataSourceService = { listForAgent: jest.fn().mockResolvedValue([]) } as any;
    service.mysqlIntrospection = {} as any;
    service.mysqlQuery = {} as any;

    const result = await service.resolve({
      agent: { id: 9, toolsets: ['mysql'], tools: { disabled: [] } } as any,
      modelCapability: 'text',
    });

    expect(result.toolNames).toEqual(['mysql_list_sources', 'mysql_schema', 'mysql_table_sample', 'mysql_query']);
    expect(result.tools.map(tool => tool.name)).toEqual(['mysql_list_sources', 'mysql_schema', 'mysql_table_sample', 'mysql_query']);
  });

  it('routes mysql tools through main process proxy for subagent workers', async () => {
    service.toolRegistry = {
      all: jest.fn().mockResolvedValue([
        { name: 'mysql_query', status: 1, capability: 'text', promptHint: null, isCore: 0, toolset: 'mysql' },
      ]),
    } as any;
    service.skillLoader = {} as any;
    service.dataSourceService = {} as any;
    service.mysqlIntrospection = {} as any;
    service.mysqlQuery = {} as any;

    const result = await service.resolve({
      agent: {
        tools: { inheritCoreTools: false, enabled: ['mysql_query'], disabled: [] },
      } as any,
      modelCapability: 'text',
      delegationRole: 'subagent',
    });

    expect(result.toolManifest).toEqual([
      expect.objectContaining({
        name: 'mysql_query',
        supportedInWorker: false,
        workerRoutingHint: 'main-process-proxy',
      }),
    ]);
  });
```

- [ ] **Step 2: Run resolver test to verify RED**

Run:

```bash
pnpm --filter @neta/backend test -- --runInBand test/tool_resolver.test.ts
```

Expected: FAIL because mysql tools are not registered or injected.

- [ ] **Step 3: Implement MySQL tools**

Create `packages/backend/src/modules/netaclaw/tools/builtin/mysql.ts`:

```ts
import { Type } from '@sinclair/typebox';
import { AgentToolWithMeta, jsonResult } from '../common.js';
import { registerSchema } from '../catalog.js';
import { NetaClawDataSourceService } from '../../service/data_source.js';
import { MysqlIntrospectionService } from '../../service/mysql_schema.js';
import { MysqlQueryService } from '../../service/mysql_query.js';

export const TOOLSET_MYSQL = 'mysql' as const;

const Runtime = Type.Optional(Type.Any());
const SourceParam = Type.Object({
  source: Type.String({ description: '数据源名称' }),
  _netaRuntime: Runtime,
});

function readAgentId(params: any): number {
  const id = params?._netaRuntime?.currentAgent?.id;
  if (!id) throw new Error('current_agent_id_missing');
  return Number(id);
}

function readUserId(params: any): number | undefined {
  const id = params?._netaRuntime?.userId;
  return id === undefined || id === null ? undefined : Number(id);
}

export function createMysqlListSourcesTool(deps: {
  dataSourceService: NetaClawDataSourceService;
}): AgentToolWithMeta<typeof Type.Object<any>, unknown> {
  const Params = Type.Object({ _netaRuntime: Runtime });
  return {
    name: 'mysql_list_sources',
    label: 'MySQL 数据源列表',
    description: '列出当前 Agent 授权可用的 MySQL 数据源，不返回连接密码。',
    parameters: Params,
    async execute(_id, params) {
      return jsonResult({ sources: await deps.dataSourceService.listForAgent(readAgentId(params)) });
    },
  } as any;
}

export function createMysqlSchemaTool(deps: {
  dataSourceService: NetaClawDataSourceService;
  mysqlIntrospection: MysqlIntrospectionService;
}): AgentToolWithMeta<any, unknown> {
  const Params = Type.Intersect([SourceParam, Type.Object({
    tables: Type.Optional(Type.Array(Type.String())),
  })]);
  return {
    name: 'mysql_schema',
    label: 'MySQL Schema',
    description: '读取授权 MySQL 表的字段、索引、主键和外键信息。',
    parameters: Params,
    async execute(_id, params) {
      const source = await deps.dataSourceService.getAuthorizedSource(params.source, readAgentId(params));
      return jsonResult(await deps.mysqlIntrospection.listSchema(source, { tables: params.tables }));
    },
  };
}

export function createMysqlTableSampleTool(deps: {
  dataSourceService: NetaClawDataSourceService;
  mysqlIntrospection: MysqlIntrospectionService;
}): AgentToolWithMeta<any, unknown> {
  const Params = Type.Intersect([SourceParam, Type.Object({
    table: Type.String(),
    columns: Type.Optional(Type.Array(Type.String())),
    limit: Type.Optional(Type.Number()),
  })]);
  return {
    name: 'mysql_table_sample',
    label: 'MySQL 表样例',
    description: '读取授权表的少量样例值，用于判断字段含义和 JOIN key。',
    parameters: Params,
    async execute(_id, params) {
      const source = await deps.dataSourceService.getAuthorizedSource(params.source, readAgentId(params));
      return jsonResult(await deps.mysqlIntrospection.sampleTable(source, {
        table: params.table,
        columns: params.columns,
        limit: params.limit,
      }));
    },
  };
}

export function createMysqlQueryTool(deps: {
  dataSourceService: NetaClawDataSourceService;
  mysqlQuery: MysqlQueryService;
}): AgentToolWithMeta<any, unknown> {
  const Params = Type.Intersect([SourceParam, Type.Object({
    sql: Type.String(),
  })]);
  return {
    name: 'mysql_query',
    label: 'MySQL 查询',
    description: '执行授权数据源上的只读 SELECT SQL。支持授权表范围内的跨表 JOIN。',
    parameters: Params,
    async execute(id, params) {
      const source = await deps.dataSourceService.getAuthorizedSource(params.source, readAgentId(params));
      return jsonResult(await deps.mysqlQuery.executeReadOnly({
        source,
        sql: params.sql,
        agentId: readAgentId(params),
        userId: readUserId(params),
        toolCallId: id,
      }));
    },
  };
}

for (const name of ['mysql_list_sources', 'mysql_schema', 'mysql_table_sample', 'mysql_query']) {
  registerSchema({
    name,
    toolset: TOOLSET_MYSQL,
    description: name,
    visibility: 'tool',
    capability: 'text',
    isCore: false,
    canDisable: true,
  });
}
```

- [ ] **Step 4: Wire catalog, manifest, resolver**

Modify `packages/backend/src/modules/netaclaw/tools/catalog.ts`:

```ts
export const TOOLSET_MYSQL = 'mysql' as const;
import './builtin/mysql.js';
```

Modify `packages/backend/src/modules/netaclaw/tools/manifest.ts`:

```ts
const MAIN_PROCESS_PROXY_TOOLS = new Set([
  'write_file',
  'edit',
  'patch',
  'read_skill',
  'read_skill_file',
  'skill_manage',
  'memory_save',
  'memory_recall',
  'mysql_list_sources',
  'mysql_schema',
  'mysql_table_sample',
  'mysql_query',
]);

export function toToolKind(name: string): SubagentToolManifestItem['kind'] {
  if (name.startsWith('mysql_')) return 'custom';
  if (name.startsWith('memory_')) return 'memory';
  if (name.startsWith('read_skill') || name === 'skill_manage') return 'skill';
  if (name === 'delegate_task' || name === 'delegate_parallel') return 'delegation';
  if (name === 'escalate') return 'admin';
  if (
    name === 'bash'
    || name === 'read_file'
    || name === 'write_file'
    || name === 'list_dir'
    || name === 'find_files'
    || name === 'grep'
    || name === 'edit'
    || name === 'patch'
    || name === 'clarify'
  ) {
    return 'builtin';
  }
  return 'custom';
}
```

Modify `packages/backend/src/modules/netaclaw/service/tool_resolver.ts`:

- Import services and factories:

```ts
import { NetaClawDataSourceService } from './data_source.js';
import { MysqlIntrospectionService } from './mysql_schema.js';
import { MysqlQueryService } from './mysql_query.js';
import {
  createMysqlListSourcesTool,
  createMysqlQueryTool,
  createMysqlSchemaTool,
  createMysqlTableSampleTool,
} from '../tools/builtin/mysql.js';
```

- Add injected services:

```ts
@Inject()
dataSourceService: NetaClawDataSourceService;

@Inject()
mysqlIntrospection: MysqlIntrospectionService;

@Inject()
mysqlQuery: MysqlQueryService;
```

- Add to `getBaseToolMap()`:

```ts
['mysql_list_sources', createMysqlListSourcesTool({ dataSourceService: this.dataSourceService })],
['mysql_schema', createMysqlSchemaTool({ dataSourceService: this.dataSourceService, mysqlIntrospection: this.mysqlIntrospection })],
['mysql_table_sample', createMysqlTableSampleTool({ dataSourceService: this.dataSourceService, mysqlIntrospection: this.mysqlIntrospection })],
['mysql_query', createMysqlQueryTool({ dataSourceService: this.dataSourceService, mysqlQuery: this.mysqlQuery })],
```

- [ ] **Step 5: Run resolver tests**

Run:

```bash
pnpm --filter @neta/backend test -- --runInBand test/tool_resolver.test.ts
```

Expected: PASS.

- [ ] **Step 6: Commit**

```bash
git add packages/backend/src/modules/netaclaw/tools/builtin/mysql.ts packages/backend/src/modules/netaclaw/tools/catalog.ts packages/backend/src/modules/netaclaw/tools/manifest.ts packages/backend/src/modules/netaclaw/service/tool_resolver.ts packages/backend/test/tool_resolver.test.ts
git commit -m "feat: register mysql agent tools"
```

---

## Task 6: Admin Controller And Prompt Skill

**Files:**

- Create: `packages/backend/src/modules/netaclaw/controller/admin/data_source.ts`
- Create: `packages/backend/skills/data-analyst-mysql/SKILL.md`

- [ ] **Step 1: Implement admin controller**

Create `packages/backend/src/modules/netaclaw/controller/admin/data_source.ts`:

```ts
import { Body, Controller, Get, Inject, Post, Provide, Query } from '@midwayjs/core';
import { NetaClawDataSourceService } from '../../service/data_source.js';

@Provide()
@Controller('/admin/netaclaw/data-source')
export class NetaClawDataSourceAdminController {
  @Inject()
  dataSourceService: NetaClawDataSourceService;

  @Get('/list')
  async list(@Query('agentId') agentId?: number) {
    if (agentId) {
      return { code: 1000, data: await this.dataSourceService.listForAgent(Number(agentId)) };
    }
    return { code: 1000, data: await this.dataSourceService.listAdminSafe() };
  }

  @Post('/save')
  async save(@Body() body: any) {
    return { code: 1000, data: await this.dataSourceService.saveConfig(body) };
  }

  @Post('/delete')
  async delete(@Body() body: { id: number }) {
    await this.dataSourceService.delete(Number(body.id));
    return { code: 1000, message: 'success' };
  }

  @Post('/test')
  async test(@Body() body: any) {
    const result = await this.dataSourceService.testConnection(body);
    return result.ok
      ? { code: 1000, data: result }
      : { code: 1001, message: result.error || 'connection_failed', data: result };
  }
}
```

- [ ] **Step 2: Add prompt skill**

Create `packages/backend/skills/data-analyst-mysql/SKILL.md`:

```md
---
name: data-analyst-mysql
description: 使用 MySQL 工具进行只读智能问数，支持 schema 分析、跨表 JOIN、口径澄清和 SQL 结果解释。
version: 1.0.0
metadata:
  skillType: llm
  tags: [mysql, data-analysis, question-answering]
  conditions:
    requires_tools: ["mysql_list_sources", "mysql_schema", "mysql_query"]
---

# MySQL 智能问数

你负责使用 MySQL 工具回答业务数据问题。所有数据库访问必须通过 `mysql_*` 工具完成。

## 工作流程

1. 先调用 `mysql_list_sources` 查看当前 Agent 可用的数据源。
2. 根据用户问题选择数据源；如果无法判断，先澄清。
3. 调用 `mysql_schema` 读取相关表结构、字段、主键、索引和外键。
4. 跨表 JOIN 时优先使用外键；没有外键时，根据字段名、类型、索引和 `mysql_table_sample` 的样例值推断关联。
5. 业务口径不明确时必须先澄清，不要硬猜。
6. 只调用 `mysql_query` 执行只读 SELECT。

## JOIN 规则

- JOIN 必须有明确的 `ON` 或 `USING` 条件。
- 不确定关联字段时，先解释拟采用的关联并询问用户。
- 若关联来自推断，最终回答必须说明“该关联为推断”。
- 查询超时或被拒绝时，缩小时间范围、减少表数量或向用户澄清。

## 回答格式

回答必须包含：

- 结论。
- SQL。
- 口径说明，特别是时间范围、筛选条件、JOIN 字段。
- 限制说明，例如 LIMIT、样例、脱敏列、推断关联或字段缺失。

不要暴露数据库 host、账号、密码、连接串或未授权表信息。
不要承诺“全量统计”，除非 SQL 口径和过滤范围明确。
```

- [ ] **Step 3: Run build or focused tests**

Run:

```bash
pnpm --filter @neta/backend test -- --runInBand test/tool_resolver.test.ts test/skill_env_schema.test.ts
```

Expected: PASS.

- [ ] **Step 4: Commit**

```bash
git add packages/backend/src/modules/netaclaw/controller/admin/data_source.ts packages/backend/skills/data-analyst-mysql/SKILL.md
git commit -m "feat: add mysql data source admin api and analyst skill"
```

---

## Task 7: Verification And Documentation Sync

**Files:**

- Modify: `docs/code-wiki/entities/tool-system.md`
- Modify: `docs/code-wiki/entities/skill-system.md`
- Modify: `docs/code-wiki/entities/netaclaw-module.md`
- Modify: `docs/code-wiki/comparisons/database-entity-overview.md`

- [ ] **Step 1: Update code wiki**

Add concise references:

- `tool-system.md`: add `mysql_list_sources`, `mysql_schema`, `mysql_table_sample`, `mysql_query` under a `MySQL 问数` toolset.
- `skill-system.md`: add `data-analyst-mysql` as a prompt skill.
- `netaclaw-module.md`: increment table count and list `netaclaw_data_source`, `netaclaw_data_source_query_audit`.
- `database-entity-overview.md`: add field summary for both new tables.

- [ ] **Step 2: Run all focused backend tests**

Run:

```bash
pnpm --filter @neta/backend test -- --runInBand test/secret_crypto.test.ts test/mysql_sql_guard.test.ts test/mysql_schema_mapper.test.ts test/mysql_query_service.test.ts test/mysql_pool_manager.test.ts test/tool_resolver.test.ts test/entity_exports.test.ts
```

Expected: PASS.

- [ ] **Step 3: Run backend build**

Run:

```bash
pnpm --filter @neta/backend build
```

Expected: exit code 0.

- [ ] **Step 4: Inspect final diff**

Run:

```bash
git status --short
git diff --stat
```

Expected: only MySQL question answering implementation files and docs are changed.

- [ ] **Step 5: Commit**

```bash
git add docs/code-wiki packages/backend
git commit -m "docs: update code wiki for mysql question answering"
```

---

## Final Verification Checklist

- [ ] SQL guard rejects `SHOW`, `DESCRIBE`, `EXPLAIN`, comments, semicolons, DML, DDL, `UNION`, CTE, user variables, dangerous functions, blocked tables, unallowed tables, missing JOIN condition, too many JOIN tables.
- [ ] Data source password is AES-GCM encrypted; admin APIs never return password ciphertext/plaintext, and `mysql_list_sources` never returns host, username, password, connection string, or permission details.
- [ ] `mysql_schema` returns only authorized table metadata and marks masked columns.
- [ ] `mysql_table_sample` validates table and column identifiers against schema plus allowlist/blocklist, and rejects masked columns.
- [ ] `mysql_query` rejects masked column references, explicit LIMIT above `maxRows`, missing per-JOIN `ON/USING`, blocked tables, and unallowed tables before querying MySQL.
- [ ] `mysql_query` writes audit for success, rejected, and failed SQL.
- [ ] MySQL tools are visible only when selected by `mysql` toolset or explicit tool config.
- [ ] Subagent worker route for MySQL tools is `main-process-proxy`, not `worker-local`.
- [ ] `data-analyst-mysql` skill is discoverable and condition-gated by MySQL tools.
- [ ] Focused tests pass.
- [ ] Backend build passes.