foundry-security-spec/
├── spec.md              # Main specification (~130 functional requirements)
├── constitution.md      # 11 inviolable principles
├── GLOSSARY.md         # Terminology reference
└── README.md           # Implementation guide

undefined

Key principles to understand:
- **No unsupervised execution**: Every finding requires explicit confirmation
- **Evidence-gated findings**: Claims without evidence don't become findings
- **Reproducibility**: Every finding must be reproducible from its evidence
- **Atomic progress**: Claims are indivisible units of work
- **Fail-safe defaults**: When stuck, escalate or yield—never guess

undefined

This creates `.specify/` directory for spec-driven development.

undefined

undefined

undefined

undefined

undefined

Answer questions in these categories:

**Identity & Scope:**

**Integration Choices:**

**Policy Choices:**

**Extension Scope (recommend NO for first build):**

undefined

undefined

Your `specs/001-foundry/spec.md` now contains YOUR specification with decisions filled in.

undefined

undefined

undefined

    # Initialize agent roles
    self.planner = Planner(llm_client)
    self.detector = Detector(llm_client, rules_corpus)
    self.explorer = Explorer(llm_client)
    self.validator = Validator(llm_client)

async def evaluate(self, target_repo: str) -> Dict:
    """Run complete evaluation coordinating all agents."""
    
    # FR-001: Orchestrator creates evaluation record
    eval_id = await self.findings.create_evaluation(
        target=target_repo,
        status="running"
    )
    
    try:
        # FR-010: Planner creates work plan
        plan = await self.planner.create_plan(target_repo)
        await self.claims.store_plan(eval_id, plan)
        
        # FR-020: Distribute work to detection and exploration
        detection_task = asyncio.create_task(
            self.run_detection(eval_id, plan)
        )
        exploration_task = asyncio.create_task(
            self.run_exploration(eval_id, plan)
        )
        
        # FR-005: Monitor heartbeats and budgets
        monitor_task = asyncio.create_task(
            self.monitor_health(eval_id)
        )
        
        # Wait for completion
        await asyncio.gather(
            detection_task,
            exploration_task,
            monitor_task
        )
        
        # FR-006: Check coverage gate before completion
        coverage = await self.calculate_coverage(eval_id)
        if coverage < plan.required_coverage:
            raise InsufficientCoverageError(
                f"Coverage {coverage}% < required {plan.required_coverage}%"
            )
        
        # Mark evaluation complete
        await self.findings.update_evaluation(
            eval_id,
            status="complete",
            coverage=coverage
        )
        
        return {
            "eval_id": eval_id,
            "status": "complete",
            "findings": await self.findings.count(eval_id),
            "coverage": coverage
        }
        
    except Exception as e:
        # FR-008: Fail-safe: mark evaluation failed
        await self.findings.update_evaluation(
            eval_id,
            status="failed",
            error=str(e)
        )
        raise

async def monitor_health(self, eval_id: str):
    """Monitor agent heartbeats and budgets."""
    while True:
        await asyncio.sleep(30)
        
        # FR-007: Check heartbeats
        stalled = await self.claims.find_stalled_claims(
            eval_id,
            heartbeat_threshold=300  # 5 minutes
        )
        
        for claim in stalled:
            # FR-007: Auto-block stalled claims
            await self.claims.block_claim(
                claim.id,
                reason="heartbeat_timeout"
            )
        
        # FR-009: Check budget exhaustion
        if await self.budget.is_exhausted(eval_id):
            await self.findings.update_evaluation(
                eval_id,
                status="budget_exhausted"
            )
            break

undefined

undefined

async def process_claim(self, claim: Claim) -> List[Finding]:
    """Apply detection rules to a code claim."""
    
    # FR-031: Extract relevant code from claim
    code_units = await self.extract_code_units(claim)
    
    findings = []
    
    for unit in code_units:
        # FR-032: Run rule engine
        rule_hits = await self.rule_engine.evaluate(
            code=unit.content,
            context=unit.context,
            language=unit.language
        )
        
        for hit in rule_hits:
            # FR-033: Convert rule hit to finding
            finding = Finding(
                claim_id=claim.id,
                rule_id=hit.rule_id,
                severity=hit.severity,
                weakness_id=hit.cwe_id,
                location=hit.location,
                evidence={
                    "rule_match": hit.matched_pattern,
                    "code_snippet": unit.content,
                    "line_range": hit.line_range
                },
                verdict="confirmed",  # Rules are deterministic
                status="validated"
            )
            findings.append(finding)
            
            # FR-034: Record coverage
            await self.record_coverage(claim.id, unit.path)
    
    return findings

async def extract_code_units(self, claim: Claim):
    """Use LLM to identify relevant code units in claim scope."""
    
    prompt = f"""
    Claim: {claim.description}
    Scope: {claim.scope}
    
    Identify all code units (functions, methods, classes) that should be
    evaluated for security issues related to this claim.
    
    Return as JSON array with: path, name, start_line, end_line
    """
    
    response = await self.llm.complete(prompt)
    return parse_code_units(response)

undefined

undefined

async def investigate_claim(self, claim: Claim) -> List[Finding]:
    """Creative exploration beyond static rules."""
    
    findings = []
    
    # FR-040: Generate investigation hypotheses
    hypotheses = await self.generate_hypotheses(claim)
    
    for hypothesis in hypotheses:
        # FR-041: Execute in isolated sandbox
        async with self.sandbox.session() as session:
            result = await self.test_hypothesis(
                session,
                hypothesis,
                claim
            )
            
            if result.is_vulnerability:
                # FR-042: Evidence-gated finding
                if not result.has_reproduction:
                    # Don't create finding without evidence
                    continue
                
                finding = Finding(
                    claim_id=claim.id,
                    severity=result.severity,
                    weakness_id=result.weakness_id,
                    description=result.description,
                    evidence=result.evidence,
                    verdict="needs-validation",
                    status="pending"
                )
                findings.append(finding)
                
                # FR-043: Check if rules missed this
                if await self.should_have_detected(finding):
                    await self.record_rule_gap(finding)
    
    return findings

async def generate_hypotheses(self, claim: Claim) -> List[Dict]:
    """Use LLM to generate creative test hypotheses."""
    
    prompt = f"""
    You are exploring code for security issues that static rules may miss.
    
    Claim: {claim.description}
    Code scope: {claim.scope}
    
    Generate 3-5 security hypotheses to test:
    - Focus on logic bugs, state confusion, race conditions
    - Consider what rules can't express (context-dependent issues)
    - Prioritize high-impact scenarios
    
    For each hypothesis provide:
    - What to test
    - Why it might be vulnerable
    - How to reproduce if vulnerable
    
    Return as JSON array.
    """
    
    response = await self.llm.complete(
        prompt,
        temperature=0.7  # Higher for creative exploration
    )
    return parse_hypotheses(response)

async def record_rule_gap(self, finding: Finding):
    """Record that rules failed to detect this issue."""
    
    gap = RuleGap(
        finding_id=finding.id,
        weakness_id=finding.weakness_id,
        pattern=finding.evidence.get("vulnerable_pattern"),
        reason="explorer_found_missed_by_detector",
        suggested_rule=await self.draft_rule(finding)
    )
    
    # FR-044: Feed into rule corpus improvement
    await self.rule_gaps.store(gap)

undefined

undefined

async def validate_finding(self, finding: Finding) -> ValidationResult:
    """Reproduce and confirm finding from evidence."""
    
    # FR-050: Check evidence completeness
    if not self.has_sufficient_evidence(finding):
        return ValidationResult(
            verdict="rejected",
            reason="insufficient_evidence"
        )
    
    # FR-051: Attempt reproduction
    async with self.sandbox.session() as session:
        reproduced = await self.reproduce_issue(
            session,
            finding.evidence
        )
        
        if not reproduced:
            return ValidationResult(
                verdict="rejected",
                reason="not_reproducible"
            )
        
        # FR-052: Verify severity assessment
        actual_severity = await self.assess_severity(
            session,
            finding
        )
        
        if actual_severity != finding.severity:
            finding.severity = actual_severity
            finding.evidence["severity_adjustment"] = {
                "original": finding.severity,
                "validated": actual_severity
            }
        
        # FR-053: Generate fingerprint for deduplication
        fingerprint = await self.generate_fingerprint(finding)
        
        return ValidationResult(
            verdict="confirmed",
            fingerprint=fingerprint,
            severity=actual_severity,
            reproduction_evidence=session.get_transcript()
        )

def has_sufficient_evidence(self, finding: Finding) -> bool:
    """Check if finding has required evidence."""
    required = ["location", "description"]
    
    if finding.severity in ["critical", "high"]:
        required.extend(["reproduction_steps", "impact"])
    
    return all(k in finding.evidence for k in required)

async def generate_fingerprint(self, finding: Finding) -> str:
    """Create stable fingerprint for deduplication."""
    
    # FR-054: Fingerprint combines weakness + location + root cause
    components = [
        finding.weakness_id,
        finding.location.get("file_path"),
        finding.location.get("function_name"),
        finding.evidence.get("root_cause_pattern")
    ]
    
    fingerprint_input = "|".join(str(c) for c in components if c)
    return hashlib.sha256(fingerprint_input.encode()).hexdigest()[:16]

undefined

undefined

async def publish_finding(self, finding: Finding) -> str:
    """Create issue in tracker for confirmed finding."""
    
    # FR-060: Only publish confirmed findings
    if finding.verdict != "confirmed":
        raise ValueError(f"Cannot publish {finding.verdict} finding")
    
    # FR-061: Check for existing issue via fingerprint
    existing = await self.find_existing_issue(finding.fingerprint)
    if existing:
        return existing.issue_id
    
    # FR-062: Format issue according to tracker schema
    issue_body = self.format_issue(finding)
    
    # FR-063: Create issue
    async with aiohttp.ClientSession() as session:
        async with session.post(
            f"{self.tracker_url}/rest/api/2/issue",
            headers={
                "Authorization": f"Bearer {self.api_token}",
                "Content-Type": "application/json"
            },
            json=issue_body
        ) as resp:
            resp.raise_for_status()
            result = await resp.json()
            issue_id = result["key"]
    
    # FR-064: Update finding with issue reference
    finding.issue_id = issue_id
    finding.status = "published"
    await finding.save()
    
    return issue_id

def format_issue(self, finding: Finding) -> Dict:
    """Format finding as issue tracker ticket."""
    
    description = f"""
    *Security Finding from Foundry Evaluation*
    
    *Severity:* {finding.severity.upper()}
    *Weakness:* {finding.weakness_id}
    *Location:* {finding.location.get('file_path')}:{finding.location.get('line_number')}
    
    h3. Description
    {finding.description}
    
    h3. Evidence
    {self.format_evidence(finding.evidence)}
    
    h3. Reproduction
    {finding.evidence.get('reproduction_steps', 'See evidence above')}
    
    ---
    Fingerprint: {finding.fingerprint}
    Evaluation ID: {finding.eval_id}
    """
    
    return {
        "fields": {
            "project": {"key": self.project_key},
            "summary": f"[{finding.severity.upper()}] {finding.weakness_id}: {finding.get_short_description()}",
            "description": description,
            "issuetype": {"name": "Security Vulnerability"},
            "priority": {"name": self.map_severity_to_priority(finding.severity)},
            "labels": [
                f"foundry:eval:{finding.eval_id}",
                f"foundry:weakness:{finding.weakness_id}",
                f"foundry:fingerprint:{finding.fingerprint}"
            ]
        }
    }

undefined

undefined

vcs:
  type: "gitlab"
  url: "${GITLAB_URL}"
  token: "${GITLAB_TOKEN}"

issue_tracker:
  type: "jira"
  url: "${JIRA_URL}"
  project: "SEC"
  token: "${JIRA_TOKEN}"

datastore:
  type: "postgresql"
  connection_string: "${DATABASE_URL}"

undefined

undefined

undefined

undefined

# Initialize orchestrator
orchestrator = Orchestrator(
    llm_client=create_llm_client(config.integrations.llm),
    finding_store=FindingStore(config.integrations.datastore),
    claim_store=ClaimStore(config.integrations.datastore),
    budget_manager=BudgetManager(config.budget)
)

# Run evaluation
result = await orchestrator.evaluate(
    target_repo="https://gitlab.internal/acme/webapp"
)

print(f"Evaluation {result['eval_id']} complete")
print(f"Findings: {result['findings']}")
print(f"Coverage: {result['coverage']}%")

undefined

undefined

try:
    # Send heartbeat while working
    heartbeat_task = asyncio.create_task(
        self.send_heartbeats(claim.id)
    )
    
    # Do the work
    result = await self.do_work(claim)
    
    # Mark complete
    await self.claims.complete(claim.id, result)
    
except Exception as e:
    # Fail claim, don't retry (constitution principle)
    await self.claims.fail(claim.id, str(e))
finally:
    heartbeat_task.cancel()

undefined

undefined

if not issue.get("reproduction"):
    logger.warning(f"No reproduction for issue in {claim.id}, skipping")
    return None

# Evidence is sufficient
return Finding(
    claim_id=claim.id,
    location=issue["location"],
    evidence={
        "reproduction": issue["reproduction"],
        "impact": issue["impact"],
        "code_snippet": issue["code"]
    },
    verdict="needs-validation"
)

undefined

class BudgetManager:
    async def check_budget(self, eval_id: str, tokens_requested: int) -> bool:
        """Check if budget allows operation."""
        used = await self.get_tokens_used(eval_id)
        limit = self.config.max_tokens
        
        if used + tokens_requested > limit:
            await self.notify_budget_exhausted(eval_id)
            return False
        
        return True
    
    async def record_usage(self, eval_id: str, tokens: int):
        """Record token usage."""
        await self.db.execute(
            "INSERT INTO token_usage (eval_id, tokens, timestamp) VALUES ($1, $2, NOW())",
            eval_id, tokens
        )

undefined

**Check explorer creativity:**
```python

undefined

undefined

**Check for deadlocks:**
```sql
-- Find stalled claims
SELECT claim_id, agent_id, last_heartbeat, status
FROM claims
WHERE status = 'in_progress'
  AND last_heartbeat < NOW() - INTERVAL '5 minutes';

undefined

**Check deduplication:**
```python

undefined

undefined

**Check claim distribution:**
```python

undefined

undefined

async def scan(self, code: str, language: str):
    # Rules execute deterministically
    return await self.rules.evaluate(code, language)

# Write to CodeGuard format
await write_codeguard_rule(rule, "rules/corpus/custom/")

undefined