Craft Beer Production Optimization

SKILL.md - craft-beer-production-optimization

Skill Registry Documentation

This document is the canonical, machine-correlated description of how the craft-beer-production-optimization skill is registered, resolved, executed, and validated. It is the contract between the skill registry (config/skill_registration.json), the harness (skills/main.md), the tool layer (tools/), and the lifecycle hooks (skills/hooks/).

Overview

• Skill Name: craft-beer-production-optimization
• Version: 1.0.0
• Domain: Craft Brewing Process & Quality Engineering
• Purpose: Deliver structured, evidence-backed optimization of the

craft-brewing process (mash, hop, ferment, quality) for a target beer style, grounded in brewing science (BJCP, ASBC, MBAA, IBD) and academic research, with risk/limitation-disclosed outputs and a continuously-updating knowledge pipeline.

Skill Registration

Registration File

Skills are registered in config/skill_registration.json:

{
  "schema_version": "1.0.0",
  "registry": {
    "skills": [
      {
        "id": "craft-beer-production-optimization",
        "name": "Craft Beer Production Process Optimization",
        "version": "1.0.0",
        "description": "...",
        "domain": "Craft Brewing Process & Quality Engineering",
        "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 incl. trigger phrases | | domain | string | Yes | Domain of expertise | | triggers | array | Yes | Phrases/contexts that trigger the skill | | skills | array | Yes | Sub-skills this skill orchestrates | | quality_gates | object | Yes | Quality gate definitions | | tools | array | Yes | Tools used by the skill | | degradation_levels | array | No | Graceful-degradation ladder |

Trigger Phrases

The skill triggers on phrases like:

• "craft beer production optimization"
• "brewing process analysis"
• "mash enzymatic conversion"
• "hopping IBU bitterness aroma"
• "yeast fermentation attenuation diacetyl"
• "beer style OG FG ABV IBU SRM"
• "off flavor beer quality"
• "recipe optimization craft brewing"
• "ASBC brewing quality control"
• "toi uu hoa san xuat bia thu cong"

Skill Resolution

Resolution Process

When a user query is received, the resolver:

1. Analyzes the query for trigger phrases and domain keywords.
2. Matches against registered skills using fuzzy/keyword matching.
3. Scores candidates by trigger overlap (70%), domain relevance (20%),

context similarity (10%).

1. Selects the highest-scoring skill if score >= 0.6.
2. Loads the skill definition from config/skill_registration.json.
3. Initializes the execution context via the pre-execution hooks.

Resolution Algorithm

from pathlib import Path
import json

THRESHOLD = 0.6

def resolve_skill(user_query: str, registry_path: str = "config/skill_registration.json") -> str | None:
    """Resolve a user query to a skill id, or None if no skill matches."""
    registry = json.loads(Path(registry_path).read_text(encoding="utf-8"))
    q = user_query.lower()
    best_id, best_score = None, 0.0
    for skill in registry["registry"]["skills"]:
        triggers = [t.lower() for t in skill.get("triggers", [])]
        if not triggers:
            continue
        overlap = sum(1 for t in triggers if t in q) / len(triggers)
        domain = 1.0 if skill.get("domain", "").lower().split()[0] in q else 0.0
        score = overlap * 0.7 + domain * 0.2 + 0.1  # context placeholder
        if score >= THRESHOLD and score > best_score:
            best_id, best_score = skill["id"], score
    return best_id

Skill Execution

Execution Flow

User Query
    |
[Pre-Flight Checks]   (skills/hooks/pre_execution.py)
    |- Language detection (vi/en)
    |- Input validation (per-step schema)
    |- Degradation + quality-gate init
    |
[Step 1: sub-gather-requirements]  -> structured requirements
    |
[Step 2: sub-evidence-collector]   -> evidence bundle (fallback chain)
    |
[Step 3: sub-core-analysis]        -> style target + mash + hopping + ferm + QC
    |
[Step 4: sub-knowledge-updater]    -> academic citations + gaps
    |
[Step 5: sub-advisor]              -> risk-disclosed conclusion
    |
[Final Quality Gate Review]        (skills/main.md + post_execution.py)
    |- Check U1-U6 + G1-G4
    |- Auto-fix (2 retries max)
    |- Limitation banner on degradation
    |
[Output Formatting] -> report -> user

Execution Context

{
    "execution_id": str,
    "start_time": str (ISO),
    "current_step": str,
    "detected_language": "en" | "vi",
    "output_language": "en" | "vi",
    "inputs": dict,
    "results": dict,
    "validation_passed": bool,
    "degradation_info": {"level": 0-4, "source_failures": [], ...},
    "quality_gates": {"universal": {...}, "domain": {...}, "all_passed": bool},
    "execution_metadata": dict,
    "limitation_banner": str | None,
}

Sub-Skill Invocation

from tools.base_tool import SkillTool

def invoke_sub_skill(skill_id: str, context: dict) -> dict:
    """Resolve and load a sub-skill definition, then run the harness step."""
    loader = SkillTool()
    spec = loader.execute(skill_id=skill_id, parameters=context.get("inputs", {}))
    # The harness (main.md) drives the LLM execution of spec["prompt"];
    # results are written back to context["results"].
    context["results"][skill_id] = spec
    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 | Style target set (OG/FG/ABV/IBU/SRM) | Set from BJCP/category | Required for recipe | | G2 | Mash enzymatic conversion & IBU computed | Compute mash rest + Tinseth IBU | Required for process | | G3 | Fermentation & quality controlled | Add ferm schedule + off-flavor checks | Required for batch | | G4 | ASBC quality methods cited | Cite ASBC methods | Required for quality |

Gate Enforcement Logic

MAX_ATTEMPTS = 2

def enforce_quality_gate(gate_id, context, check_fn, auto_fix_fn):
    for attempt in range(MAX_ATTEMPTS):
        if check_fn(context):
            context["quality_gates"][bucket(gate_id)][gate_id]["passed"] = True
            return True
        context = auto_fix_fn(context)
        context["quality_gates"][bucket(gate_id)][gate_id]["attempts"] += 1
    context["quality_gates"][bucket(gate_id)][gate_id]["passed"] = False
    return False

Input/Output Schemas

Schemas live in config/schemas/:

• input_schemas.json - requirements_input, evidence_collector_input,

core_analysis_input, knowledge_query_input, advisor_input.

• output_schemas.json - requirements_output, evidence_collector_output,

core_analysis_output, knowledge_query_output, advisor_output, final_report.

Example (final report, abridged):

{
  "type": "object",
  "properties": {
    "metadata": {"type": "object", "properties": {"date": {}, "analyst": {}, "version": {}, "language": {}, "domain": {}}},
    "executive_summary": {"type": "string"},
    "inputs_and_scope": {"type": "object"},
    "evidence_collected": {"type": "object"},
    "analysis_scorecard": {"type": "object"},
    "action_plan": {"type": "object"},
    "academic_evidence": {"type": "array"},
    "disclosure": {"type": "string"},
    "recommendation": {"type": "object"},
    "post_execution_checklist": {"type": "object"}
  },
  "required": ["metadata", "executive_summary", "disclosure", "post_execution_checklist"]
}

Validation is performed by tools/utils/validation.py (SchemaValidator, InputValidator, OutputValidator, DataQualityChecker).

Graceful Degradation

| Level | Condition | Behavior | |-------|-----------|----------| | 0 | All sources reachable | Full evidenced analysis | | 1 | Some sources fail | Use secondary sources; flag each | | 2 | Most sources fail | Knowledge base only; flag historical | | 3 | Input missing/stale | Proceed with available; mark DATA UNAVAILABLE | | 4 | All sources + KB fail | Emit UNAVAILABLE notice; do NOT fabricate |

Implementation: skills/hooks/error_recovery.py (FallbackChain, GracefulDegradation) + skills/hooks/post_execution.py (generate_limitation_banner).

Tool Integration

Tools are registered in config/tool_registry.json and implemented in tools/base_tool.py:

from tools.base_tool import get_tool

search = get_tool("WebSearch")
result = search.execute_with_retry(query="Tinseth IBU formula", max_retries=3)

WebSearch performs a real DuckDuckgo Lite search; WebFetch performs an HTTP GET; Read/Write/Bash are real filesystem/subprocess tools; Skill resolves and loads sub-skill definitions from the registry.

Error Handling

| Type | Recovery | Limit | |------|----------|-------| | sourcetimeout | Retry + backoff, fallback chain | 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 UNAVAILABLE notice | n/a |

Entry point: skills/hooks/error_recovery.py::on_error.

Performance Monitoring

from tools.utils.metrics import get_metrics_collector, PerformanceTimer

collector = get_metrics_collector()
with PerformanceTimer(collector, "core_analysis"):
    run_core_analysis(...)
print(collector.get_summary())

Metrics tracked: execution time, token usage, quality-gate pass rate, source success rate, degradation distribution.

Configuration

• config/settings.json - application, execution, knowledge-base, data-source,

quality-gate, logging, metrics, caching, state, features, limits.

• config/logging.yaml - console + rotating file + error file + JSON handlers.

Extension Points

• New sub-skill: add to config/skill_registration.json, create

skills/sub-.md, add schemas to config/schemas/, wire into skills/main.md.

• New quality gate: add to quality_gates in the registry, implement the

check/auto-fix in the relevant hook.

• New tool: add to config/tool_registry.json, subclass BaseTool in

tools/base_tool.py, register in TOOL_REGISTRY.

Testing

python tools/test_knowledge_updater.py        # unit tests
python tools/run_test_scenarios.py            # structural + content validator
python tools/validate_project.py              # 8-File Contract
python tools/final_validation.py              # production readiness
python tests/integration/test_full_pipeline.py

Troubleshooting

See references/troubleshooting.md for common issues, error-handling strategies, performance tuning, and debugging.

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
• Knowledge Base: SECOND-KNOWLEDGE-BRAIN.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/craft-beer-production-optimization-agent-skill
• License: MIT

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