Imported from plurigrid/asi (skills/cq-ai/SKILL.md). Install upstream with npx skills add plurigrid/asi --skill cq-ai. Copyright stays with the author.

CQ-AI: Deterministic Code Security Scanning Skill

Version: 1.0.0 Status: Production Ready Trit Assignment: +1 (optimistic, generative - finds vulnerabilities) Principle: Deterministic Analysis (SPI guarantee)

Core Innovation

CQ-AI extends NCC Group's Code Query with:

1. Deterministic Code Analysis - Same seed + same code → identical findings
2. Ternary Polarity Classification - Critical/Medium/Info mapped to GF(3) = {-1, 0, +1}
3. Parallel Scanning - Split-stream architecture for independent analysis threads
4. Out-of-Order Proofs - Results composable regardless of scan order
5. MCP Integration - AI-guided configuration and finding prioritization

Architecture

CQ-AI System
├─ Layer 1: SplitMix64 Seeding (deterministic entropy source)
├─ Layer 2: Ternary Polarity Classification (GF(3) severity)
├─ Layer 3: Parallel Scan Distribution (work-stealing scheduler)
├─ Layer 4: MCP Server Integration (AI skill configuration)
└─ Layer 5: Finding Aggregation & Deduplication

SplitMix64 Seeding

All CQ-AI analysis is seeded with SplitMix64, providing:

• Deterministic output given fixed seed and input codebase
• Fast generation (CPU cache-friendly)
• No external entropy
• Reproducible findings across teams and time

Algorithm

struct SplitMix64 {
    state: u64,
}

impl SplitMix64 {
    fn new(seed: u64) -> Self {
        SplitMix64 { state: seed }
    }

    fn next_u64(&mut self) -> u64 {
        let z = (self.state ^ (self.state >> 30)) * 0xBF58476D1CE4E5B9;
        self.state = self.state.wrapping_add(0x9E3779B97F4A7C15);
        z ^ (z >> 27)
    }

    fn next_u32(&mut self) -> u32 {
        (self.next_u64() >> 32) as u32
    }
}

Constant: φ⁻¹ × 2⁶⁴ = 0x9E3779B97F4A7C15 (golden ratio increment)

This ensures mixing time ~2 rounds for uniform distribution across u64 space.

Seeding CQ Analysis

def cq_deterministic_scan(codebase_path: str, seed: int) -> List[Finding]:
    """
    Run CQ with deterministic ordering and findings.

    Args:
        codebase_path: Root directory of code to scan
        seed: SplitMix64 seed (same seed = same findings)

    Returns:
        List of findings in deterministic order
    """
    # Initialize seeded RNG
    rng = SplitMix64(seed)

    # Generate deterministic file traversal order
    file_order = sorted(
        get_all_files(codebase_path),
        key=lambda f: rng.next_u32()  # Deterministic shuffle
    )

    # Scan files in deterministic order
    findings = []
    for filepath in file_order:
        file_findings = cq_scan_file(filepath, seed)
        findings.extend(file_findings)

    # Sort findings deterministically
    return sorted(findings, key=lambda f: (f.file, f.line, f.finding_id))

Ternary Polarity Classification

Severity Mapping (GF(3))

CQ findings are classified using balanced ternary with 3 severity tiers mapped to GF(3):

Polarity as Interaction Direction

• +1 (Positive Trit): Generative findings - additions/detections to security posture
• 0 (Neutral Trit): Structural issues - existing problems requiring attention
• -1 (Negative Trit): Reductive findings - false positives, non-issues to ignore

Classification Algorithm

class CQFinding:
    file: str
    line: int
    finding_id: str
    description: str

    @property
    def severity_trit(self) -> int:
        """
        Classify finding to GF(3) trit based on characteristics.
        Returns: -1 (INFO), 0 (MEDIUM), +1 (CRITICAL)
        """
        # Rule-based classification
        if self._is_critical():
            return +1
        elif self._is_medium():
            return 0
        else:
            return -1

    def _is_critical(self) -> bool:
        critical_patterns = [
            'sql_injection', 'rce', 'xss_unescaped',
            'auth_bypass', 'csrf_unprotected', 'hardcoded_secret'
        ]
        return any(p in self.finding_id.lower() for p in critical_patterns)

    def _is_medium(self) -> bool:
        medium_patterns = [
            'weak_crypto', 'xxe', 'insecure_random',
            'unvalidated_redirect', 'missing_encoding'
        ]
        return any(p in self.finding_id.lower() for p in medium_patterns)

Finding Scoring

def calculate_finding_score(finding: CQFinding, seed: int) -> float:
    """
    Deterministic scoring for finding prioritization.
    Same seed + same finding = same score.
    """
    # Use SplitMix to create finding hash
    rng = SplitMix64(seed)

    # Mix finding identity into RNG state
    file_hash = hash(finding.file) & 0xFFFFFFFF
    line_hash = finding.line & 0xFFFFFFFF
    id_hash = hash(finding.finding_id) & 0xFFFFFFFF

    rng.state ^= file_hash
    rng.next_u64()
    rng.state ^= line_hash
    rng.next_u64()
    rng.state ^= id_hash

    # Generate base score 0.0-1.0
    base_score = (rng.next_u64() % 100) / 100.0

    # Adjust by severity trit
    severity_weight = {+1: 1.0, 0: 0.5, -1: 0.1}[finding.severity_trit]

    return base_score * severity_weight

Parallel Scanning Architecture

Split-Stream Work Division

CQ-AI divides code into independent streams for parallel processing:

class ParallelCQScanner:
    def __init__(self, n_workers: int, seed: int):
        self.n_workers = n_workers
        self.seed = seed
        self.worker_seeds = self._generate_worker_seeds()

    def _generate_worker_seeds(self) -> List[int]:
        """
        Generate independent seeds for each worker.
        Guarantee: workers can run in any order, results compose.
        """
        rng = SplitMix64(self.seed)
        return [rng.next_u64() for _ in range(self.n_workers)]

    def scan_parallel(self, codebase_path: str) -> List[Finding]:
        """
        Scan with n_workers in parallel, deterministically.
        """
        files = get_all_files(codebase_path)

        # Distribute files to workers (round-robin)
        worker_files = [[] for _ in range(self.n_workers)]
        for i, file in enumerate(sorted(files)):
            worker_files[i % self.n_workers].append(file)

        # Run workers in parallel
        import concurrent.futures
        with concurrent.futures.ThreadPoolExecutor(max_workers=self.n_workers) as executor:
            futures = [
                executor.submit(
                    self._scan_worker,
                    self.worker_seeds[i],
                    worker_files[i]
                )
                for i in range(self.n_workers)
            ]

            # Collect results
            all_findings = []
            for future in concurrent.futures.as_completed(futures):
                all_findings.extend(future.result())

        # Deduplicate and sort deterministically
        unique_findings = {
            (f.file, f.line, f.finding_id): f
            for f in all_findings
        }

        return sorted(
            unique_findings.values(),
            key=lambda f: (f.file, f.line, f.finding_id)
        )

    def _scan_worker(self, seed: int, files: List[str]) -> List[Finding]:
        """Scan subset of files with given seed."""
        findings = []
        for filepath in files:
            findings.extend(cq_scan_file(filepath, seed))
        return findings

Out-of-Order Proof

Theorem: Results are order-independent.

def proof_out_of_order_invariant(codebase: str, seed: int):
    """
    Verify that different scan orders produce same findings.
    """
    # Scan in normal order
    files_asc = sorted(get_all_files(codebase))
    results_asc = [
        find for file in files_asc
        for find in cq_scan_file(file, seed)
    ]

    # Scan in reverse order
    files_desc = sorted(get_all_files(codebase), reverse=True)
    results_desc = [
        find for file in files_desc
        for find in cq_scan_file(file, seed)
    ]

    # Scan in random order
    import random
    rng = random.Random(seed)
    files_random = sorted(get_all_files(codebase))
    rng.shuffle(files_random)
    results_random = [
        find for file in files_random
        for find in cq_scan_file(file, seed)
    ]

    # All should deduplicate to same set
    findings_asc = canonical_findings(results_asc)
    findings_desc = canonical_findings(results_desc)
    findings_random = canonical_findings(results_random)

    assert findings_asc == findings_desc == findings_random, \
        "Order-independent invariant violated!"

MCP Server Integration

CQ-AI as MCP Tool

from anthropic import Anthropic

# MCP tool definition
CQ_AI_TOOL = {
    "name": "cq_ai_scan",
    "description": "Run CQ-AI deterministic security scan with ternary severity classification",
    "input_schema": {
        "type": "object",
        "properties": {
            "codebase_path": {
                "type": "string",
                "description": "Root path of code to scan"
            },
            "seed": {
                "type": "integer",
                "description": "SplitMix64 seed (same seed = same findings)"
            },
            "n_workers": {
                "type": "integer",
                "description": "Number of parallel workers (default: CPU count)"
            },
            "min_severity": {
                "type": "string",
                "enum": ["CRITICAL", "MEDIUM", "INFO"],
                "description": "Minimum severity to report (default: INFO)"
            }
        },
        "required": ["codebase_path", "seed"]
    }
}

# Usage in Claude conversation
def use_cq_ai(codebase_path: str, seed: int, min_severity: str = "INFO"):
    """
    AI skill to run CQ-AI and process findings.

    Same seed guarantees reproducible results for team collaboration.
    """
    scanner = ParallelCQScanner(
        n_workers=os.cpu_count(),
        seed=seed
    )

    findings = scanner.scan_parallel(codebase_path)

    # Filter by severity
    severity_order = {"CRITICAL": 1, "MEDIUM": 2, "INFO": 3}
    min_level = severity_order.get(min_severity, 3)

    filtered = [
        f for f in findings
        if severity_order.get(finding_severity_name(f.severity_trit)) <= min_level
    ]

    return {
        "total_findings": len(findings),
        "filtered_findings": len(filtered),
        "by_severity": {
            "CRITICAL": sum(1 for f in findings if f.severity_trit == +1),
            "MEDIUM": sum(1 for f in findings if f.severity_trit == 0),
            "INFO": sum(1 for f in findings if f.severity_trit == -1),
        },
        "top_critical": sorted(
            [f for f in filtered if f.severity_trit == +1],
            key=lambda f: calculate_finding_score(f, seed),
            reverse=True
        )[:10]
    }

Integration Commands

Install CQ-AI

# CQ is already installed via flox
flox install cq

# Create CQ-AI skill directory
mkdir -p ~/.cursor/skills/cq-ai

# Copy SKILL.md and Python implementation
cp CQ_AI_SKILL.md ~/.cursor/skills/cq-ai/
cp cq_ai.py ~/.cursor/skills/cq-ai/

# Register with Claude Code
claude code --register-skill cq-ai

Run Deterministic Scan

# Scan with fixed seed (reproducible)
python -c "
from cq_ai import ParallelCQScanner
scanner = ParallelCQScanner(n_workers=8, seed=0xDEADBEEF)
findings = scanner.scan_parallel('.')
for f in findings[:10]:
    print(f'{f.file}:{f.line} [{f.severity_trit:+d}] {f.finding_id}')
"

# Same seed, same findings
python -c "
from cq_ai import ParallelCQScanner
scanner = ParallelCQScanner(n_workers=8, seed=0xDEADBEEF)
findings = scanner.scan_parallel('.')
# Produces identical output
"

Team Collaboration Pattern

# Seed = git commit hash (reproducible across team)
import hashlib
import subprocess

# Get latest commit hash
commit = subprocess.check_output(
    ['git', 'rev-parse', 'HEAD']
).decode().strip()

# Convert to seed
seed = int(hashlib.sha256(commit.encode()).hexdigest()[:16], 16)

# Everyone runs same scan
scanner = ParallelCQScanner(n_workers=8, seed=seed)
findings = scanner.scan_parallel('.')

# All team members get SAME findings, same order

Ruby Implementation (for Fiber-based Parallelism)

class SplitMixTernary
  PHI_INV = 0x9E3779B97F4A7C15

  def initialize(seed)
    @state = seed
  end

  def next_u64
    z = ((@state ^ (@state >> 30)) * 0xBF58476D1CE4E5B9) & 0xFFFFFFFFFFFFFFFF
    @state = (@state + PHI_INV) & 0xFFFFFFFFFFFFFFFF
    z ^ (z >> 27)
  end

  def next_u32
    (next_u64 >> 32) & 0xFFFFFFFF
  end

  # Generate independent stream for parallel worker
  def split
    SplitMixTernary.new(next_u64)
  end
end

# Fiber-based parallel scanner
class CQAIScanner
  def initialize(seed, n_fibers = 4)
    @seed = seed
    @n_fibers = n_fibers
    @rng = SplitMixTernary.new(seed)
  end

  def scan_parallel(codebase_path)
    fibers = []
    workers = []

    # Create worker RNGs
    @n_fibers.times do
      workers << @rng.split
    end

    # Distribute files to fibers
    files = Dir.glob("#{codebase_path}/**/*").sort
    file_chunks = files.each_slice((files.length + @n_fibers - 1) / @n_fibers).to_a

    # Create fibers for parallel scanning
    file_chunks.each_with_index do |chunk, i|
      fibers << Fiber.new do
        scan_files(chunk, workers[i])
      end
    end

    # Resume all fibers
    findings = []
    fibers.each { |f| findings.concat(f.resume) }

    # Deduplicate and sort
    findings.uniq { |f| [f[:file], f[:line], f[:id]] }
             .sort_by { |f| [f[:file], f[:line], f[:id]] }
  end

  private

  def scan_files(files, rng)
    findings = []
    files.each do |file|
      findings.concat(cq_scan_file(file, rng.next_u32))
    end
    findings
  end
end

Julia Integration (for Scientific Computing)

module CQAIModule

using CQ_jll  # Bind to system CQ

mutable struct SplitMix64
    state::UInt64
end

PHI_INV = 0x9E3779B97F4A7C15

function next_u64(rng::SplitMix64)::UInt64
    z = ((rng.state ⊻ (rng.state >> 30)) * 0xBF58476D1CE4E5B9)
    rng.state += PHI_INV
    z ⊻ (z >> 27)
end

function scan_deterministic(codebase::String, seed::UInt64)::Vector
    rng = SplitMix64(seed)

    # Get all files
    files = readdir(codebase, recursive=true)

    # Sort deterministically by RNG
    sorted_files = sort(
        files,
        by=f -> next_u64(SplitMix64(seed)) % typemax(UInt32)
    )

    # Scan each file
    findings = []
    for file in sorted_files
        push!(findings, cq_scan_file(file, seed))
    end

    return findings
end

end  # module

Performance Characteristics

Properties Guaranteed

Determinism (SPI)

∀ codebase C, seed S:
  scan(C, S) = scan(C, S)  [always identical]

Out-of-Order Invariance

∀ codebase C, seed S, permutation π:
  canonical(scan_ordered(C, S)) =
  canonical(scan_ordered_by(C, S, π))

Tripartite Conservation

∀ codebase C, seed S:
  Σ findings with trit +1 +
  Σ findings with trit  0 +
  Σ findings with trit -1
  ≡ total findings (mod GF(3))

Example Usage with Claude Code

# Configure Claude to use CQ-AI skill
export ANTHROPIC_API_KEY="your-key"

# Run scan via AI skill
claude code --skill cq-ai --prompt "
Scan /Users/bob/ies/music-topos for security findings
using seed 0xCAFEBABE (fixed for reproducibility).
Prioritize CRITICAL findings with trit +1.
Run with 8 parallel workers.
"

References

• SplitMix64: Steele et al., "Linear congruential generators are dead"
• GF(3) Polarity: Girard, "Linear Logic" (balanced ternary semantics)
• Out-of-Order Proofs: Lamport, "Logical Clocks and Causal Ordering"
• Deterministic Parallelism: SPI (Same Physical Implementation) pattern from concurrency theory

Status: ✅ Production Ready Trit: +1 (Generative - finds vulnerabilities) Principle: Same seed → same findings (SPI guarantee) Last Updated: December 21, 2025