Halal Kosher Compliance Monitor

Halal-Kosher Compliance Monitor — Production-Grade Skill Registry

Overview

The halal-kosher-compliance-monitor is a sophisticated, production-grade agent system for Halal and Kosher food certification compliance analysis. This skill implements a flexible agent architecture with intelligent routing, comprehensive error handling, graceful degradation, and continuous self-improvement through an automated knowledge pipeline.

Architecture

Component Hierarchy

halal-kosher-compliance-monitor (Main Skill)
├── Core Systems
│   ├── config/settings.py — Configuration management
│   ├── hooks/system.py — Lifecycle hooks and event emission
│   ├── core/registry.py — Skill registration and resolution
│   ├── core/router.py — Agent chain-of-thought router
│   ├── core/tools.py — Tool definitions and execution
│   └── core/errors.py — Error handling and recovery
│
├── Agents (Specialized)
│   ├── GatherRequirementsAgent — Intake and requirements clarification
│   ├── EvidenceCollectorAgent — Multi-source data aggregation
│   ├── CoreAnalyzerAgent — Compliance analysis engine
│   ├── KnowledgeUpdaterAgent — Academic evidence integration
│   └── AdvisorAgent — Synthesis and recommendation
│
├── Sub-Skills (Modular)
│   ├── skills/sub-gather-requirements.md
│   ├── skills/sub-evidence-collector.md
│   ├── skills/sub-core-analysis.md
│   ├── skills/sub-knowledge-updater.md
│   └── skills/sub-advisor.md
│
└── Supporting Systems
    ├── tools/knowledge_updater.py — Knowledge crawl pipeline
    ├── tools/run_test_scenarios.py — Test orchestrator
    └── SECOND-KNOWLEDGE-BRAIN.md — Living knowledge base

Skill Registration System

Registration Schema

All skills in the registry must conform to this JSON schema:

{
  "type": "object",
  "properties": {
    "skill_id": {
      "type": "string",
      "pattern": "^[a-f0-9]{16}$",
      "description": "Unique identifier generated from name:version"
    },
    "name": {
      "type": "string",
      "minLength": 1,
      "maxLength": 100,
      "description": "Human-readable skill name"
    },
    "version": {
      "type": "string",
      "pattern": "^\\d+\\.\\d+\\.\\d+$",
      "description": "Semantic version"
    },
    "description": {
      "type": "string",
      "minLength": 50,
      "maxLength": 500,
      "description": "Detailed skill description with trigger contexts"
    },
    "skill_type": {
      "type": "string",
      "enum": ["main", "sub_skill", "agent", "tool", "validator", "middleware"],
      "description": "Type classification"
    },
    "tags": {
      "type": "array",
      "items": {"type": "string"},
      "minItems": 1,
      "description": "Discovery tags"
    },
    "dependencies": {
      "type": "array",
      "items": {"type": "string"},
      "description": "Required skill IDs"
    },
    "input_schema": {
      "type": "object",
      "description": "JSON Schema for input validation"
    },
    "output_schema": {
      "type": "object",
      "description": "JSON Schema for output validation"
    },
    "execution_handler": {
      "description": "Function implementing the skill logic"
    },
    "validation_handler": {
      "description": "Optional custom validation function"
    },
    "config": {
      "type": "object",
      "description": "Skill-specific configuration"
    },
    "status": {
      "type": "string",
      "enum": ["registered", "loaded", "active", "disabled", "error", "unloaded"],
      "description": "Current lifecycle status"
    }
  },
  "required": ["skill_id", "name", "version", "description", "skill_type"]
}

Registration Process

Skills are registered through the SkillRegistry class:

from core.registry import get_skill_registry

registry = get_skill_registry()

skill_id = registry.register(
    name="my_skill",
    version="1.0.0",
    description="My skill description",
    skill_type=SkillType.SUB_SKILL,
    tags=["analysis", "compliance"],
    dependencies=[],
    execution_handler=my_execution_function,
    validation_handler=my_validation_function
)

Skill Resolution

Skills can be resolved by:

1. Direct ID: registry.resolve(skill_id)
2. Name: registry.resolve("skill_name")
3. Tag: registry.resolve_by_tag("compliance")
4. Type: registry.resolve_by_type(SkillType.AGENT)

Resolution includes caching for performance and automatic fallback to dependencies.

Agent Routing System

Router Architecture

The AgentRouter uses chain-of-thought reasoning to determine optimal execution paths:

1. Task Analysis: Estimate complexity and required capabilities
2. Agent Selection: Choose agents based on capabilities and constraints
3. Mode Decision: Determine sequential, parallel, hybrid, or adaptive execution
4. Execution: Orchestrate agent chain with hooks and error handling
5. Result Synthesis: Combine agent outputs into final response

Routing Decision Schema

{
  "type": "object",
  "properties": {
    "selected_agents": {
      "type": "array",
      "items": {"type": "string"},
      "description": "Agent types in execution order"
    },
    "execution_mode": {
      "type": "string",
      "enum": ["sequential", "parallel", "hybrid", "adaptive"],
      "description": "How agents should be executed"
    },
    "reasoning": {
      "type": "string",
      "description": "Chain-of-thought explanation"
    },
    "confidence": {
      "type": "number",
      "minimum": 0.0,
      "maximum": 1.0,
      "description": "Confidence in routing decision"
    },
    "alternative_routes": {
      "type": "array",
      "description": "Fallback routing options"
    },
    "metadata": {
      "type": "object",
      "description": "Additional routing metadata"
    }
  }
}

Execution Modes

Sequential: Agents execute one after another, passing state forward.

• Use when: Dependencies between agents, limited parallel capacity, strict ordering required

Parallel: Agents execute simultaneously with independent contexts.

• Use when: No dependencies, sufficient capacity, independent operations

Hybrid: Mixed sequential and parallel execution.

• Gather requirements first → parallel collection/analysis → sequential synthesis
• Use when: Some agents can run in parallel while others require sequential execution

Adaptive: Mode switches based on execution results and failures.

• Start with hybrid, fall back to sequential if failures occur
• Use when: Uncertain dependencies, potential for runtime optimization

Hooks System

Hook Events

Standard hook events throughout the agent lifecycle:

| Event | When Emitted | Context Data | |-------|-------------|--------------| | ON_INIT | Agent starts | agent, context, correlationid | | ON_EXECUTE | Agent executes main logic | agent, inputdata, state | | ON_COMPLETE | Agent completes successfully | agent, result, executiontime | | ON_ERROR | Agent encounters error | agent, error, stacktrace | | ON_CLEANUP | Agent cleanup | agent, resourcesreleased | | ON_STATE_CHANGE | Agent state updates | agent, key, oldvalue, newvalue | | ON_GATE_CHECK | Quality gate checked | gatename, checkresult | | ON_GATE_FAIL | Quality gate fails | gatename, error, attempt |

Hook Registration

from hooks.system import hook, HookEvent

@hook(HookEvent.ON_INIT, priority=50)
async def my_init_handler(context: HookContext):
    print(f"Agent {context.agent_id} initializing with data: {context.data}")

Context Object

@dataclass
class HookContext:
    event: HookEvent
    timestamp: datetime
    data: Dict[str, Any]
    metadata: Dict[str, Any]
    agent_id: Optional[str]
    correlation_id: Optional[str]
    error: Optional[Exception]

Tool Definitions

Tool Schema

All tools conform to this structure:

{
  "type": "object",
  "properties": {
    "name": {
      "type": "string",
      "description": "Unique tool identifier"
    },
    "description": {
      "type": "string",
      "description": "What the tool does"
    },
    "category": {
      "type": "string",
      "enum": ["web", "knowledge", "analysis", "system", "validation", "compliance"],
      "description": "Tool category for organization"
    },
    "input_schema": {
      "type": "object",
      "description": "JSON Schema for tool input validation"
    },
    "config": {
      "type": "object",
      "properties": {
        "timeout_ms": {"type": "integer"},
        "max_retries": {"type": "integer"},
        "enable_cache": {"type": "boolean"},
        "rate_limit_per_minute": {"type": "integer"}
      }
    }
  }
}

Tool Execution Flow

1. Input Validation: Schema validation if enabled
2. Rate Limit Check: Verify within rate limits
3. Cache Lookup: Return cached result if available and valid
4. Execution with Retry: Execute with automatic retry and backoff
5. Result Caching: Store successful results
6. Return: ToolResult with status, data, metadata

Built-in Tools

• web_search: Search web for compliance information
• web_fetch: Fetch and parse web pages
• knowledge_query: Query SECOND-KNOWLEDGE-BRAIN.md
• compliance_analysis: Analyze ingredients for compliance
• validation: Validate outputs against quality gates

Error Handling

Error Classification

Errors are automatically classified into categories:

• Network: Connection, timeout, DNS failures
• Validation: Schema validation, type errors
• Execution: Runtime errors, exceptions
• Resource: File not found, out of memory
• Authentication: Auth failures
• Authorization: Permission denied
• Dependency: Missing dependencies
• Business Logic: Domain-specific errors
• Unknown: Unclassified errors

Recovery Strategies

@dataclass
class RecoveryStrategy:
    action: RecoveryAction  # RETRY, FALLBACK, SKIP, ABORT, DEGRADE
    max_attempts: int
    delay_ms: int
    exponential_backoff: bool
    fallback_handler: Optional[Callable]
    circuit_breaker_threshold: int

Circuit Breakers

Prevent cascading failures by stopping calls to failing services:

@with_circuit_breaker(key="external_api", failure_threshold=5, timeout_ms=60000)
async def call_external_service():
    # Will stop executing after 5 failures within timeout window
    pass

Configuration Management

Configuration Schema

{
  "llm": {
    "provider": "anthropic|openai|cohere|huggingface",
    "model": "string",
    "temperature": "number (0-2)",
    "max_tokens": "integer (1-200000)",
    "timeout_ms": "integer",
    "max_retries": "integer",
    "enable_caching": "boolean"
  },
  "knowledge": {
    "update_interval_hours": "integer (1-720)",
    "max_entries": "integer",
    "dedup_enabled": "boolean",
    "scoring_weights": {"recency": 0.4, "relevance": 0.4, "citation_count": 0.2}
  },
  "compliance": {
    "default_standard": "halal|kosher|both",
    "strict_mode": "boolean",
    "enable_cross_contamination_check": "boolean",
    "enable_traceability_check": "boolean"
  },
  "features": {
    "enable_agent_router": "boolean",
    "enable_hooks": "boolean",
    "enable_caching": "boolean",
    "enable_metrics": "boolean",
    "max_parallel_agents": "integer (1-20)"
  },
  "logging": {
    "level": "DEBUG|INFO|WARNING|ERROR|CRITICAL",
    "format": "string",
    "file": "string (path)",
    "enable_console": "boolean",
    "enable_structured": "boolean"
  }
}

Loading Configuration

from config.settings import get_config

# Load from environment variables (automatic)
config = get_config()

# Or load from file
from config.settings import SystemConfig
config = SystemConfig.from_file("config/production.json")

Quality Gates

Universal Gates (U1-U6)

Apply to all outputs:

• U1: ≥3 sources cited, ≥1 academic/authoritative
• U2: Disclosure/limitations before recommendation
• U3: Evidence hierarchy stated per source (Tier 1–4)
• U4: Language matches user preference
• U5: Output uses declared template (all sections)
• U6: Every claim traceable to ≥1 source or flagged

Domain Gates (G1-G4)

Apply to compliance analysis:

• G1: Ingredient scan complete (E-numbers/enzymes/gelatin)
• G2: Halal & Kosher criteria both checked
• G3: Cross-contamination & traceability assessed
• G4: Certification/audit process mapped

Gate Enforcement

Gates are checked in sequence with auto-fix attempts:

1. Check gate condition
2. On failure: execute auto-fix procedure
3. Re-check gate
4. After 2 failures: emit limitation notice
5. Continue to next gate

Graceful Degradation

Degradation Levels

| Level | Condition | Behavior | |-------|-----------|----------| | 0 | All primary sources reachable | Full evidenced analysis | | 1 | Some primary sources fail | Use secondary sources; flag each substitution | | 2 | Most live sources fail | Knowledge base only; flag as historical context | | 3 | Required input missing/stale | Proceed with available; mark DATA UNAVAILABLE | | 4 | All sources + knowledge base fail | Emit DATA UNAVAILABLE notice |

Degradation Banner

---
⚠️ LIMITATION NOTICE
This output was generated with reduced data availability (Level [0-4]).
Cross-check with current data before acting on it. Substituted/missing sources
are flagged inline.
---

Knowledge Pipeline

Crawl Configuration

KNOWLEDGE_CONFIG = {
    "KEYWORDS": ["halal", "kosher", "compliance", "certification", "ingredient"],
    "ARXIV_CATEGORIES": ["q-bio", "cs.AI"],
    "RSS_FEEDS": [
        "https://www.isaaa.org/kc/cropbiotechnupdate/feed/default.aspx",
        "https://www.foodstandards.gov.au/feed.xml"
    ],
    "AUTHORITATIVE_DOCS": [
        "https://www.isaaa.org/gmapprovaldatabase/default.asp",
        "https://www.codexalimentarius.org/standards/cxs/"
    ],
    "SCORING_WEIGHTS": {"recency": 0.4, "relevance": 0.4, "citation_count": 0.2}
}

Deduplication

Entries are deduplicated using SHA256 hashes of DOI/URL to prevent duplicates.

Scoring

Composite score = recencyscore × 0.4 + relevancescore × 0.4 + citation_score × 0.2

Testing

Test Scenarios

Located in tests/test-scenarios.md:

1. Standard Analysis: Full compliance check with ingredient list
2. Minimal Input: Analysis with minimal user input
3. Comparison: Compare two products/standards
4. Risk/Conflict: Handle conflicting information
5. Degraded Mode: Test graceful degradation behavior

Running Tests

# Run all test scenarios
python tools/run_test_scenarios.py --all

# Run specific scenario
python tools/run_test_scenarios.py --scenario standard-analysis

# Run knowledge updater tests
python tools/test_knowledge_updater.py

File Organization

Required Files (8-File Contract)

• CLAUDE.md — Skill identity and configuration
• PROJECT-detail.md — Full technical specification
• PROJECT-DEVELOPMENT-PHASE-TRACKING.md — Build roadmap
• README.md — Public-facing documentation
• skills/main.md — Primary harness orchestrator
• SECOND-KNOWLEDGE-BRAIN.md — Self-improving knowledge base
• tools/knowledge_updater.py — Knowledge crawl pipeline
• tests/test-scenarios.md — Test scenario definitions

Additional Files

• config/default.json — Default configuration
• config/settings.py — Configuration management
• hooks/system.py — Lifecycle hooks
• core/registry.py — Skill registry
• core/router.py — Agent router
• core/tools.py — Tool definitions
• core/errors.py — Error handling
• SKILL.md — This file

Usage Examples

Basic Usage

from core.router import execute_task

result = await execute_task(
    input_data={
        "ingredients": ["E120", "gelatin", "enzymes"],
        "standard": "both"
    },
    requirements={
        "need_evidence": True,
        "need_analysis": True,
        "depth": "standard"
    }
)

Adv

…

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/halal-kosher-compliance-monitor-agent-skill
• License: MIT

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