Call this to discover what iaiops can do before choosing a protocol/tool.
Lists implemented protocols (OPC-UA incl. HDA, Modbus, S7comm, Mitsubishi MC,
MTConnect, MQTT/Sparkplug B full-decode, EtherNet/IP Logix) and the EtherCAT
roadmap stub, plus cross-protocol analytics (OEE/downtime, asset inventory,
CoV), each with its read/write tools and the endpoint params it needs.

Also reports whether this server runs under the no-egress gate, so a model is
TOLD the posture instead of having to infer it from tools it cannot see.
Read/write authorisation is NOT a server posture here — it is the caller's
decision; every call (read or write, MCP or CLI) is audited.

Returns dict: {tool, posture, implemented_protocols:[...], roadmap_stubs:[...],
    protocols:[{protocol, status, library, transport, auth, read_tools,
    write_tools, params}], diagnostics:[...], analytics:[...], tool_counts,
    safety, write_note, no_egress_mode, no_egress_note}.

Example: protocols_supported().

Classifies tag node-ids against warn/alarm thresholds. Returns
ok/warn/alarm/unknown counts plus the offending tags. Thresholds
come from config tags, or per-ref overrides in ``thresholds``.

Args:
    endpoint: Endpoint name from config.
    node_ids: Tag node ids to evaluate; omit to use configured tags.
    thresholds: Optional {ref: {warn_high, alarm_high, warn_low, alarm_low}}.

Samples a node over a bounded window and flags statistical outliers.
Computes mean/stddev/min/max and flags samples outside mean ± sigma*stddev.
Simple statistics only — no ML, no persisted model.

Args:
    node_id: The OPC-UA node id to scan.
    endpoint: Endpoint name from config.
    samples: Max samples (capped server-side).
    interval_ms: Delay between samples in milliseconds.
    sigma: Outlier band width in standard deviations.

Probes connect → read(ref) → freshness → variance and returns a verdict with
per-hop detail and a recommended action. The #1 OT triage: distinguishes
"cannot connect" (network/PLC down) from "comms OK but value stale"
(upstream/field/source) from "good status but flatline" (sensor stuck).

Args:
    endpoint: Endpoint name from config (any protocol).
    ref: Tag/node/address/device to read (OPC-UA node id, Modbus address,
        S7 address string, MELSEC device). Omit to test connectivity only.
    freshness_threshold_s: Max value-age (seconds) before 'stale' (default 60).
    series: Optional injected samples (scalars or {value,timestamp}) for
        flatline/variance reasoning when a live historian is out of reach.
    flatline_eps: Spread at/below which a series counts as flatline.

Returns dict: {verdict ('cannot_connect'|'comms_ok_value_unreadable'|
    'comms_ok_bad_quality'|'comms_ok_value_stale'|'comms_ok_flatline'|
    'healthy'), diagnosis, recommended_action, hops:[{hop, ok, detail}]}.

Example: diagnose_dataflow(endpoint="line1", ref="ns=2;i=5", freshness_threshold_s=30).

Pure analysis over an injected sample series — no live historian needed.

Args:
    series: Samples — scalars or {value, timestamp (ISO-8601), quality|good}.
    gap_threshold_s: Time gap (seconds) between consecutive samples that counts
        as a data gap (default 60).
    flatline_eps: Spread at/below which the series counts as flatline.

Returns dict: {samples, numeric_samples, bad_quality_count, flatline (bool),
    gap_count, gaps:[{after, gap_seconds}], stdev,
    verdict ('ok'|'degraded'|'gappy'|'flatline'|'bad_tag')}.

Example: historian_health(series=[{"value":10,"timestamp":"2026-06-28T10:00:00Z"}, ...]).

Args:
    events: Alarm/condition events — {source, timestamp (ISO-8601), priority?,
        state? (ACTIVE/RTN/ACK)}.
    window_minutes: Analysis window; omitted → inferred from event timestamps.
    chatter_window_s: A source with >=3 transitions inside this window chatters.
    standing_s: An alarm active longer than this is 'standing/stale' (default 24h).
    top_n: How many top offenders to return.

Returns dict: {event_count, window_minutes, alarms_per_hour,
    isa_18_2:{ok_max:6, manageable_max:12, flood_min:30},
    flood_verdict ('ok'|'manageable'|'over_target'|'flood'),
    priority_distribution, pareto_sources_for_80pct, top_offenders:[{source,
    count, share_pct, chattering, standing}], chattering:[...], standing:[...]}.

Example: alarm_bad_actors(events=[{"source":"FIC101","timestamp":"...",
    "priority":"high"}, ...]).

Args:
    tags: Per-tag dicts — {ref, label?, samples:[scalars or {value, good|quality}],
        warn_high?, alarm_high?, warn_low?, alarm_low?}.
    thresholds: Optional {ref: {warn_high, alarm_high, warn_low, alarm_low}} override.

Returns dict: {evaluated, overall ('ok'|'warn'|'alarm'), offender_count,
    offenders:[{ref, label, samples, latest, flags:[...], anomaly_count,
    severity (0..3)}], results:[...]}. Flags include bad_quality, flatline,
    out_of_range_warn/alarm, statistical_anomaly.

Example: tag_health(tags=[{"ref":"ns=2;i=5","samples":[70,71,70,99]}]).

Detects dropped notifications (sequence gaps), duplicates / out-of-order, a high
republish-rejection rate, and overloaded channels — the classic Kepware
"too many tags on one channel → republish/queue-flush dropouts" fault.

Args:
    sequence: Sequence numbers actually received, in arrival order.
    republish_requested: How many republish requests were made.
    republish_rejected: How many were rejected (server couldn't keep up).
    tags_per_channel: {channel/endpoint: tag_count} — flags channels over the max.
    max_tags_per_channel: Density above which a channel is flagged (default 5000).
    wrap_at: Modulus for rolling counters (e.g. 256 for Sparkplug B seq); omit
        for monotonic OPC-UA counters.

Returns dict: {received, missed_count, duplicate_count, out_of_order_count,
    republish_requested, republish_rejected, republish_reject_rate,
    overloaded_channels:[{channel, tags}], max_tags_per_channel,
    verdict ('ok'|'reordered'|'lossy'|'overloaded'), recommendation}.

Example: subscription_health(sequence=[1,2,4,5], tags_per_channel={"ch1":7000}).

Correlates whatever evidence you supply around a downtime/incident window —
alarm events, tag samples, a diagnose_dataflow verdict, a machine-state series —
ranks candidate root causes, and cites the REAL signals behind each. Read-first:
it proposes a human-approved, undoable (MOC-gated) action but executes nothing.
Anti-hallucination: only signals present in the input are cited; thin evidence
downgrades to 'insufficient_evidence' with a 'recommended_next_data' list rather
than a confident guess. Confidence combines independent, time-correlated evidence
(signals BEFORE onset outweigh signals during it).

Args:
    window: {start (ISO-8601), end? (ISO-8601), asset?, category?}. If 'end' is
        omitted but state_series is given, the first running→stopped span bounds it.
    alarms: Alarm/condition events — {source, timestamp, message?, priority?, state?}.
    tags: Per-tag samples — {ref, samples:[scalars or {value, good|quality}],
        warn_high?, alarm_high?, ...} (scored via tag_health).
    dataflow: A diagnose_dataflow result dict (its 'verdict' localizes comms vs field).
    state_series: {timestamp, state} samples to bound the window if 'end' is absent.
    lead_window_s: How far before onset a signal may sit and still count as a cause
        (default 300s); signals after onset are treated as consequences.
    cause_weights: Optional per-site {cause: multiplier} override (e.g. from
        learn_cause_weights) — scales each cause's evidence (1.0 = neutral
        default) before the noisy-OR. Unknown causes / non-numeric weights are
        rejected; values are clamped. Omit for the shipped default weighting.
    include_graph: When true, also return a 'graph' block — the SAME verdict
        re-projected as a causal graph {nodes, edges, mermaid, meta} (signal →
        cause → downtime) for a frontend/Grafana. Pure re-shape: signal→cause
        edge weights are the evidence contribution scores, cause→symptom edge
        weights are the hypothesis confidences — no new reasoning. Omit for the
        flat verdict only (default).

When a per-site 'historian:' block is configured (~/.iaiops/config.yaml, A7),
the 2h pre-incident window is additionally pulled from that reader and scored
as historian trend evidence — cited with its source ('historian:<name>'),
window, and sample count. Without the config, behaviour is unchanged.

Returns dict: {window, verdict ('root_cause_identified'|'multiple_candidates'|
    'insufficient_evidence'), primary_cause, hypotheses:[{cause, confidence (0..1),
    confidence_band, evidence:[{signal, ref, at?, lead_time_s?, detail, weight}],
    recommended_action}], evidence_summary, recommended_next_data?,
    anti_hallucination, graph? (when include_graph): {nodes:[{id, kind
    (signal|cause|symptom), label, score, ...}], edges:[{from, to, weight,
    relation (supports|attributed_to)}], mermaid, meta}}.

Example: downtime_root_cause(window={"start":"2026-06-28T10:00:00Z","asset":"line1"},
    alarms=[{"source":"M1_DRIVE","timestamp":"2026-06-28T09:59:50Z",
             "message":"motor overload trip"}], dataflow={"verdict":"healthy"}).

Same advisory, read-only, evidence-cited contract as downtime_root_cause — but
instead of hand-injecting evidence you give an endpoint + incident window and it
pulls the evidence itself: a cross-protocol diagnose_dataflow probe, a short
sampled series per ref (so flatline/bad-quality/anomaly surface via tag_health),
and active OPC-UA conditions. Light read load; non-destructive; nothing executed.
The gathered bundle is echoed under 'collected_evidence' (no hidden inputs).

Args:
    endpoint: Endpoint name from config (any protocol). Omit for the default.
    window: {start (ISO-8601), end?, asset?, category?, freshness_threshold_s?}.
    refs: Tags/nodes/addresses to sample for this incident (first is also the
        diagnose_dataflow target). Capped at 20.
    sample_count: Reads per ref to build its series (1..60, default 8).
    interval_ms: Delay between reads (>=50ms, default 200).
    include_alarms: Surface active OPC-UA conditions as alarm evidence (OPC-UA only).
    lead_window_s: Causal lead window before onset (default 300s).
    include_graph: When true, also return the 'graph' block (same {nodes, edges,
        mermaid, meta} causal-graph re-projection as downtime_root_cause). Pure
        re-shape of the verdict; no new reasoning. Omit for the flat verdict.

Returns dict: same shape as downtime_root_cause plus 'collected_evidence'
    {endpoint, protocol, refs_sampled, alarms_found, dataflow_verdict}.

Example: downtime_root_cause_live(endpoint="line1",
    window={"start":"2026-06-28T10:00:00Z","asset":"line1"},
    refs=["ns=2;i=5","ns=2;i=6"]).

Derives a per-site cause-weight profile from a corpus of CONFIRMED past
incidents so downtime_root_cause adapts to what THIS site's evidence actually
predicts. Pure + explainable: each weight is the smoothed signal→cause
precision relative to chance (>1 = evidence for that cause is reliable here,
<1 = often misleading) — no black box. Anti-overfit: Laplace smoothing + a
per-cause min-sample guard, and a fall-back to the shipped defaults when the
corpus is too thin. Feed the returned 'cause_weights' to downtime_root_cause's
cause_weights argument. Advisory: it tunes ranking, never executes anything.

Args:
    history: Confirmed incidents — [{cause, signals:[...]}] where 'cause' is the
        known root cause and 'signals' are the cause labels the evidence pointed
        at (both from the copilot taxonomy: mechanical_fault, comms_loss,
        sensor_fault, material_starvation, quality_reject, changeover, utility_fault).
    min_samples: Minimum confirmed incidents before adapting at all (default 8);
        below it the defaults are kept.
    smoothing: Laplace pseudo-count pulling each estimate toward chance (default 1.0).

Returns dict: {cause_weights:{cause: multiplier}, n_incidents, per_cause:{cause:
    {support, hits, precision, weight, note}}, rationale}.

Example: learn_cause_weights(history=[{"cause":"mechanical_fault",
    "signals":["mechanical_fault"]}, {"cause":"comms_loss","signals":["comms_loss"]}]).

Auto-builds the labeled history learn_cause_weights needs from closed maintenance
records: an explicit taxonomy cause column wins; else a built-in EN/中文 CMMS
synonym table (extendable via 'synonyms'); else UNAMBIGUOUS keyword inference
over the row's free text using the copilot's own cause keywords. Rows it cannot
map land in 'unmapped' with the reason — never silently guessed. 'signals' come
from an explicit column or the symptom/alarm text (may stay empty — no fabricated
evidence). Pure + advisory; with learn=true the learned weights are included.

Args:
    rows: Work-order records, one dict each. Recognized cause columns:
        cause / root_cause / failure_class / category / problem_code; free-text
        columns: description / problem / notes / comment / text / 故障描述;
        signal text: symptom(s) / alarm(s) / 现象.
    synonyms: Extra site vocabulary, e.g. {"spindle crash": "mechanical_fault"};
        values must be taxonomy causes.
    learn: Also run learn_cause_weights on the mapped corpus (default true).
    min_samples: Passed to learn_cause_weights (default 8).
    smoothing: Passed to learn_cause_weights (default 1.0).

Returns dict: {corpus:[{cause, signals}], n_rows, n_mapped, unmapped:[{row,
    reason, excerpt}], mapped_via, weights?, next_step}.

Example: rca_corpus_from_maintenance(rows=[{"category":"轴承损坏",
    "symptom":"drive overload alarm"}], synonyms={"spindle crash":"mechanical_fault"}).

Scores each tag 0-100 on whether its data can be BELIEVED — staleness, dead
heartbeat, bad-quality, flatline, gaps, anomaly — then rolls up per endpoint
and across the fleet. NOT process health (it does not score whether a value is
alarming, only whether it is trustworthy). Pure analysis over provided feeds.

Args:
    feeds: Per-endpoint feeds — {endpoint, tags:[{ref, label?, samples:[scalars
        or {value, good|quality, timestamp?}], expected_update_s?, heartbeat?}]}.
    default_staleness_s: Max sample-age before 'stale' when a tag sets no
        expected_update_s (default 300).
    now: ISO-8601 reference time for staleness (deterministic); omit for now-UTC.

Returns dict: {evaluated_endpoints, evaluated_tags, fleet_score (0-100),
    fleet_status, issue_breakdown{}, worst_endpoints[], worst_tags[],
    endpoints:[{endpoint, score, status, status_counts, worst_tag}]}.

Example: data_quality_scorecard(feeds=[{"endpoint":"line1","tags":[{"ref":"hb",
    "heartbeat":true,"samples":[5,5,5,5]}]}]).

Builds on data_quality_scorecard to give a fleet-wide view: endpoints ranked by
their single worst tag, bad-quality tag counts aggregated across every endpoint,
and a first-class liveness rollup (dead-heartbeat / flatline). Staleness and gap
budgets are configurable per tag (staleness_s / gap_threshold_s) and per feed,
so a slow daily counter is not judged like a 1Hz sensor. Pure analysis.

Args:
    feeds: Per-endpoint feeds — {endpoint, staleness_s?, tags:[{ref, label?,
        samples:[scalars or {value, good|quality, timestamp?}], expected_update_s?,
        staleness_s?, gap_threshold_s?, flatline_after_s?, heartbeat?}]}.
    default_staleness_s: Fallback max sample-age (seconds) before 'stale' when a
        tag/feed sets no staleness_s/expected_update_s (default 300).
    now: ISO-8601 reference time for staleness (deterministic); omit for now-UTC.
    top_n: How many endpoints / bad-quality rows to return (default 10).

Returns dict: {evaluated_endpoints, evaluated_tags, fleet_score (0-100),
    fleet_status, endpoints_ranked_by_worst_tag:[...], bad_quality_rollup:
    {total_bad_quality_tags, endpoints_affected, by_endpoint:[{endpoint,
    bad_quality_tags, fully_bad, partial_bad}]}, liveness_rollup:
    {dead_heartbeat_count, flatline_count, dead_heartbeats[], flatlines[]},
    issue_breakdown{}}.

Example: data_quality_fleet_rollup(feeds=[{"endpoint":"line1","tags":[{"ref":"t",
    "samples":[{"value":None,"good":false}]}]}]).

A heartbeat must keep CHANGING; a flatlined one means the upstream is dead even
when comms/quality look fine. With timestamped samples + max_interval_s, also
flags the longest stall.

Args:
    series: Heartbeat samples — scalars or {value, timestamp?} (a counter/toggle).
    max_interval_s: Max allowed gap between changes; exceeding it = not alive.

Returns dict: {alive (bool), samples, distinct_transitions, spread,
    longest_stall_s, reason}.

Example: heartbeat_health(series=[1,2,3,4,5], max_interval_s=10).

Deepens alarm_bad_actors: detects flood *episodes* (start/end/count/peak rate/
top contributors + each episode's first-out annunciation, per ISA-18.2's >=10
alarms per 10 min per operator), alarms chattering ACTIVE↔CLEARED, standing/
stale alarms, and percent-time-in-flood vs the ISA-18.2 targets (~1-2 alarms/
10 min steady state, <1% time in flood). Also returns an ISA-18.2 'load_profile'
(per-bucket rate band + peak period + trend) and per-source 'suppression_advice'
(deadband/on-off-delay for chatter, time-limited shelve for standing alarms).
The suppression advice is ADVISORY ONLY — starting values for a human to review
and approve via your ISA-18.2 / management-of-change process; this tool never
applies suppression, shelving, deadband, or delay changes. Pass 'events' for
pure analysis, or an endpoint to collect live via the same OPC-UA active-
condition scan the RCA copilot uses (polled over duration_s; other protocols
contribute no alarms). Output is bounded; 'truncated' flags say when caps bit.

Args:
    endpoint: Endpoint name from config (used only when events is omitted).
    duration_s: Live collection window in seconds (1..300, default 60).
    window_s: Flood analysis window in seconds (ISA-18.2 default 600).
    threshold: Annunciations per window that start a flood (default 10).
    events: Injected alarm events — {source, timestamp (ISO-8601), state?
        (ACTIVE/RTN/CLEARED)}; skips live collection entirely.
    stale_after_s: Continuously-active age that marks a standing alarm (default 24h).
    max_episodes: Cap on returned flood episodes (default 20).
    max_rows: Cap on chattering / stale / suppression-advice / worksheet rows (default 50).
    load_bucket_s: Load-profile bucket width in seconds (ISA-18.2 default 600 = 10 min).

Returns dict: {event_count, summary:{insufficient_data, percent_time_in_flood,
    avg_alarms_per_10min, peak_alarms_per_10min, isa_18_2_targets, ...},
    load_profile:{overall_band, peak_bucket, band_distribution, trend,
    busiest_buckets:[...], ...}, flood_episodes:[{start, end, ..., top_contributors,
    first_out:{source, ts}}], chattering:[{source, cycles, cycles_per_hour, ...}],
    stale_standing:[{source, active_since, active_for_s}], suppression_advice:[{source,
    kind, technique, suggested_on_delay_s, suggested_off_delay_s, suggested_shelve_max_s,
    basis, advisory}], worksheet_preview:[...], advisory_note, truncated:{...}, collected?}.

Example: alarm_flood_analysis(events=[{"source":"FIC101",
    "timestamp":"2026-06-28T10:00:00Z","state":"ACTIVE"}, ...]).

Answers "which alarm to look at first" when 100+ alarms hit in minutes: groups annunciations
into cascades (a new cascade starts after a quiet gap > window_s) and reports the FIRST-OUT
alarm (earliest in the burst) as the likely root, plus downstream members and any chattering
sources. First-out is a transparent heuristic cited by timestamp — NOT causal (use
downtime_root_cause for causality). Pass 'events' for pure analysis, or an endpoint to collect
live via the OPC-UA active-condition scan. Read-only; bounded.

Args:
    endpoint: Endpoint name from config (used only when events is omitted).
    duration_s: Live collection window in seconds (1..300, default 60).
    window_s: Quiet gap (seconds) that separates one cascade from the next (default 60).
    min_cascade: Minimum annunciations for a group to count as a cascade (default 2).
    events: Injected alarm events — {source, timestamp (ISO-8601), state?}; skips live collect.

Returns dict: {cascade_count, total_activations, cascades:[{root:{source, ts}, size,
    distinct_sources, span_s, members[], chattering[]}], collected?}.

Example: alarm_cascade(events=[{"source": "PT101", "timestamp": "2026-06-28T10:00:00Z"}, ...]).

One row per alarm source, count-descending: count, % of total annunciations,
chattering?, flood contributor?, and a recommendation stub — the starting
document for an ISA-18.2 rationalization review. Pass 'events' for pure
analysis, or an endpoint to collect live via the same OPC-UA active-condition
scan the RCA copilot uses. With out_path the full worksheet is written as CSV
and the path returned; otherwise bounded inline rows (truncation noted).

Args:
    endpoint: Endpoint name from config (used only when events is omitted).
    duration_s: Live collection window in seconds (1..300, default 60).
    events: Injected alarm events — {source, timestamp (ISO-8601), state?}.
    window_s: Flood analysis window in seconds (ISA-18.2 default 600).
    threshold: Annunciations per window that start a flood (default 10).
    out_path: Optional CSV destination; parent directory must exist.

Returns dict: {row_count, columns:[alarm_id, count, pct_of_total, chattering,
    in_flood, recommendation], csv_path? , rows?:[...], truncated (bool)}.

Example: alarm_rationalization_worksheet(events=[...], out_path="worksheet.csv").

Connects to each target with our own protocol client and reads its identity
call (S7 CPU info, EtherNet/IP controller info, OPC-UA server build info,
Modbus device identification FC43, Mitsubishi CPU type, MTConnect device
model), aggregating vendor/model/firmware/serial per device.

Honest scope: ACTIVE fingerprinting (we connect to each device), NOT passive
SPAN/tap discovery. Only finds devices we are configured to reach.

Args:
    endpoints: Endpoint names to fingerprint; omit to fingerprint ALL
        configured endpoints.

Returns dict: {asset_count, reachable_count, unreachable_count, method:
    'active_fingerprint', assets:[{endpoint, protocol, address, vendor, model,
    firmware, serial, reachable, last_seen, error}]}.

Example: asset_inventory(endpoints=["press1","cell5"]).

Unifies the two per-protocol tag models (OPC-UA address-space discovery +
Modbus register templates) into one cross-protocol asset/tag/alias model. Tags
are re-classified with the SAME semantic classifier the OPC-UA layer uses,
grouped into assets ACROSS protocols (a ``Line1`` OPC-UA folder + a ``Line1``
Modbus block become one asset), and each gets a canonical alias
``<site>.<asset>.<class_or_name>``. Advisory only — aliases are SUGGESTIONS,
never a server-side rename (OT-dangerous).

Args:
    feeds: List of per-protocol feeds, each
        ``{protocol, source, asset?, tags:[...]}``. ``tags`` may be OPC-UA
        discovery descriptors (from opcua_discover_tags), Modbus template tags
        (from modbus_apply_template), or already-normalized tags. A feed-level
        ``asset`` is applied to its tags that don't carry their own.
    site: Site prefix for canonical aliases (default 'site').

Returns dict: {site, protocols, tag_count, asset_count, assets:[{asset,
    protocols, tag_count, classes, tags:[{protocol, source, name, ref, asset,
    unit, klass, canonical_alias, suggested_alias}]}], naming_quality:
    {alias_collisions, cross_protocol_overlaps, cryptic_names, verdict}}.

Example: cross_protocol_asset_model(feeds=[
    {"protocol":"opcua","source":"line1","tags":[...]},
    {"protocol":"modbus","source":"meter1","asset":"Line1","tags":[...]}],
    site="plant").

Writes a local owner-only advisory JSON file (NOT an OT-device write — hence
risk=low); see the persistence note below.

Runs the cross-protocol asset model over ``feeds``, extracts the adopted map
``{canonical_alias: {ref, protocol, asset, name, class}}``, and persists it as
the site's baseline (owner-only JSON under the iaiops home). Re-running
overwrites the baseline. Advisory — the map is a SUGGESTION, never a
server-side rename (OT-dangerous).

Args:
    feeds: Per-protocol tag feeds ``[{protocol, source, asset?, tags:[...]}]``,
        the SAME shape ``cross_protocol_asset_model`` takes.
    site: Site label (a safe file leaf: alphanumeric/_/-). Default 'site'.

Returns dict: {site, path, tag_count, adopted:{alias: {...}}}.

Example: adopt_alias_map(feeds=[{"protocol":"opcua","source":"l1","tags":[...]}],
    site="plant").

Loads the site's previously adopted alias map, re-runs the cross-protocol
asset model over ``feeds``, and reports how the address space moved: tags
added / removed / renamed (same ref, new alias) / reclassified (same ref+alias,
new semantic class), plus a stable|changed verdict. Adopt a baseline first
with ``adopt_alias_map``.

Args:
    feeds: Fresh per-protocol tag feeds (same shape as adopt_alias_map).
    site: Site label whose baseline to diff against. Default 'site'.

Returns dict: {site, verdict, counts:{added,removed,renamed,reclassified},
    added:[...], removed:[...], renamed:[...], reclassified:[...]}.

Example: diff_alias_map(feeds=[{"protocol":"opcua","source":"l1","tags":[...]}],
    site="plant").

Args:
    planned_time_s: Planned production time (seconds).
    run_time_s: Actual running time (seconds) — planned minus downtime.
    ideal_cycle_time_s: Ideal/nameplate cycle time per part (seconds).
    total_count: Total parts produced.
    good_count: Good (non-reject) parts produced.
    breakdown_time_s: Optional — unplanned-stop seconds (splits availability loss).
    setup_time_s: Optional — changeover/setup seconds (splits availability loss).
    minor_stop_time_s: Optional — minor-stop seconds (splits performance loss;
        the remainder is speed loss).
    startup_reject_count: Optional — startup/warm-up rejects (splits quality
        loss; the remainder is production rejects).
    actual_kwh: Optional — measured energy for this run; enables the energy block.
    baseline_kwh: Optional — expected/baseline energy for the actual-vs-baseline
        deviation verdict.
    emission_factor_kg_per_kwh: Optional — carbon factor (kg CO2e/kWh). Default is
        a flagged placeholder (see the tool's carbon note); pass the grid's value.
    energy_tolerance: ± band (fraction) for the over/under/on-target verdict.

Returns dict: OEE factors + oee/oee_pct + inputs + losses, plus
    ``six_big_losses`` (breakdown/setup/minor-stops/speed/startup/production-reject
    time-ladder that sums with OEE to 100%) and, when ``actual_kwh`` is given,
    ``energy`` (kwh_per_unit, carbon, and baseline deviation).

Example: oee_compute(planned_time_s=28800, run_time_s=25200,
    ideal_cycle_time_s=2.0, total_count=12000, good_count=11800,
    setup_time_s=1800, actual_kwh=940, baseline_kwh=880).

Args:
    series: Timestamped samples — {timestamp (ISO-8601), state} where state is
        a string (RUNNING/IDLE/FAULT…), a bool, or a number.
    category_map: Optional {state_label: category} override (else keyword
        heuristics map to changeover/material/mechanical/quality/break/unknown).
    min_duration_s: Ignore stoppages shorter than this (seconds).

Returns dict: {samples, event_count, total_downtime_s, by_category:{cat:
    {count, downtime_s}}, events:[{start, end, duration_s, state, category}]}.

Example: downtime_events(series=[{"timestamp":"2026-06-28T08:00:00Z","state":"RUNNING"},
    {"timestamp":"2026-06-28T08:05:00Z","state":"FAULT"}, ...]).

Args:
    records: Labelled records — {<dimension labels>, planned_time_s, run_time_s,
        ideal_cycle_time_s, total_count, good_count} plus optional actual_kwh /
        baseline_kwh to enable the energy rollup.
    dimensions: Dimension keys to group by (default ['machine','part','shift']);
        use ['shift'] for the classic by-shift energy comparison.
    emission_factor_kg_per_kwh: Optional carbon factor (kg CO2e/kWh); default is a
        flagged placeholder — pass the grid's published value.
    energy_tolerance: ± band (fraction) for the actual-vs-baseline verdict.

Returns dict: {dimensions, group_count, mean_oee, worst_performers:[...],
    matrix:[{dimensions, oee, oee_pct, availability, performance, quality,
    energy?}]}. When any record carries energy, adds an ``energy_baseline`` block
    that flags cross-group deviation anomalies (tolerance + robust-outlier rules).

Example: oee_multidim(records=[{"shift":"day","planned_time_s":28800,
    "run_time_s":25000,"ideal_cycle_time_s":2,"total_count":12000,
    "good_count":11800,"actual_kwh":940,"baseline_kwh":880}], dimensions=["shift"]).

Polls ``ref`` and returns only the changes (with timestamps), not every
sample — the OT deadband-report pattern. Works across OPC-UA / Modbus / S7 /
Mitsubishi MC / EtherNet/IP. Hard-capped by duration_s and max_changes (never
an infinite loop).

Args:
    ref: Point to watch — OPC-UA node id, Modbus address, S7 address string,
        MELSEC device, or Logix tag (per the endpoint's protocol).
    endpoint: Endpoint name from config.
    duration_s: Wall-clock window in seconds (1..120, capped server-side).
    interval_ms: Poll interval in milliseconds (>=50).
    deadband: Numeric change must exceed this to count (0 = any change).
    max_changes: Stop after this many changes (1..500, capped server-side).

Returns dict: {endpoint, ref, duration_s, interval_ms, deadband, samples_polled,
    change_count, changes:[{value, previous, source_timestamp, wall_clock}]}.

Example: monitor_changes(ref="ns=2;i=5", endpoint="line1", duration_s=20, deadband=0.5).

An honest onboarding/sales self-assessment across the pillars 分区隔离 / 可审计 /
双向认证 / 最小权限 / 数据保护 / 自主可控. Each control names how iaiops addresses
it, an honest status (addressed / partial / 待核实), and the remaining gap.

Returns dict: {framework, frameworks[], pillars[], control_count, status_summary
    {addressed, partial, 待核实}, controls:[{pillar, requirement, iaiops, status,
    gap, crosswalk{dengbao, iec62443}}]}. See compliance_frameworks for the full
    cross-framework 对照.

Example: compliance_mapping().

One row per governance pillar, showing the matching 《工控系统网络安全防护指南》
requirement, 等保 2.0 control class, IEC 62443 foundational requirement, and the
current iaiops status. Companion to compliance_mapping (which carries the honest
per-control gap); use this to answer "which 等保 / 62443 clause does this satisfy".

Returns dict: {frameworks:[{id,name,region,kind}], framework_count, pillar_count,
    crosswalk:[{pillar, gjzn, dengbao, iec62443, iaiops_status}], note}.

Example: compliance_frameworks().

等保 2.0 (GB/T 22239) is graded — the same control tightens as the level rises.
Per governance pillar this shows the 二级 baseline, what 三级 additionally requires,
and how far iaiops moves you toward it (with the honest per-control status/gap).
An onboarding/self-assessment aid, NOT a certification.

Args:
    level: Focus on one level — 'l2'/'l3', '二级'/'三级', or '2'/'3'. Omit for both.

Returns dict: {framework, levels:[{id,name,note}], selected_level, pillar_count,
    deltas:[{pillar, l2_requires?, l3_adds?, iaiops, iaiops_status, gap}], note}.

Example: compliance_dengbao_levels(level="三级").

Turns the compliance crosswalk into a deliverable document a CISO can read:
title-page metadata (site / date / iaiops version), per-pillar 等保 L2/L3 status
table, IEC 62443 FR1–6 crosswalk, honest gap list, and a governance-controls
appendix (audit hash chain / approval tokens / dry-run+undo / mTLS). An
onboarding/self-assessment aid, NOT a certification.

Args:
    level: 等保 2.0 target level — 'l2'/'l3', '二级'/'三级', '2'/'3'. Omit for both.
    site: Site / plant name stamped on the title page.
    out_path: Optional file to write the markdown to (.md). Required when the
        report exceeds the inline bound (~400 lines): without it the inline
        markdown is truncated with a note.

Returns dict: {format, level, line_count, path?} plus either the full inline
    {markdown} (when within bounds and no out_path) or {markdown (truncated),
    truncated: true} with a hint to pass out_path.

Example: compliance_report(level="三级", site="示例水厂",
    out_path="/tmp/compliance-report.md").

Packages the governance evidence trail into one deterministic zip:
audit_rows.jsonl (secrets already redacted upstream), chain_verification.json
(SHA-256 hash-chain walk result), rules.yaml (if present), doctor_summary.json
(non-probing config/secret-store facts), and manifest.json. Path is validated
(no '..' traversal; parent created 0700).

Args:
    out_path: Destination zip path (must end in .zip).
    since: Optional ISO-8601 floor on the audit row timestamp (inclusive).
    until: Optional ISO-8601 ceiling on the audit row timestamp (inclusive).

Returns dict: {path, row_count, chain{ok, checked, unhashed, ...}, files[],
    since, until}.

Example: compliance_evidence_bundle(out_path="/tmp/evidence.zip",
    since="2026-06-01T00:00:00+00:00").

Writes already-collected points to a domestic historian (信创) — TDengine or
IoTDB — instead of binding InfluxDB. Data egress to the operator's OWN database,
NOT a control-system write. Non-numeric points are skipped (numeric value column).

Args:
    points: Collected points — {ref|metric, value|present_value, timestamp?, ...}
        (e.g. the output of interrogate / integrity_poll / read_points / monitor).
    sink: 'tdengine' or 'iotdb'.
    host/port/user/password: TSDB connection params (sensible defaults per sink
        when blank/0).
    database: Target database (TDengine db / IoTDB storage group, e.g. 'root.iaiops').

Returns dict: {sink, received, written, skipped_non_numeric, database}.

Example: historian_push(points=[{"ref":"line1.temp","value":21.5}], sink="tdengine",
    host="10.0.0.20", database="iaiops").

Source is ~/.iaiops/data.db — the local queryable store written by
historian_push(sink="sqlite") — NOT a live device read. Writes csv (Excel),
sqlite (SQL browser / Power BI) or parquet (pandas/Spark; needs
pip install 'iaiops[export]'), and returns the file path + row count with a
bounded inline preview (first 200 rows max) so the response never floods.

Args:
    fmt: 'csv' | 'sqlite' | 'parquet'.
    since/until: Optional ISO-8601 time bounds (inclusive).
    endpoint: Only samples from this endpoint label.
    tag: Only samples for this tag.
    limit: Max rows exported (1..100000; default 10000).
    out_path: Output file; default ~/.iaiops/exports/iaiops-export-<ts>.<ext>.

Returns dict: {format, path, rows, preview_rows:[{ts, endpoint, protocol, tag,
    value, quality, unit}] (≤200), preview_truncated}.

Example: export_data(fmt="csv", tag="line1.temp", since="2026-07-01T00:00:00").

Source is ~/.iaiops/data.db — the local store written by
historian_push(sink="sqlite") — NOT a live device read. Learns robust
percentiles (p1/p99 + median/MAD, no ML) from the tag's own samples,
segmented at the latest change recorded via baseline_record_change (the band
reflects only the post-change regime). REFUSES with an explicit
insufficient_data verdict (listing exactly what is missing) below 100 usable
samples or under 24h of span — it never invents a band from thin data. On
success the band is persisted to ~/.iaiops/baselines.json (owner-only local
metadata, not an OT write).

Args:
    tag: Tag name to learn, e.g. 'line1.temp'.
    endpoint: Only samples from this endpoint label.
    since: Only samples at/after this ISO-8601 time.

Returns dict: {status: 'ok'|'insufficient_data', tag, band:{p1,p99,median,mad},
    n_samples, window:{from_ts,to_ts,span_s}, segment, missing?:[...], note}.

Example: baseline_learn(tag="line1.temp", since="2026-06-01T00:00:00").

Reads the last window_s seconds from ~/.iaiops/data.db (no device I/O) and
judges them against the stored band. Conservative by design: a violation is
reported ONLY when values are beyond p1/p99 by more than 3×MAD AND sustained
for >=3 consecutive samples — a single spike is never flagged. Every
violation cites the baseline window (from/to ts, n samples), the band
values, and the offending samples' timestamps/values. No stored baseline →
an explicit no_baseline answer (never a guess). Bounded output (<=10
violations, <=20 cited samples each).

Args:
    tag: Tag name to check, e.g. 'line1.temp'.
    endpoint: Only samples from this endpoint label.
    window_s: Recent window to check, seconds (60..604800; default 3600).

Returns dict: {status: 'ok'|'violation'|'no_baseline', tag, checked_samples,
    thresholds, baseline_citation, violations:[{direction, from_ts, to_ts,
    consecutive_samples, samples:[{ts,value}], baseline}], note}.

Example: baseline_check(tag="line1.temp", window_s=7200).

Writes ONLY local metadata (~/.iaiops/baselines.json, owner-only) — never an
OT device write, hence risk=low. A recorded change (setpoint moved, valve
replaced, probe swapped) marks a regime boundary: the next baseline_learn
uses only samples AFTER the latest change, so the band never mixes
pre-change and post-change behavior. This operator change log — not a
black-box score — is what makes the baseline trustworthy.

Args:
    tag: Tag whose process changed, e.g. 'line1.temp'.
    note: What changed (required), e.g. 'setpoint 60→70C'.

Returns dict: {tag, change:{ts, note}, changes_recorded}.

Example: baseline_record_change(tag="line1.temp", note="setpoint 60→70C").

Read from the local store only (no history scan, no device I/O) and never
guesses: 'no_baseline' (nothing learned, no refused attempt), 'learning'
(last learn refused — still accumulating history), 'ok' (band learned, last
check clean), 'violation' (last check flagged a sustained excursion). With
no tag, lists every tracked tag (bounded to 100 entries).

Args:
    tag: Optional tag name; omit to list all tracked tags.

Returns dict: {tag, status, band?, baseline_window?, changes_recorded?, ...}
    for one tag, or {tracked_tags, listed, truncated, tags:[...]} for all.

Example: baseline_status(tag="line1.temp").

Reads history back OUT of the store the sinks write — the local SQLite
store (~/.iaiops/data.db), TDengine, or IoTDB — so the RCA copilot / an
agent can see real pre-incident windows instead of only short live samples.
Read-only over the operator's OWN historian; no device I/O. Bounded: rows
are capped and a truncation flag is set when more history exists.

Args:
    tag: Tag/metric name as stored by historian_push (e.g. 'line1.temp').
    since/until: Optional ISO-8601 time bounds (inclusive).
    endpoint: Only samples from this endpoint label (sqlite reader only —
        the TSDB layout stores no endpoint label).
    reader: 'sqlite' | 'tdengine' | 'iotdb'. Omit to use the per-site
        'historian:' block in ~/.iaiops/config.yaml, else the local sqlite
        store. TSDB readers need their extra: pip install iaiops[tdengine|iotdb].
    limit: Max rows returned (1..10000; default 1000).

Returns dict: {reader, source, tag, since, until, rows,
    samples:[{ts, endpoint, protocol, tag, value, quality, unit}], truncated}
    plus the standard return envelope (items_returned, items_total,
    items_total_is_exact, is_truncated, truncation_note). Trust
    `is_truncated`: an empty `samples` with is_truncated=false means the
    history really is empty, NOT that the result was cut short.

Example: historian_query(tag="line1.temp", since="2026-07-02T06:00:00Z",
    until="2026-07-02T08:00:00Z").

Answers the question every RCA starts with: which tags have stored history,
how many rows, and over what time span — per tag {rows, first_ts, last_ts}
from the same store historian_push writes. Read-only, bounded (tag list is
capped with a truncation flag); no device I/O.

Args:
    reader: 'sqlite' | 'tdengine' | 'iotdb'. Omit to use the per-site
        'historian:' block in ~/.iaiops/config.yaml, else the local sqlite
        store. TSDB readers need their extra: pip install iaiops[tdengine|iotdb].
    limit: Max tags returned (1..2000; default 500).

Returns dict: {reader, source, tag_count, tags:[{tag, rows, first_ts,
    last_ts}], truncated} plus the standard return envelope
    (items_returned, items_total, items_total_is_exact, is_truncated,
    truncation_note).

Example: historian_coverage().

Egress of data the agent already READ — NOT a control write. Each numeric point becomes a JSON
message on ``<subject_prefix>.tag.<metric>``; non-numeric points are skipped (use a historian
sink for text/state). Needs the extra: pip install iaiops[nats].

Args:
    points: Collected point dicts (e.g. from *_read_many): {ref/metric, value, timestamp, ...}.
    subject_prefix: NATS subject root (default 'iaiops').
    servers: Comma-separated NATS server URLs (default nats://localhost:4222).
    token: Optional NATS auth token.
    tls: Use TLS to the broker.
    publisher: Bus kind (currently 'nats').

Returns dict: {publisher, subject_prefix, received, published, skipped_non_numeric}.

Example: stream_publish(points=[{"ref": "line1.temp", "value": 21.5}], subject_prefix="plant").

Egress of a finding the brain already COMPUTED — e.g. an RCA verdict or an alarm episode — to
``<subject_prefix>.<subject>`` as JSON. NOT a control write. Needs: pip install iaiops[nats].

Args:
    subject: Event subject suffix (e.g. 'rca.verdict', 'alarm.flood').
    event: The event payload dict (published as JSON).
    servers/token/tls/subject_prefix/publisher: bus connection (see stream_publish).

Returns dict: {publisher, subject, published}.

Example: stream_publish_event(subject="rca.verdict", event={"primary_cause": "seal"}).

Air-gapped: hands the already-computed, already-cited verdict to a LOCAL model (Ollama) that
ONLY rephrases it — it never adds a cause, number, or citation (strict prompt; see docs/RCA.md).
Read-only; no device I/O. Needs the extra + a running local model: pip install iaiops[ollama].

Args:
    verdict: An RCA verdict dict (e.g. the output of downtime_root_cause).
    base_url: Ollama server URL (default http://localhost:11434).
    model: Local model name (default 'llama3.1').
    provider: LLM provider (currently 'ollama').

Returns dict: {provider, model, narration}.

Example: rca_narrate(verdict=<downtime_root_cause output>, model="llama3.1").

The tier above data_quality_fleet_rollup (per-endpoint within one site): this aggregates across
many edge SITES for central management. A site is 'offline' if its last_seen is older than
stale_after_s; fleet_status is the worst site status present. Read-only, pure; no device I/O.

Args:
    sites: Per-site reports, each
        [{site, location?, profile?, status?, score?, issues?, last_seen?}]; status ∈
        ok|degraded|critical|offline (else derived from score); score 0..1.
    stale_after_s: A site with no report newer than this is 'offline' (default 300).
    now: Optional ISO-8601 'now' for deterministic staleness (default: current UTC).

Returns dict: {site_count, fleet_status, fleet_score, by_status, worst_sites[], sites[]}.

Example: fleet_status(sites=[{"site":"sh","score":0.9},{"site":"bj","status":"critical"}]).

Aggregates the incidents each site reports into a fleet picture: how many incidents, which sites
are affected, and the most common root causes across the whole fleet. Read-only; no device I/O.

Args:
    sites: Per-site reports carrying incidents: [{site, incidents:[{cause|primary_cause,
        confidence?}]}].

Returns dict: {total_incidents, sites_with_incidents, affected_sites[], top_causes[]}.

Example: fleet_incidents(sites=[{"site":"plant-sh","incidents":[{"cause":"network"}]}]).

The predictive step above baseline_check (which flags a violation that already happened): fits a
robust Theil-Sen trend to the recent history and, if it continues, estimates the ETA to the
nearest limit in the direction of travel — the early warning that makes maintenance predictive
(inverter/turbine degradation, bearing drift, filter clogging). Refuses thin history; read-only,
pure over the provided series; no device I/O.

Beyond the trend, the result deepens into three explainable, stdlib-only views: a degradation
'pattern' (gradual vs sudden vs cyclic), a remaining-useful-life 'rul' block when degrading
(linear + exponential extrapolation to the limit, a confidence band from the slope spread, and a
fit R^2), and optional time-domain 'waveform' features (RMS/kurtosis/crest/... for
vibration-type signals). Each states its own uncertainty rather than guessing.

Args:
    series: Time-ordered samples: [{value, timestamp?}] (timestamp ISO-8601; if all present the
        ETA is in seconds, otherwise in samples). >= 30 numeric samples required.
    warn_high/alarm_high/warn_low/alarm_low: Optional limits; the forecast targets the nearest
        one in the trend's direction (rising → highs, falling → lows).
    imminent_within_s: ETA (seconds) at/under which status is 'imminent' (default 86400 = 24h).
    include_waveform: Add the time-domain 'waveform' feature block (default True). Set False for
        slow trend-only signals where vibration features do not apply.

Returns dict: {status (insufficient_data|stable|degrading|imminent), samples, direction,
    slope_per_unit, unit (s|samples), current, limit:{name,value}, eta_to_limit,
    degradation:{pattern,confidence,rationale,metrics},
    waveform:{rms,crest_factor,kurtosis,...} (when include_waveform),
    rul:{linear,exponential,eta_band,recommended_model,confidence,...} (when degrading)}.

Example: pdm_forecast(series=[{"value": 62.1, "timestamp": "2026-07-12T00:00:00Z"}, ...],
    warn_high=75, alarm_high=85).

Answers the operator's three simultaneous questions on a stopped line — which
alarm to look at first, the likely cause, and whether anything warned us —
then cross-checks whether the first-out alarm agrees with the RCA verdict.
Composes alarm_cascade + downtime_root_cause + pdm_forecast over ONE incident;
every field traces to a sub-report echoed under 'cascade'/'rca'/
'precursor_forecasts'. Read-first and advisory: it proposes but executes
nothing. Thin evidence downgrades honestly rather than guessing.

Args:
    window: {start (ISO-8601), end?, asset?, category?}. If 'end' is omitted
        but state_series is given, the first running→stopped span bounds it.
    alarms: Alarm/condition events — {source, timestamp, message?, priority?,
        state?}. Feeds BOTH the first-out cascade and the RCA.
    tags: Per-tag samples — {ref, samples:[...], warn_high?, ...} (via tag_health).
    dataflow: A diagnose_dataflow result dict (localizes comms vs field).
    state_series: {timestamp, state} samples to bound the window if 'end' is absent.
    precursors: Signals to check for a pre-incident trend — [{signal, series:
        [scalars or {value, timestamp}], warn_high?, alarm_high?, warn_low?,
        alarm_low?}]; each is run through pdm_forecast and kept only when it was
        degrading/imminent before the trip.
    cascade_window_s: Quiet gap (s) separating alarm cascades (default 60).
    lead_window_s: Causal lead window before onset (default 300s).
    cause_weights: Optional per-site {cause: multiplier} RCA override.
    imminent_within_s: ETA horizon that marks a precursor 'imminent' (default 24h).
    include_graph: When true, the echoed 'rca' sub-report also carries a 'graph'
        block — the SAME verdict re-projected as a causal graph {nodes, edges,
        mermaid, meta} (signal → cause → downtime) for a frontend. Pure re-shape;
        no new reasoning. Omit to keep the flat rca summary (default).

Returns dict: {window, triage:{first_look:{source, ts, cascade_size, basis},
    likely_cause:{cause, verdict, confidence, confidence_band,
    recommended_action}, cross_check:{status ('corroborated'|'diverging'|
    'no_alarm_root'|'no_rca_primary'), detail}, precursors_missed:[{signal,
    status, direction, eta_to_limit, unit, limit}], recommended_next_data},
    cascade:{...}, rca:{verdict, primary_cause, top_hypotheses, graph?},
    precursor_forecasts:[...], anti_hallucination}.

Example: downtime_triage(window={"start":"2026-06-28T10:00:00Z","asset":"line1"},
    alarms=[{"source":"M1_DRIVE","timestamp":"2026-06-28T09:59:50Z",
             "message":"motor overload trip"}],
    precursors=[{"signal":"M1_temp","series":[...],"warn_high":80}]).

Parses one exported text file (Siemens SCL/ST .scl/.st, AWL/STL .awl,
Rockwell Studio 5000 .L5X — .txt is content-sniffed) and returns blocks
(FB/FC/OB/DB/routines/AOIs) with VAR sections, IF/CASE branch inventory,
timers/counters, and the call graph. Never uploads from a live PLC; reads
exactly the named file (≤5 MB). Every element cites source_file + line
(rung number for L5X ladder) — quote those citations when explaining.
Malformed sections degrade to entries in parse_errors, never a crash.

Args:
    path: Exported program file (.st/.scl/.awl/.l5x/.txt; must exist, ≤5 MB).

Returns dict: {source_file, format, stats:{blocks, variables, call_edges,
    branches, timers_counters, comments, lines, parse_errors},
    blocks:[{name, kind, language, line, end_line, variables (≤100,
    variables_truncated), calls, branches, timers_counters, networks,
    comment}] (≤50, blocks_truncated), call_graph:[{caller, callee,
    source_file, line}], parse_errors, citation_note}.

Example: plc_program_outline(path="~/exports/Line3_Conveyor.scl").

Finds every read/write/call/declare site of a symbol or absolute address
(e.g. Motor_Run, "FB_Conveyor", DB10.DBX0.1, M0.0, Tank[2].Level) in one
exported file, quoting the surrounding source line verbatim so the agent
cites real code. Access classification is heuristic (op/regex based, not
data-flow analysis): SCL ':='→write, '('→call; AWL T/=/S/R→write,
L/A/O…→read, CALL→call; L5X OTE/OTL/OTU/RES and MOV-dest→write. For L5X,
line is the rung number.

Args:
    path: Exported program file (.st/.scl/.awl/.l5x/.txt; must exist, ≤5 MB).
    symbol: Symbol / tag / absolute address to trace (word-bounded match).

Returns dict: {source_file, format, symbol, hit_count,
    hits:[{symbol, access, block, source_file, line, source_line}] (≤200),
    hits_truncated, by_access:{read, write, call, declare, reference}}.

Example: plc_program_xref(path="~/exports/OB1.awl", symbol="M10.0").

Returns the exact source of a single block (FB/FC/OB/DB name for SCL/AWL;
Program.Routine or routine name for L5X — rungs are rendered as
'[rung N] ...'), capped at 200 lines with an explicit truncated flag, so
the agent reads exactly the section it is explaining instead of guessing.
Unknown block names fail with the list of available blocks.

Args:
    path: Exported program file (.st/.scl/.awl/.l5x/.txt; must exist, ≤5 MB).
    block: Block/routine name (case-insensitive; quotes optional).

Returns dict: {source_file, format, block, kind, start_line, end_line,
    lines_returned, truncated, source, parse_errors}.

Example: plc_program_section(path="~/exports/Line3.scl", block="FB_Conveyor").

The "what am I inheriting?" view over one EXPORTED program (SCL/ST, AWL/STL,
Rockwell L5X): folds the structural outline into documentation coverage, the
least-commented blocks, blocks nothing references (possible dead code), the
complexity hotspots, risky constructs (unconditional JMPs, retentive RTO
timers, loops), and a TRANSPARENT additive risk score whose every point cites
its reason. Structural only — it anchors an engineer's review of a line
somebody else left behind, not a semantic understanding. Reads exactly the
named file (≤5 MB); never a live PLC upload. Every finding cites source_file +
line (rung number for L5X ladder).

Args:
    path: Exported program file (.st/.scl/.awl/.l5x/.txt; must exist, ≤5 MB).

Returns dict: {source_file, fmt, stats:{blocks, call_edges, line_count,
    comment_count, comment_ratio, variables, branches, timers_counters},
    documentation:{comment_ratio, band ('well_commented'|'sparse'|
    'undocumented'), uncommented_block_count, uncommented_blocks},
    entry_points:[{name, kind}], unreferenced_blocks:[{name, kind,
    source_file, line}], complexity_hotspots:[{block, kind, score, branches,
    calls, timers_counters, source_file, line}], risky_constructs:{
    unconditional_jumps, unconditional_jump_count, loops, loop_count,
    retentive_timers, retentive_timer_count}, risk:{score (0..100), band
    ('low'|'medium'|'high'), reasons[]}, parse_errors, note}.

Example: plc_program_visibility(path="~/exports/Line3_Conveyor.scl").

Args:
    endpoint: Endpoint name from config (protocol 'iec104'); omit for default.

Returns dict: {endpoint, host, port, connected, configured_common_address,
    station_count, common_addresses[]}.

Example: iec104_connection_info(endpoint="rtu1").

Args:
    common_address: ASDU common address; omit for the configured/first station.
    endpoint: Endpoint name from config (protocol 'iec104').

Returns dict: {endpoint, common_address, point_count, points:[{io_address, type,
    value, quality, recorded_at}]}.

Example: iec104_interrogate(common_address=1, endpoint="rtu1").

Args:
    io_address: The point's information-object address (IOA).
    common_address: ASDU common address; omit for the configured/first station.
    endpoint: Endpoint name from config (protocol 'iec104').

Returns dict: {endpoint, common_address, found, io_address, type, value, quality,
    recorded_at}.

Example: iec104_read_point(io_address=1001, common_address=1, endpoint="rtu1").

Args:
    endpoint: Endpoint name from config (protocol 'dnp3'); omit for default.

Returns dict: {endpoint, host, port, outstation_address, master_address, online}.

Example: dnp3_link_status(endpoint="rtu2").

Returns all static points grouped by measurement type (binary_input,
analog_input, counter, …).

Args:
    endpoint: Endpoint name from config (protocol 'dnp3').

Returns dict: {endpoint, outstation_address, point_count, by_type{},
    points:[{type, group, index, value, quality, timestamp}]}.

Example: dnp3_integrity_poll(endpoint="rtu2").

Args:
    include_children: Also browse each logical device's immediate model children.
    endpoint: Endpoint name from config (protocol 'iec61850'); omit for default.

Returns dict: {endpoint, logical_device_count, logical_devices:[{logical_device,
    children[]?, child_count?}]}.

Example: iec61850_device_directory(include_children=True, endpoint="ied1").

Args:
    reference: Model reference, e.g. 'IED1LD0' or 'IED1LD0/LLN0'.
    endpoint: Endpoint name from config (protocol 'iec61850').

Returns dict: {endpoint, reference, child_count, children[]}.

Example: iec61850_browse(reference="IED1LD0/MMXU1", endpoint="ied1").

Args:
    reference: Data-attribute object reference, e.g. 'IED1MMXU1.TotW.mag.f'.
    fc: Functional constraint — MX (measurands), ST (status), CF (config), …
    endpoint: Endpoint name from config (protocol 'iec61850').

Returns dict: {endpoint, reference, fc, value, error}.

Example: iec61850_read(reference="IED1MMXU1.TotW.mag.f", fc="MX", endpoint="ied1").

Pure structural analysis over INJECTED events (no live protocol I/O, no
endpoint): given relay pickups/trips, breaker open/close, lockouts and bus
undervoltage with ISO-8601 timestamps, decide what tripped and whether
protection coordinated — a selective trip (one zone contained), a
non-selective backup operation (wider outage), or a breaker failure.
Monitor-only, advisory, cite-first (every claim ties to a timestamped event).

Args:
    events: SOE list of {ref, timestamp (ISO-8601), type, label?}; ``type``
        is one of protection_pickup / protection_trip / breaker_open /
        breaker_close / lockout / bus_undervoltage (a free-text ``label`` is
        keyword-matched when ``type`` is absent or unknown). ``ref`` is the
        point name / IOA.
    breaker_fail_window_s: Breaker-failure timer — the tripped breaker's own
        open must be seen within this many seconds (default 0.25).
    backup_margin_s: Backup coordination margin — a breaker opening later
        than this past the first protection event is treated as backup
        operation (default 0.5).

Returns dict: {events_analyzed, ignored, verdict, first_protection,
    first_breaker_open, breakers_opened, breaker_open_count, affected_refs,
    timeline, coordination{status, detail}, note}. ``verdict`` is one of
    selective_trip / backup_operation / breaker_failure / insufficient.

Example: substation_event_analysis(events=[
    {"ref": "R1", "type": "protection_trip", "timestamp": "2026-07-12T10:00:00Z"},
    {"ref": "BK1", "type": "breaker_open", "timestamp": "2026-07-12T10:00:00.08Z"}]).