Co-Fita Core Components

Component Map

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:

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:

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

@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:

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:

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

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:

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:

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:

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

Node States

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:

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

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)

Promotes winners to production when conditions are met

Promotion Criteria

An artifact qualifies for promotion when:

Elo >= 1200 (the threshold)
At least 3 matches played
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):

Correcao tecnica — Technical correctness
Completude — Coverage of necessary cases
Clareza — Clarity and ease of following
Densidade — Respects the Parity Law (rigor >= context)
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

@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:

Builds a structured prompt containing:
- The failed approach
- Failure reason and category
- Context (iteration number, audit confidence)
- Unexplored sibling approaches
Sends to model for root cause analysis
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
Saves to JSONL for persistence across sessions

Reflection Output

Each reflection is exported as 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:

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:

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

Reads progress.json via StateManager
Checks last_seen timestamp against current time
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

# 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:

Phase V of the orchestrator — Exports failure tree, reflections, Elo arena, and governance state after each iteration
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

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

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:

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