harness-core

Harness Core Library

Overview

harness-core is the base SDK / plugin library for implementing NeuroClaw harness engineering standards across all skills.

Instead of reimplementing validation, checkpointing, logging, and drift detection in every skill, harness-core provides reusable, well-tested Python classes and utilities that all other skills can import and extend.

Key design principle: Harness-awareness is not optional — every data-processing, model-execution, and experiment-running skill should import from this library to achieve:

• Standardized self-verification across all skills
• Reproducible, hash-verified experiment logs
• Automatic checkpoint/resume capability
• Drift detection and anomaly alerts
• Privacy-preserving audit trails

When to Use This Skill

Directly: Rarely — this is a library, not a user-facing skill.

Indirectly (as a dependency):

• When developing or modifying skills like experiment-controller, run_models, fmri-skill, smri-skill, etc.
• When using skills that have been enhanced to support harness engineering
• When integrating external tools or models into NeuroClaw (inherit harness patterns)

Core Components

1. HarnessController (Main Orchestrator)

Purpose: Manages the full lifecycle of harness-compliant execution.

Usage:

from skills.harness_core import HarnessController

controller = HarnessController(
    task_name="fmri_preprocessing",
    session_id="exp_20260405_143000",
    checkpoint_dir="./checkpoints",
    log_dir="./logs"
)

# Automatic environment snapshot capture
controller.initialize()

# Define task phases
controller.add_phase("quality_check", description="Verify input BIDS compliance")
controller.add_phase("preprocessing", description="Apply fMRI preprocessing")
controller.add_phase("feature_extraction", description="Extract ROI time series")

# Execute with auto-checkpointing
for phase_name in ["quality_check", "preprocessing", "feature_extraction"]:
    try:
        phase = controller.get_phase(phase_name)
        result = execute_phase(phase.name)  # User-defined function
        controller.record_phase_success(phase_name, result)
        controller.save_checkpoint(f"after_{phase_name}")
    except Exception as e:
        controller.record_phase_failure(phase_name, str(e))
        controller.save_checkpoint(f"failed_{phase_name}")
        raise

# Auto-generates: audit_report.md, environment_manifest.json, checkpoints/