---
name: co-fita-components
type: reference
title: "Co-Fita Core Components Reference"
description: "Deep-dive into every component of the Hyper-Harness: Red Team, HashMath, Failure Tree, Elo, Reflector, Watchdog, Wiki Sync, MCP Server"
tags: [co-fita, components, red-team, hash-math, failure-tree, elo, reflector, watchdog, wiki-sync, mcp]
timestamp: 2026-07-21
---

# Co-Fita Core Components

## Component Map

```mermaid
graph LR
    subgraph Verification
        RT["Red Team Auditor"]
        HM["HashMath Verifier"]
    end

    subgraph Learning
        FT["Failure Tree"]
        RF["Reflector"]
        EC["Elo Consolidator"]
    end

    subgraph Infrastructure
        SM["State Manager"]
        WD["Watchdog"]
        WS["Wiki Sync"]
        MC["MCP Server"]
    end

    subgraph Plugins
        RG["Registry"]
        PL["Plugins (SearXNG, Crawl4AI, Telegram)"]
    end

    RT -->|"failure categorization"| FT
    FT -->|"unexplored siblings"| RF
    RF -->|"reflection data"| EC
    HM -->|"hash chain"| SM
    WD -->|"stall detection"| SM
    SM -->|"state files"| WS
    RG -.->|"registers"| PL
```

---

## 1. Red Team Auditor

**Source:** `harness_core/red_team.py`
**Purpose:** Adversarial verification layer that intercepts diffs between execution and commitment, preventing reward hacking and LLM steganography.
**Classification:** L3 Verifier (between L2 Execution and L4 Commit)

### How It Works

The `RedTeamAuditor` performs synchronous heuristic and cryptographic analysis on diffs produced by the model:

#### Check 1: Reward Hacking Detection

Scans diff headers (`+++ b/...`) for modifications to forbidden files:

```python
forbidden_paths = [
    r"tests/.*",          # Test files
    r"eval\.py",          # Evaluation logic
    r"\.github/.*",       # CI/CD configuration
    r"metrics\.json",     # Metric definitions
]
```

If the agent modifies any of these, it is attempting to cheat the evaluation — modifying the yardstick rather than the result. The audit immediately fails with `violation_type="REWARD_HACKING"` and `confidence=1.0`.

#### Check 2: Steganography Detection

Extracts added lines (`+` lines) from the diff and calculates Shannon entropy:

```python
entropy = -sum(p_x * log2(p_x) for each character in line)
```

- **Normal code/natural language:** entropy ~3.5-4.0
- **Base64/encrypted payloads:** entropy > 4.5-5.0
- **Threshold:** 4.8

Lines shorter than 20 characters are excluded (too short for meaningful payloads). If any line exceeds the threshold, the audit fails with `violation_type="STEGANOGRAPHY_DETECTED"` and `confidence=0.85`.

#### Future: LLM-as-a-Judge

The code includes a placeholder for a third check: sending the diff to a smaller constrained model (e.g., Llama 3 8B) to ask "Is this code malicious or does it cheat the metric?" Not yet implemented.

### AuditResult Dataclass

```python
@dataclass
class AuditResult:
    passed: bool                    # Whether the audit passed
    violation_type: Optional[str]   # "REWARD_HACKING" or "STEGANOGRAPHY_DETECTED"
    confidence: float               # 0.0-1.0 confidence in the result
    mitigation_feedback: str        # Human-readable explanation
```

### Integration Point

Called in Phase III of the orchestrator:
```python
audit_res: AuditResult = self.red_team.audit_trace(diff_result, [])
```

If `audit_res.passed` is False, the iteration takes the failure path.

---

## 2. HashMath Verifier

**Source:** `harness_core/hash_math.py`
**Purpose:** Cryptographic anchoring of discoveries via SHA-256 hash chain.
**Classification:** L4 Commit (ensures chronological integrity)

### How It Works

Functions like a blockchain block producer. Every successful empirical test is sealed with a hash that depends on:

1. **Prior hash** (linking to history)
2. **Diff content** (the action taken)
3. **Metrics** (the outcome, JSON-serialized with sorted keys for determinism)

```python
payload = f"{prior_hash}::{diff_content}::{metrics_str}"
new_hash = hashlib.sha256(payload.encode('utf-8')).hexdigest()
```

### Chain Verification

`verify_chain()` reconstructs the hash for each entry in the history and compares it against the stored hash. If anyone tampers with `progress.json`, changes metrics to fake success, or modifies the history chain, the verification fails:

```python
expected_hash = self.seal_discovery(
    prior_hash=current_hash,
    diff_content=entry.get("diff", ""),
    metrics=entry.get("metrics", {})
)
if expected_hash != entry.get("hash"):
    return False  # Chain broken
```

### Design Rationale

- **Deterministic serialization:** `json.dumps(metrics, sort_keys=True)` ensures identical metrics always produce identical hashes
- **Genesis anchor:** Chain starts at `"GENESIS_HASH"`, creating an unbreakable link from the first iteration
- **Tamper detection:** Even a single character change in any diff or metric breaks the entire chain from that point forward

### Integration Point

Called in Phase IV on success path:
```python
new_hash = self.hash_math.seal_discovery(
    self.current_state_hash, diff_result, metrics
)
```

---

## 3. Failure Tree

**Source:** `harness_core/failure_tree.py`
**Purpose:** Execution tree that preserves failed branches and identifies unexplored opportunities.
**Inspired by:** EnCompass (MIT CSAIL), Reflexion (Shinn et al., 2023), Tree of Thoughts (Yao et al., 2023)

### Core Concept

When a path fails, instead of discarding it, the system:

1. Records the branch point and failure reason
2. Generates a verbal reflection (Reflexion-style)
3. Identifies sibling branches never explored
4. Prioritizes unexplored siblings for future iterations

### Node States

```mermaid
stateDiagram-v2
    PENDING --> RUNNING : branch created
    RUNNING --> SUCCESS : audit passed + anchored
    RUNNING --> FAILED : audit failed
    PENDING --> UNEXPLORED : non-first child
```

| State | Meaning |
|:---|:---|
| `PENDING` | Created but not yet queued |
| `RUNNING` | Currently being explored |
| `SUCCESS` | Audit passed, hash anchored |
| `FAILED` | Audit failed, reason recorded |
| `UNEXPLORED` | Preserved for future exploration |

### Failure Taxonomy

The `FailureCategory` enum classifies agent failure modes:

| Category | Description |
|:---|:---|
| `goal_drift` | Agent strayed from the original objective |
| `tool_misuse` | Agent used a tool incorrectly |
| `logic_error` | Code or reasoning contains a logical flaw |
| `stalling` | Agent entered a repetitive loop |
| `silent_failure` | Agent reported success but produced nothing |
| `hallucination` | Agent fabricated facts or code |
| `external_error` | Transient failure (timeout, API error) |
| `unknown` | Unclassified failure |

### Priority Scoring

`get_promising_unexplored()` uses heuristic scoring:

- **Base score:** `10.0 - depth` (shallower = more promising)
- **+3.0 bonus** if a sibling failed from `EXTERNAL_ERROR` (transient, might work now)
- **+1.0 bonus** if a sibling failed from `HALLUCINATION` (different approach may avoid it)

### Mermaid Export

`to_mermaid()` generates a visual diagram with color-coded nodes:
- Green: SUCCESS
- Red: FAILED
- Yellow: RUNNING
- Blue-gray: UNEXPLORED
- Dark gray: PENDING

### Persistence

File-based at `workspace/failure_tree/failure_tree.json`. Loaded on init, saved on every mutation. Thread-safe via file-based locking (JSON serialization).

---

## 4. Elo Consolidator

**Source:** `harness_core/consolidator.py`
**Purpose:** Tournament-style ranking for skills, rules, and system prompts.
**Parameters:** K=32 (standard chess Elo), promotion threshold=1200, min matches=3

### How It Works

Each artifact (skill, rule, prompt) has an Elo rating starting at 1200. The Consolidator:

1. **Registers** artifacts in a tournament registry
2. **Pairs** artifacts by (category, name) for matches
3. **Generates judge prompts** asking the model to compare two versions
4. **Records match results** and updates Elo ratings using the standard formula:

```python
expected = 1.0 / (1.0 + math.pow(10, (opponent_elo - artifact_elo) / 400))
actual = 1.0 if won else 0.0
artifact_elo += K_FACTOR * (actual - expected)
```

5. **Promotes** winners to production when conditions are met

### Promotion Criteria

An artifact qualifies for promotion when:

1. Elo >= 1200 (the threshold)
2. At least 3 matches played
3. Not already active in production

Promotion deactivates the current champion and activates the challenger.

### Judge Prompt

The judge evaluates artifacts on 5 criteria (in Portuguese, per the project's bilingual convention):

1. **Correcao tecnica** — Technical correctness
2. **Completude** — Coverage of necessary cases
3. **Clareza** — Clarity and ease of following
4. **Densidade** — Respects the Parity Law (rigor >= context)
5. **Robustez** — Handles edge cases

### Deployment

When `agents_dir` is set, promoted artifacts are written to:

- Skills: `.agents/skills/{name}/SKILL.md`
- Rules: `.agents/rules/{name}.md`

### Data Model

```python
@dataclass
class EloArtifact:
    id: str              # "{category}-{name}-{uuid8}"
    category: str        # "skill", "rule", "prompt"
    name: str
    content: str
    elo: float           # Starting at 1200
    matches_played: int
    wins: int
    losses: int
    is_active: bool      # Currently deployed
    proposed_by: str     # "human", "self-improver", "consolidator"
```

### Wiki Export

`export_leaderboard_markdown()` generates a formatted table for the unified-harness-wiki.

---

## 5. Reflector

**Source:** `harness_core/reflector.py`
**Purpose:** Reflexion-style verbal reinforcement engine.
**Based on:** Reflexion (Shinn et al., 2023)

### How It Works

When an approach fails, the Reflector:

1. **Builds a structured prompt** containing:
   - The failed approach
   - Failure reason and category
   - Context (iteration number, audit confidence)
   - Unexplored sibling approaches

2. **Sends to model** for root cause analysis

3. **Parses the response** into a `Reflection` object:
   - `root_cause` — Why exactly it failed
   - `alternative_approaches` — What else could work
   - `prevention_strategy` — How to avoid similar failures
   - `promising_siblings` — Which unexplored branches look most promising

4. **Saves to JSONL** for persistence across sessions

### Reflection Output

Each reflection is exported as Markdown:

```markdown
## Reflexao: {approach[:60]}

**Categoria**: `logic_error`
**Timestamp**: 2026-07-21 14:30

### Razao da Falha
{failure_reason}

### Root Cause Analysis
{root_cause}

### Abordagens Alternativas
- {alt1}
- {alt2}

### Estrategia de Prevencao
{prevention_strategy}

### Ramos Irmãos Promissores
- {sibling1}
```

### Pattern Analysis

`get_failure_patterns()` analyzes recent reflections to find:

- Failure distribution by category
- Recurring root cause keywords (timeout, api, memory, format, parsing, hallucination)
- Total reflection count

### Integration Point

Called in Phase III on the failure path, after the Red Team audit fails:

```python
prompt = self.reflector.build_reflection_prompt(
    approach=approaches[0],
    failure_reason=f"Red Team audit failed: {audit_res.violation_type}",
    category=FailureCategory.LOGIC_ERROR,
    context={"iteration": iteration, "audit_confidence": audit_res.confidence},
    sibling_approaches=[s.approach for s in siblings]
)
reflection_raw = self.acp_client.execute_prompt(prompt)
```

---

## 6. Watchdog

**Source:** `harness_core/watchdog.py`
**Purpose:** Stall detection and recovery initiation.
**Implementation:** Daemon thread with configurable thresholds

### How It Works

The Watchdog runs as a daemon thread, checking the harness liveness at regular intervals:

```python
class Watchdog(threading.Thread):
    def __init__(self, state_mgr, check_interval=60, stale_threshold=7200):
```

| Parameter | Default | Meaning |
|:---|:---|:---|
| `check_interval` | 60s | How often to check liveness |
| `stale_threshold` | 7200s (2h) | How long before declaring a stall |

### Detection Logic

1. Reads `progress.json` via StateManager
2. Checks `last_seen` timestamp against current time
3. If `current_time - last_seen > stale_threshold`:
   - Logs `"stall_detected"` event
   - Sets `progress.status = "stalled"`
   - Calls `_initiate_recovery()`

### Recovery

`_initiate_recovery()` currently logs the event. In the full implementation described in the `Deli_AutoResearch` skill, this would spin up an emergency headless agent to continue the loop.

### Lifecycle

```python
# Start (called by orchestrator.start_loop())
self.watchdog.start()

# Stop (called by orchestrator._shutdown())
self.watchdog.stop()
```

The watchdog is a daemon thread (`daemon=True`), so it dies automatically when the main process exits.

---

## 7. Wiki Sync

**Source:** `harness_core/wiki_sync.py`
**Purpose:** Auto-populates `unified-harness-wiki/` from governance proposals and congress sessions.

### How It Works

Queries the Odysseus SQLAlchemy database for implemented proposals and exports each as a Markdown file:

```
unified-harness-wiki/proposals/prop_{id}.md
```

Each file contains:
- Proposal ID and title
- Category and proposer
- Description, rationale, and strategy notes

### Integration

Wiki sync is called from two places:

1. **Phase V of the orchestrator** — Exports failure tree, reflections, Elo arena, and governance state after each iteration
2. **Standalone script** — `uv run python wiki_sync.py` for manual sync

### Wiki Output Files

| File | Content |
|:---|:---|
| `failure-tree.md` | Mermaid diagram + JSON statistics |
| `reflections.md` | Recent reflections as structured Markdown |
| `elo-arena.md` | Elo leaderboard table |
| `proposals/prop_*.md` | Individual implemented proposal docs |

---

## 8. MCP Server

**Source:** `harness_core/harness_mcp_server.py`
**Purpose:** Expose harness capabilities via Model Context Protocol.

### Exposed Tools

| Tool | Description |
|:---|:---|
| `send_telegram_notification` | Send message to Lider Suprema via Telegram |
| `propose_governance_change` | Submit a new governance proposal to Kanban |
| `list_governance_proposals` | List current proposals (filterable by status) |
| `chancela_proposal` | Apply final approval/rejection (restricted to Alefita) |
| `search_wiki` | Search the unified-harness-wiki |
| `audit_red_team` | Run Red Team audit against a strategy |
| `elo_debate` | Run Elo-ranked debate between two approaches |

### Architecture

The MCP server uses FastMCP and connects to:

- `TelegramGateway` for notifications
- `GovernanceEngine` (via dashboard API at `localhost:7000/api/governance`)
- `RedTeamTools` for auditing
- `HashMath` for Elo evaluation

### Usage

```bash
uv run python harness_mcp_server.py
```

The server enables external AI agents to interact with Co-Fita governance without direct Python imports.

---

## 9. State Manager

**Source:** `harness_core/state_manager.py`
**Purpose:** Implements the Deli_AutoResearch state persistence pattern.

### Design Principle

"All progress is written to state files. Never use conversation memory."

This is critical for long-running autonomous sessions where context windows get compacted.

### File Layout

```
workspace/
├── state/
│   ├── task_spec.md              # The research goal
│   ├── progress.json             # Current iteration, findings, status
│   └── directions_tried.json     # Approaches already attempted
└── logs/
    ├── orchestrator.jsonl        # Orchestrator events
    ├── heartbeat.jsonl           # Watchdog events
    └── {source}.jsonl            # Per-component logs
```

### Key Methods

| Method | Purpose |
|:---|:---|
| `init_task(spec)` | Create task_spec.md and baseline progress |
| `get_progress()` | Read current progress from disk |
| `update_progress(updates)` | Merge updates and refresh `last_seen` |
| `log_event(source, level, event, detail)` | Append JSONL log entry |

The `last_seen` timestamp is updated on every `update_progress()` call, which is how the Watchdog detects stalls.

---

## 10. Registry and Plugins

**Source:** `harness_core/registry.py` and `harness_core/plugins.py`
**Purpose:** Open-Closed architecture for extensible plugin system.

### Registry Pattern

```python
class PluginRegistry(Generic[T]):
    _plugins: dict[str, type[T]] = {}

    @classmethod
    def register(cls, name: str):
        def decorator(plugin_cls):
            cls._plugins[name] = plugin_cls
            return plugin_cls
        return decorator

    @classmethod
    def create(cls, name: str, **kwargs) -> T:
        return cls.get(name)(**kwargs)
```

Each registry type (`SearchProviderRegistry`, `WebCrawlerRegistry`, etc.) inherits from `PluginRegistry` with a specific type parameter. Registration happens at import time via decorators.

### Concrete Implementations

| Plugin | Registry | Endpoint | Validation |
|:---|:---|:---|:---|
| `SearXNGProvider` | `SearchProviderRegistry` | localhost:8888 | Docker, 6+ days uptime |
| `Crawl4AIProvider` | `WebCrawlerRegistry` | localhost:11235 | Docker, v0.8.9, Playwright |
| `TelegramGateway` | `GatewayRegistry` | Telegram Bot API | Credentials from ~/.hermes/.env |
| `OpenCodeProvider` | `ModelProviderRegistry` | ACP -> Odysseus | DB access verified |

### Interface Contracts

| Interface | Methods |
|:---|:---|
| `SearchProvider` | `async search(query, n) -> list[dict]` |
| `WebCrawler` | `async crawl(url) -> dict` |
| `ModelProvider` | `async generate(prompt)`, `async chat(messages)`, `async health_check()` |
| `VerificationLayer` | `async verify(artifact, context) -> dict` |
| `NotificationGateway` | `async send_message(text)`, `async start_listening(callback)` |
| `LearningBackend` | `async learn(findings) -> str` (placeholder) |

---

## Cross-Component Interactions

The components form a pipeline with feedback loops:

```mermaid
sequenceDiagram
    participant ORC as Orchestrator
    participant RT as Red Team
    participant FT as Failure Tree
    participant RF as Reflector
    participant HM as HashMath
    participant EC as Elo
    participant GOV as Governance

    ORC->>RT: audit_trace(diff)
    alt Audit Passed
        RT-->>ORC: AuditResult(passed=True)
        ORC->>HM: seal_discovery(prior_hash, diff, metrics)
        HM-->>ORC: new_hash
        ORC->>FT: mark_success(node_id, metrics)
    else Audit Failed
        RT-->>ORC: AuditResult(passed=False, violation_type)
        ORC->>FT: mark_failed(node_id, reason, category)
        ORC->>FT: get_unexplored_siblings(node_id)
        FT-->>ORC: [sibling_nodes]
        ORC->>RF: build_reflection_prompt(...)
        ORC->>ORC: execute_prompt(reflection)
        ORC->>RF: save_reflection(reflection)
        ORC->>GOV: process_iteration(audit_passed=False)
    end
```

This pipeline ensures that:
- Every execution is verified before it changes state
- Every failure produces a learning artifact
- Every success is cryptographically anchored
- The system improves its own components through Elo tournaments
- The human retains final authority through the chancela mechanism