Mountain Landslide Early Warning Agent Skill

SKILL.md — Skill Registry Documentation

> Single source of truth for how skills are registered, resolved, executed, > and validated in the mountain-landslide-early-warning runtime. This > document is the contract between the markdown harness (skills/*.md), the > Python skill registry (src/landslide_ews/skills/), and any external LLM > executor that drives the qualitative skills.

---

1. Overview

The skill system follows a skill-registry pattern with a chain-of-thought router. Skills are declarative descriptors (metadata + I/O schemas + optional handler) registered in a single registry. The router scores skills against an intent, emits a rationale, and returns an ordered dispatch plan. The orchestrator executes the plan, invoking programmatic handlers for the deterministic core and an optional LLM executor for the markdown-only qualitative skills.

This is intentionally LLM-agnostic: the deterministic core (slope stability, rainfall thresholds, knowledge retrieval, language detection) runs without any model, so the pipeline is fully testable in CI and never fabricates outputs when an LLM is unavailable.

 intent ─► language_detect (U4) ─► ChainOfThoughtRouter ─► Orchestrator
   │                                                            │
   │                                                            ▼
   │              ┌──── Skill Registry (register / resolve) ────┐
   │              │  sub-gather-requirements  (intake)            │
   │              │  sub-evidence-collector  (evidence)           │
   │              │  sub-core-analysis       (analysis)          │
   │              │  sub-knowledge-updater   (knowledge)         │
   │              │  sub-advisor             (synthesis)         │
   │              │  mountain-landslide-early-warning (gate)     │
   │              └─────────────────────────────────────────────┘
   │                                                            │
   └── Tools (compute_slope_stability, rainfall_threshold, knowledge_query) ◄┘
   │
   └── Hooks + Event Bus (lifecycle, gates, degradation, compaction)
   │
   └── ContextWindow (bounded, deterministic compaction)
   │
   ▼
RunResult { state, gate_checklist, report }

---

2. Registration

2.1 From markdown (skills/*.md)

SkillLoader scans skills/*.md, parses the YAML frontmatter (name, description, optional stage, version), and derives the rest from a data-driven blueprint keyed by skill name (see src/landslide_ews/skills/loader.py → _SKILL_BLUEPRINT). Each blueprint declares: stage, keywords (router relevance), tools (allowed tools), quality_gates, and the inputs_required / outputs_required field lists that materialize the I/O schemas.

To add a new skill: drop a skills/sub-.md with frontmatter and add a blueprint entry. No orchestrator code changes required.

2.2 Programmatically

from landslide_ews.skills import SkillDescriptor, SkillInputSchema, SkillOutputSchema, SkillStage

descriptor = SkillDescriptor(
    name="sub-region-threshold",
    description="Regional rainfall threshold calibration",
    stage=SkillStage.ANALYSIS,
    inputs=SkillInputSchema(required=["region"]),
    outputs=SkillOutputSchema(required=["alpha", "beta"]),
    tools=["rainfall_threshold"],
    keywords=["regional threshold", "calibration"],
)
registry.register(descriptor, handler=my_handler)

2.3 Validation at registration

SkillRegistry._validate_descriptor enforces:

• non-empty name and description;
• at least one keyword for router relevance (the main harness gate is exempt

because it is resolved by name, not keywords).

Duplicate registration raises unless replace=True.

---

3. Resolution

| Method | Returns | Cost | |--------|---------|------| | registry.get(name) | SkillDescriptor | O(1) | | registry.by_stage(stage) | list of descriptors in that stage | O(n) | | registry.has(name) | bool | O(1) | | registry.all() / names() | full catalog | O(n) | | registry.describe() | serializable catalog (for docs/SKILL.md) | O(n) |

Unknown names raise SkillNotFoundError.

---

4. Execution

registry.run(name, payload, tool_catalog):

1. Validate inputs against descriptor.inputs (missing-required →

SkillExecutionError).

1. Invoke handler: descriptor.handler(payload, tool_catalog).
2. Validate outputs against descriptor.outputs (missing-required →

SkillExecutionError).

Skills with no Python handler are markdown-only: they are executed by an llm_executor plugged into the orchestrator, or by the deterministic fallback handlers in src/landslide_ews/runtime/skill_handlers.py.

4.1 The LLM executor seam

LlmExecutor = Callable[[SkillDescriptor, dict, ContextWindow, ToolCatalog], dict]
orchestrator = Orchestrator(llm_executor=my_llm_executor)

When supplied, the LLM executor is called for every markdown-only skill in the plan. Its return value must conform to that skill's output schema.

4.2 Deterministic fallbacks

When no LLM executor is supplied, these programmatic handlers run so the pipeline is never empty (see runtime/skill_handlers.py):

| Skill | Fallback behaviour | |-------|--------------------| | sub-gather-requirements | regex-parse the user message; flag DATA UNAVAILABLE for missing fields | | sub-evidence-collector | surface configured authoritative docs; flag live data DATA UNAVAILABLE | | sub-core-analysis | invoke compute_slope_stability + rainfall_threshold tools | | sub-knowledge-updater | invoke knowledge_query tool against SECOND-KNOWLEDGE-BRAIN.md | | sub-advisor | rule-based verdict from the most-severe warning level; full disclosure |

All fallbacks surface gaps as explicit limitation fields rather than fabricating values.

---

5. Validation (I/O JSON schemas)

Each skill carries a JSON-schema-like input/output contract (SkillInputSchema / SkillOutputSchema). Canonical schemas are also published as standalone JSON Schema files in assets/schemas/:

| Schema file | Used by | |-------------|---------| | requirements.schema.json | sub-gather-requirements output | | evidence_bundle.schema.json | sub-evidence-collector output | | analysis_package.schema.json | sub-core-analysis output | | advisor_conclusion.schema.json | sub-advisor output | | skill_descriptor.schema.json | SkillDescriptor.to_dict() | | tool_descriptor.schema.json | Tool.describe() | | run_result.schema.json | Orchestrator RunResult |

The runtime validates required-field presence without an external JSON Schema library (dependency-free). External validators (jsonschema, optional) can additionally validate against the assets/schemas/*.schema.json files.

---

6. Routing (chain-of-thought)

ChainOfThoughtRouter.route(intent):

1. Tokenize the intent.
2. Score every registered skill: score = stage_prior + min(keyword_hits, 3.0).
3. Choose skills within an epsilon of the top score + all positive scorers.
4. Order by the default pipeline stage order (intake → evidence → analysis →

knowledge → synthesis → quality_gate; utilities last).

1. Emit a textual rationale (toggleable via feature_flags.router_cot_rationale).
2. On zero overlap, fall back to the full default pipeline + a limitation flag.

RoutingDecision.to_dict() returns a serializable, explainable decision.

---

7. Hooks & events

Hooks are callables registered against event names; the emitter is ordered and failure-isolated (one hook raising never breaks the orchestrator — it is logged and skipped). Default hooks (see hooks/lifecycle.py) wire structured logging and republish to the EventBus (append-only audit trail).

| Event | Fired when | |-------|-----------| | run.started / run.completed / run.failed | orchestrator lifecycle | | skill.invoked / skill.completed / skill.failed | per-skill lifecycle | | gate.failed | a quality gate fails (before auto-fix retry) | | degradation.escalated | degradation level increases | | context.compacted | the context window compacted | | language.detected | Pre-Flight language classification |

EventName enum + EventBus provide typed, append-only event history for replay/inspection.

---

8. Tools

Tools are declarative (name, description, JSON-schema-like parameters, handler). Validation raises ToolValidationError on bad input; handler failures are wrapped in ToolError so the orchestrator isolates them.

| Tool | Schema | Backs | |------|--------|-------| | compute_slope_stability | assets/schemas/... (in tool descriptor) | gates G1/G2/G4, sub-core-analysis | | rainfall_threshold | … | gate G2, sub-core-analysis | | knowledge_query | … | gates U1/U3, sub-knowledge-updater | | language_detect | … | gate U4, Pre-Flight |

ToolCatalog.describe_all() returns the full machine-readable tool catalog.

---

9. Quality gates

Enforced by the orchestrator after execution with a 2-retry auto-fix budget (runtime.max_retries). See skills/main.md for the gate table.

• U1 ≥3 sources cited, ≥1 academic/authoritative.
• U2 disclosure before recommendation.
• U3 evidence hierarchy (Tier 1-4) per source.
• U4 language matches detected preference.
• U5 output uses the declared template.
• U6 every claim traceable to a source or flagged.
• G1 slope stability assessed (FS).
• G2 rainfall thresholds set.
• G3 sensors & monitoring specified.
• G4 warning levels & evacuation defined.

---

10. Status model

| Status | Meaning | |--------|---------| | completed | all gates pass, no limitations | | degraded | ran to completion with explicit limitations / gate failures | | failed | unrecoverable errors at the highest degradation level |

degradation_level (0-4) tracks the graceful-degradation ladder.

---

11. Programmatic quick start

from landslide_ews.runtime import Orchestrator, RunRequest

orch = Orchestrator()
result = orch.run(RunRequest(
    intent="Analyze slope stability for a 45-degree slope",
    user_message="Analyze slope stability for a 45-degree slope ...",
    metadata={"slope_data": {"cohesion_kpa": 15, "friction_angle_deg": 25,
                             "unit_weight_knm3": 18, "soil_thickness_m": 2,
                             "slope_angle_deg": 45, "rainfall_duration_h": 24,
                             "rainfall_intensity_mmh": 5}},
))
print(result.state.status.value, result.gate_checklist, result.report["verdict"])

CLI: python scripts/run_harness.py "..." --slope '{...}'.

---

12. File map

skills/                              # markdown harness (human/LLM-facing)
  main.md
  sub-gather-requirements.md
  sub-evidence-collector.md
  sub-core-analysis.md
  sub-knowledge-updater.md
  sub-advisor.md
src/landslide_ews/                   # Python skill-registry runtime
  skills/   descriptor.py loader.py registry.py router.py
  hooks/    base.py events.py lifecycle.py
  tools/    base.py slope_stability.py rainfall_threshold.py knowledge_query.py catalog.py
  runtime/  state.py orchestrator.py skill_handlers.py
  config.py context.py errors.py logging.py __init__.py
config/      default.yaml config.py __init__.py README.md
references/  prompt_templates/ domain_knowledge/ grounding/
assets/      schemas/ diagrams/
scripts/     setup_project.py seed_knowledge_base.py ingest_sources.py run_harness.py
tests/       test_*.py
SKILL.md     (this file)
SECOND-KNOWLEDGE-BRAIN.md   (living knowledge base)
tools/knowledge_updater.py  (crawl pipeline, legacy CLI)

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/mountain-landslide-early-warning-agent-skill
• License: MIT

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