Ocean Plastic Satellite Detection

SKILL.md — ocean-plastic-satellite-detection

Skill Registry Documentation

This document provides comprehensive documentation for the ocean-plastic-satellite-detection skill, including registration, resolution, execution, validation, and integration patterns.

Overview

Skill Name: ocean-plastic-satellite-detection

Version: 1.0.0

Domain: Remote-Sensing Ocean Plastic Detection & Collection Logistics

Purpose: Provides evidence-backed analysis for satellite-based detection of floating ocean plastic and optimal collection route planning, with rigorous quality gates and graceful degradation.

Skill Registration

Registration File

Skills are registered in config/skill_registration.json with the following structure:

{
  "schema_version": "1.0.0",
  "registry": {
    "skills": [
      {
        "id": "ocean-plastic-satellite-detection",
        "name": "Satellite Hyperspectral Detection of Floating Ocean Plastic",
        "version": "1.0.0",
        "description": "...",
        "domain": "Remote-Sensing Ocean Plastic Detection & Collection Logistics",
        "triggers": [...],
        "skills": [...],
        "quality_gates": {...},
        "tools": [...]
      }
    ]
  }
}

Registration Fields

| Field | Type | Required | Description | |-------|------|----------|-------------| | id | string | Yes | Unique skill identifier | | name | string | Yes | Human-readable skill name | | version | string | Yes | Semantic version | | description | string | Yes | Detailed description including trigger phrases | | domain | string | Yes | Domain of expertise | | triggers | array | Yes | Phrases/contexts that trigger the skill | | skills | array | Yes | Sub-skills that this skill orchestrates | | quality_gates | object | Yes | Quality gate definitions | | tools | array | Yes | Tools used by the skill |

Trigger Phrases

The skill triggers on phrases like:

• "ocean plastic detection"
• "satellite marine debris"
• "hyperspectral plastic"
• "collection route planning"
• "remote sensing ocean plastic"
• "marine debris satellite"
• "plastic accumulation zones"
• "ocean garbage patch"

Skill Resolution

Resolution Process

When a user query is received, the system:

1. Analyzes the query for trigger phrases and domain keywords
2. Matches against registered skills using fuzzy matching
3. Scores potential matches based on:

• Trigger phrase overlap (70% weight)
• Domain relevance (20% weight)
• Context similarity (10% weight)

1. Selects highest-scoring skill if score exceeds threshold (0.6)
2. Loads skill definition from SKILL.md
3. Initializes execution context

Resolution Algorithm

def resolve_skill(user_query: str) -> Optional[str]:
    """
    Resolve user query to skill ID.

    Args:
        user_query: User's input query

    Returns:
        Skill ID if resolved, None otherwise
    """
    # Load skill registry
    registry = load_skill_registration()

    # Score each skill
    scores = []
    for skill in registry['skills']:
        score = calculate_match_score(user_query, skill)
        if score >= 0.6:
            scores.append((skill['id'], score))

    # Return highest-scoring skill
    if scores:
        scores.sort(key=lambda x: x[1], reverse=True)
        return scores[0][0]

    return None

Skill Execution

Execution Flow

User Query
    │
    ▼
[Pre-Flight Checks]
    ├─ Language detection
    ├─ Input validation
    └─ Context initialization
    │
    ▼
[Step 1: sub-gather-requirements]
    ├─ Validate input schema
    ├─ Extract requirements
    └─ Quality gate check
    │
    ▼
[Step 2: sub-evidence-collector]
    ├─ Fetch primary sources
    ├─ Apply fallback chain if needed
    ├─ Update degradation level
    └─ Quality gate check
    │
    ▼
[Step 3: sub-core-analysis]
    ├─ Apply detection algorithms
    ├─ Map accumulation zones
    ├─ Plan collection routes
    ├─ Quantify uncertainty
    └─ Quality gate check
    │
    ▼
[Step 4: sub-knowledge-updater]
    ├─ Query knowledge base
    ├─ Surface citations with tiers
    ├─ Flag gaps
    └─ Quality gate check
    │
    ▼
[Step 5: sub-advisor]
    ├─ Synthesize all analysis
    ├─ Generate conclusion
    ├─ Apply disclosure
    └─ Quality gate check
    │
    ▼
[Final Quality Gate Review]
    ├─ Check all gates (U1-U6, G1-G4)
    ├─ Apply auto-fixes
    ├─ Generate limitation banner
    └─ Validate output schema
    │
    ▼
[Output Formatting]
    ├─ Apply template
    ├─ Translate to target language
    └─ Deliver to user

Execution Context

The execution context contains:

{
    'execution_id': str,
    'start_time': str (ISO format),
    'current_step': str,
    'detected_language': str ('en' or 'vi'),
    'output_language': str,
    'inputs': dict,
    'results': dict,
    'validation_passed': bool,
    'degradation_info': dict,
    'quality_gates': dict,
    'execution_metadata': dict,
    'limitation_banner': Optional[str]
}

Sub-Skill Invocation

Sub-skills are invoked using:

def invoke_sub_skill(skill_id: str, context: Dict[str, Any]) -> Dict[str, Any]:
    """
    Invoke a sub-skill with context.

    Args:
        skill_id: Sub-skill identifier
        context: Execution context

    Returns:
        Updated context with sub-skill results
    """
    # Load sub-skill definition
    sub_skill = load_sub_skill(skill_id)

    # Validate inputs
    validate_schema(context['inputs'], sub_skill['input_schema'])

    # Execute sub-skill
    result = execute_skill_logic(sub_skill, context)

    # Validate outputs
    validate_schema(result, sub_skill['output_schema'])

    # Update context
    context['results'][skill_id] = result

    return context

Quality Gates

Universal Gates (U1-U6)

| Gate | Check | Auto-Fix | Enforcement | |------|-------|----------|-------------| | U1 | ≥3 sources, ≥1 academic | Fetch from KB/evidence | Append sources | | U2 | Disclosure before recommendation | Prepend disclosure | Block until present | | U3 | Evidence hierarchy stated | Annotate tiers | Tag each source | | U4 | Language matches preference | Translate output | Re-detect language | | U5 | Template compliance | Reformat | Check sections | | U6 | Claim traceability | Flag unsupported | Mark claims |

Domain Gates (G1-G4)

| Gate | Check | Auto-Fix | Enforcement | |------|-------|----------|-------------| | G1 | Algorithm + sensor cited | Cite from metadata | Required field | | G2 | Accumulation zones mapped | Add zone mapping | Required for route | | G3 | Route drift-corrected | Apply correction | Required for planning | | G4 | Uncertainty quantified | Add uncertainty | Required field |

Gate Enforcement Logic

def enforce_quality_gate(
    gate_id: str,
    context: Dict[str, Any],
    max_attempts: int = 2
) -> bool:
    """
    Enforce a quality gate with auto-fix.

    Args:
        gate_id: Gate identifier (U1-U6, G1-G4)
        context: Execution context
        max_attempts: Maximum auto-fix attempts

    Returns:
        True if gate passed, False otherwise
    """
    gate_def = QUALITY_GATES[gate_id]
    attempts = 0

    while attempts  Dict[str, Any]:
    """
    Handle degraded execution.

    Args:
        context: Execution context

    Returns:
        Updated context with degradation handling
    """
    degradation_info = context.get('degradation_info', {})
    level = degradation_info.get('level', 0)

    if level == 0:
        return context  # No degradation

    # Generate limitation banner
    banner = generate_limitation_banner(level)
    context['limitation_banner'] = banner

    # Adjust output based on level
    if level >= 3:
        context['results']['availability'] = 'limited'
    elif level == 4:
        context['results']['availability'] = 'unavailable'

    return context

Tool Integration

Tool Registry

Tools are registered in config/tool_registry.json:

{
  "registry": {
    "tools": [
      {
        "id": "WebSearch",
        "name": "Web Search",
        "type": "external",
        "schema": {...},
        "rate_limit": {...}
      }
    ]
  }
}

Tool Invocation

def invoke_tool(
    tool_id: str,
    parameters: Dict[str, Any],
    context: Dict[str, Any]
) -> Dict[str, Any]:
    """
    Invoke a tool with parameters.

    Args:
        tool_id: Tool identifier
        parameters: Tool parameters
        context: Execution context

    Returns:
        Tool execution result
    """
    # Get tool from registry
    tool_def = TOOL_REGISTRY[tool_id]

    # Validate parameters against schema
    validate_schema(parameters, tool_def['input_schema'])

    # Check rate limits
    check_rate_limit(tool_id)

    # Execute tool
    result = execute_tool(tool_id, parameters)

    # Validate result
    validate_schema(result, tool_def['output_schema'])

    return result

Error Handling

Error Types

| Type | Recovery | Limit | |------|----------|-------| | sourcetimeout | Retry with backoff | 3 attempts | | invalidinput | Ask user | 2 attempts | | missinginput | Proceed with available | N/A | | staledata | Flag age | 1 attempt | | knowledgebasemiss | WebSearch gap-fill | 2 queries | | conflictingactions | Apply precedence | N/A | | completefailure | Emit notice | N/A |

Error Recovery

def handle_error(
    error: Exception,
    context: Dict[str, Any],
    error_type: Optional[str] = None
) -> Dict[str, Any]:
    """
    Handle execution error.

    Args:
        error: The error that occurred
        context: Execution context
        error_type: Optional error type

    Returns:
        Updated context after error handling
    """
    # Auto-detect error type if not provided
    if error_type is None:
        error_type = detect_error_type(error)

    # Get recovery strategy
    recovery = ERROR_RECOVERY_STRATEGIES[error_type]

    # Apply recovery
    context = recovery.recover(error, context)

    return context

Performance Monitoring

Metrics Collected

• Execution Metrics: Duration, success rate, token usage
• Quality Gate Metrics: Pass rates by gate
• Source Metrics: Access success rates, latency
• Degradation Metrics: Distribution of degradation levels

Metrics API

# Get metrics summary
summary = get_metrics_collector().get_summary()

# Record execution
with PerformanceTimer(get_metrics_collector(), 'operation_name'):
    # Do work
    pass

Configuration

Settings

Configuration is managed in config/settings.json:

{
  "application": {
    "name": "ocean-plastic-satellite-detection",
    "version": "1.0.0",
    "environment": "production"
  },
  "execution": {
    "timeout_ms": 300000,
    "max_retries": 3,
    "quality_gate_enforcement": "strict"
  },
  "knowledge_base": {
    "auto_update": true,
    "update_schedule": {
      "academic": "weekly",
      "news": "daily"
    }
  }
}

Logging Configuration

Logging is configured in config/logging.yaml with handlers for:

• Console output (standard format)
• File output (detailed format)
• JSON output (structured logging)

Extension Points

Adding New Sub-Skills

1. Define sub-skill in config/skill_registration.json
2. Create skills/sub-{skill_name}.md with frontmatter
3. Implement skill logic in tools/skill_{name}.py
4. Add input/output schemas to config/schemas/
5. Update main harness to invoke new sub-skill

Adding New Quality Gates

1. Define gate in config/skill_registration.json
2. Implement check logic in tools/quality_gates.py
3. Add auto-fix procedure
4. Update enforcement logic

Adding New Tools

1. Define tool in config/tool_registry.json
2. Implement tool class extending BaseTool
3. Add schema definitions
4. Register in TOOL_REGISTRY

Testing

Integration Tests

Run integration tests:

python tests/integration/test_full_pipeline.py

Test Scenarios

Test scenarios are defined in tests/test-scenarios.md:

# Run test scenarios
python tools/run_test_scenarios.py

Troubleshooting

See references/troubleshooting.md for:

• Common issues and solutions
• Error handling strategies
• Performance optimization
• Debugging techniques

References

• Architecture: ARCHITECTURE.md
• Project Details: PROJECT-detail.md
• Domain Methods: references/domain_methods.md
• Data Sources: references/data_sources.md
• Troubleshooting: references/troubleshooting.md

Source & license

This open-source skill is cataloged on AgentStack and links to its original source — we do not rehost the code.

• Author: dungnotnull
• Source: dungnotnull/ocean-plastic-satellite-detection-agent-skill
• License: MIT

Install and usage instructions live in the source repository linked above.