epics-base-rs 0.18.2

use std::collections::HashSet;
use std::sync::Arc;

use crate::error::{CaError, CaResult};
use crate::runtime::sync::RwLock;
use crate::server::record::RecordInstance;
use crate::types::EpicsValue;

use super::{PvDatabase, apply_timestamp};

/// Result of the simulation-mode check.
///
/// C `aiRecord.c:151-168` handles simulation entirely inside
/// `readValue()`; `process()` then ALWAYS runs `convert`/`checkAlarms`/
/// `monitor`/`recGblFwdLink(prec)`. A simulated record therefore must
/// NOT skip the forward-link / CP / RPRO tail — only the device read
/// and record-support body are replaced by the SIOL round-trip.
enum SimOutcome {
    /// SIMM disabled / no simulation link configured: run the record
    /// body normally.
    NotSimulated,
    /// Simulation handled the record value (SIOL read/write done).
    /// The caller must still run the forward-link / CP / RPRO tail
    /// exactly as `recGblFwdLink` does for a real process cycle.
    Simulated,
}

impl PvDatabase {
    /// Process a record by name (process_local + notify).
    /// Alias-aware (epics-base PR #336).
    pub async fn process_record(&self, name: &str) -> CaResult<()> {
        self.process_record_inner(name, true).await
    }

    /// BR-R15: `process_record` variant for a caller that already
    /// owns the record's advisory write gate — the QSRV atomic group
    /// PUT applying a `+proc` member. The gate `Mutex` is not
    /// reentrant; the atomic group path MUST use this entry. See
    /// [`crate::server::database::PvDatabase::lock_records`].
    pub async fn process_record_already_locked(&self, name: &str) -> CaResult<()> {
        self.process_record_inner(name, false).await
    }

    async fn process_record_inner(&self, name: &str, acquire_gate: bool) -> CaResult<()> {
        let rec = self.get_record(name).await;

        if let Some(rec) = rec {
            // BR-R15: advisory write gate (`dbScanLock` analogue). A
            // QSRV atomic group with a `+proc` member holds this
            // record's gate via `lock_records`; a direct
            // `process_record` on the same backing record must block
            // until the atomic group transaction completes. Skipped
            // when the caller already owns the gate.
            let _record_gate = if acquire_gate {
                let canonical = self
                    .resolve_alias(name)
                    .await
                    .unwrap_or_else(|| name.to_string());
                Some(self.lock_record(&canonical).await)
            } else {
                None
            };
            let (snapshot, alarm_posts) = {
                let mut instance = rec.write().await;
                instance.process_local()?
            };
            // Notify outside lock
            let instance = rec.read().await;
            instance.notify_from_snapshot(&snapshot);
            // Post the alarm fields (SEVR/STAT/ACKS) with their
            // individual C masks — see `process_local` / recGblResetAlarms.
            for &(field, mask) in &alarm_posts {
                instance.notify_field(field, mask);
            }
            Ok(())
        } else {
            Err(CaError::ChannelNotFound(name.to_string()))
        }
    }

    /// Process a record with full link handling (INP -> process -> alarms -> OUT -> FLNK).
    /// Uses visited set for cycle detection and depth limit.
    ///
    /// Foreign-caller entry: FLNK dispatch, scan loop, scan_event, CA put,
    /// process(PROC=1) etc. Hits the PACT entry guard (mirrors C `dbProcess`
    /// at `dbAccess.c:537-559`) when the record is mid-async.
    ///
    /// MR-R5: this is a *foreign* full-processing entry, so it acquires
    /// the record's advisory write gate (`dbScanLock` analogue) for the
    /// entry record before processing. A QSRV atomic group or pvalink
    /// atomic scan-on-update epoch that holds `lock_records` over the
    /// same record blocks a foreign scan/event/FLNK-dispatch caller
    /// here, and vice versa — restoring the `DBManyLock` exclusion. The
    /// recursive FLNK / OUT / CP fan-out within one chain does NOT
    /// re-acquire the gate (`process_record_with_links_recursive`),
    /// mirroring C `processTarget` (`dbDbLink.c:436`) which asserts the
    /// target's lock set is already owned by the calling thread; the
    /// `visited` cycle guard prevents re-processing the entry record.
    pub fn process_record_with_links<'a>(
        &'a self,
        name: &'a str,
        visited: &'a mut HashSet<String>,
        depth: usize,
    ) -> std::pin::Pin<Box<dyn std::future::Future<Output = CaResult<()>> + Send + 'a>> {
        Box::pin(async move {
            self.process_record_with_links_inner(name, visited, depth, false, true)
                .await
        })
    }

    /// MR-R5: full-processing entry for a caller that already owns the
    /// record's advisory write gate via [`PvDatabase::lock_records`] —
    /// the QSRV atomic group GET/PUT and the pvalink atomic
    /// scan-on-update epoch. The advisory gate `Mutex` is not
    /// reentrant; a transaction owner holding `lock_records` over the
    /// member set MUST use this entry to scan a member record, or it
    /// would deadlock against its own epoch guard. Foreign (non-owner)
    /// callers must use [`Self::process_record_with_links`] so the gate
    /// is taken.
    pub fn process_record_with_links_already_locked<'a>(
        &'a self,
        name: &'a str,
        visited: &'a mut HashSet<String>,
        depth: usize,
    ) -> std::pin::Pin<Box<dyn std::future::Future<Output = CaResult<()>> + Send + 'a>> {
        Box::pin(async move {
            self.process_record_with_links_inner(name, visited, depth, false, false)
                .await
        })
    }

    /// MR-R5: recursive FLNK / OUT / CP fan-out entry within a single
    /// processing chain. Does NOT re-acquire the advisory write gate:
    /// the chain is one transaction whose entry record's gate is
    /// already held by the foreign entry, and C `processTarget`
    /// (`dbDbLink.c:436`) processes a link target under the lock set
    /// already owned by the calling thread. Re-acquiring per chain
    /// member would also create a lock-ordering deadlock between
    /// reverse FLNK chains.
    pub(crate) fn process_record_with_links_recursive<'a>(
        &'a self,
        name: &'a str,
        visited: &'a mut HashSet<String>,
        depth: usize,
    ) -> std::pin::Pin<Box<dyn std::future::Future<Output = CaResult<()>> + Send + 'a>> {
        Box::pin(async move {
            self.process_record_with_links_inner(name, visited, depth, false, false)
                .await
        })
    }

    /// Owner-driven continuation re-entry — bypasses the PACT entry guard.
    ///
    /// Used by `ProcessAction::ReprocessAfter` timer fires: the spawned
    /// re-entry task IS the owner of the async cycle, equivalent to C
    /// `callbackRequestDelayed`'s direct call to the record's `process()`
    /// (which bypasses `dbProcess`). Foreign callers must still go through
    /// `process_record_with_links` so FLNK / scan / CA put cannot race
    /// during the wait window.
    ///
    /// MR-R5: the timer fire is a fresh task — the original cycle's
    /// advisory gate was released when `process_record_with_links`
    /// returned async-pending. In C, `callbackRequestDelayed` dispatches
    /// through a callback that re-takes `dbScanLock(precord)` for the
    /// completion `process()`. This entry therefore re-acquires the
    /// advisory write gate, so the continuation cannot interleave with a
    /// QSRV atomic group or another foreign scan of the same record.
    pub fn process_record_continuation<'a>(
        &'a self,
        name: &'a str,
        visited: &'a mut HashSet<String>,
        depth: usize,
    ) -> std::pin::Pin<Box<dyn std::future::Future<Output = CaResult<()>> + Send + 'a>> {
        Box::pin(async move {
            self.process_record_with_links_inner(name, visited, depth, true, true)
                .await
        })
    }

    async fn process_record_with_links_inner(
        &self,
        name: &str,
        visited: &mut HashSet<String>,
        depth: usize,
        is_continuation: bool,
        acquire_gate: bool,
    ) -> CaResult<()> {
        const MAX_LINK_DEPTH: usize = 16;
        const MAX_LINK_OPS: usize = 256;

        // Normalise to the canonical record name once at entry — both
        // for cycle-detection (`visited` would otherwise treat alias
        // and canonical as distinct entries) and for the records-map
        // lookup below. Mirrors epics-base PR #336.
        let canonical_owned;
        let name: &str = if let Some(target) = self.resolve_alias(name).await {
            canonical_owned = target;
            &canonical_owned
        } else {
            name
        };

        if depth >= MAX_LINK_DEPTH {
            eprintln!("link chain depth limit reached at record {name}");
            return Ok(());
        }
        if visited.len() >= MAX_LINK_OPS {
            eprintln!("link chain ops budget exhausted at record {name}");
            return Ok(());
        }
        if !visited.insert(name.to_string()) {
            return Ok(()); // Cycle detected, skip
        }

        let rec = {
            let records = self.inner.records.read().await;
            records.get(name).cloned()
        };

        let rec = match rec {
            Some(r) => r,
            None => return Err(CaError::ChannelNotFound(name.to_string())),
        };

        // MR-R5: advisory write gate (`dbScanLock(precord)` analogue).
        // A foreign full-processing entry (scan loop, scan_event, FLNK
        // dispatch from another chain, CA put, PINI/startup) acquires
        // the entry record's gate so it cannot interleave with a QSRV
        // atomic group or a pvalink atomic scan epoch holding
        // `lock_records` over the same record. `name` is already the
        // alias-resolved canonical name, the same key `lock_records`
        // uses. Not acquired when `acquire_gate` is false: either a
        // transaction owner already holds the gate via `lock_records`
        // (`process_record_with_links_already_locked`), or this is a
        // recursive FLNK/OUT/CP call within one chain
        // (`process_record_with_links_recursive`) — C `processTarget`
        // processes a link target under the lock set the caller already
        // owns, and re-acquiring would deadlock the non-reentrant gate.
        let _record_gate = if acquire_gate {
            Some(self.lock_record(name).await)
        } else {
            None
        };

        // 0a. PACT entry guard — mirrors C `dbProcess` (dbAccess.c:537-559).
        // If the record is currently mid-async (PACT=true), do NOT re-enter
        // the body. Instead increment LCNT; after MAX_LOCK=10 consecutive
        // attempts raise SCAN_ALARM/INVALID with "Async in progress" and
        // post a monitor on VAL (DBE_VALUE|DBE_LOG). Up to MAX_LOCK we just
        // bail out silently so transient back-to-back scans don't immediately
        // alarm the record.
        //
        // Without this guard, FLNK / scan-loop / event scans dispatched onto
        // a record whose first cycle is still pending (async device support,
        // CA put_notify on PUTF) would re-enter `record.process()` while the
        // device's first response is still in flight — corrupting the
        // record's internal state machine and bypassing the C-parity
        // contract that callers see for `dbProcess`. The pre-existing
        // `dispatch_cp_targets` path already did this check (sets RPRO=true
        // and skips); the main entry was missing it.
        if !is_continuation {
            const MAX_LOCK: i16 = 10;
            let mut instance = rec.write().await;
            if instance.is_processing() {
                // C `dbAccess.c:539-541` — when TPRO is set on a record
                // whose PACT is true, print the diagnostic line before
                // the bail decision. The C path emits:
                //   "%s: dbProcess of Active '%s' with RPRO=%d"
                // mirroring the same context format the regular trace
                // path below uses (thread/client name + record name +
                // current RPRO bit). Without this, an operator
                // debugging a stuck async record sees NO sign that the
                // entry guard is firing — they only notice the
                // eventual SCAN_ALARM after MAX_LOCK=10 attempts.
                if instance.common.tpro {
                    eprintln!(
                        "[TPRO] {}: dbProcess of Active '{}' with RPRO={}",
                        instance.name,
                        instance.name,
                        if instance.common.rpro { 1 } else { 0 },
                    );
                }
                let stat = instance.common.stat;
                let already_invalid =
                    instance.common.sevr >= crate::server::record::AlarmSeverity::Invalid;
                let already_scan_alarm = stat == crate::server::recgbl::alarm_status::SCAN_ALARM;
                let lcnt_before = instance.common.lcnt;
                instance.common.lcnt = lcnt_before.saturating_add(1);
                if already_scan_alarm || lcnt_before < MAX_LOCK || already_invalid {
                    // Bail out without raising alarm yet.
                    return Ok(());
                }
                // Raise SCAN_ALARM/INVALID, reset alarm transition,
                // and post VAL monitor (DBE_VALUE | DBE_LOG).
                crate::server::recgbl::rec_gbl_set_sevr_msg(
                    &mut instance.common,
                    crate::server::recgbl::alarm_status::SCAN_ALARM,
                    crate::server::record::AlarmSeverity::Invalid,
                    "Async in progress",
                );
                let _ = crate::server::recgbl::rec_gbl_reset_alarms(&mut instance.common);
                // Post VAL event with VALUE | LOG mask (mirrors C
                // `db_post_events(prec, &VAL, DBE_VALUE|DBE_LOG)`).
                let mut changed_fields = Vec::new();
                if let Some(val) = instance.record.val() {
                    changed_fields.push(("VAL".to_string(), val));
                }
                changed_fields.push((
                    "SEVR".to_string(),
                    EpicsValue::Short(instance.common.sevr as i16),
                ));
                changed_fields.push((
                    "STAT".to_string(),
                    EpicsValue::Short(instance.common.stat as i16),
                ));
                // Include AMSG so subscribers reading the alarm text
                // observe "Async in progress" alongside the SCAN_ALARM
                // transition (C `recGbl.c:210-211` posts STAT and AMSG
                // together when `stat_mask` is non-zero).
                changed_fields.push((
                    "AMSG".to_string(),
                    EpicsValue::String(instance.common.amsg.clone()),
                ));
                let snapshot = crate::server::record::ProcessSnapshot {
                    changed_fields,
                    event_mask: crate::server::recgbl::EventMask::VALUE
                        | crate::server::recgbl::EventMask::LOG
                        | crate::server::recgbl::EventMask::ALARM,
                };
                drop(instance);
                let inst = rec.read().await;
                inst.notify_from_snapshot(&snapshot);
                return Ok(());
            }
            // Not pact: reset lcnt (mirrors C `else { precord->lcnt = 0; }`
            // at dbAccess.c:559) so the next async cycle starts clean.
            instance.common.lcnt = 0;
        }

        // 0. SDIS disable check — C parity dbAccess.c:562-592.
        //
        // When the SDIS link evaluates to a value equal to DISV, the
        // record is disabled and bails before record support runs. C
        // ALWAYS clears rpro/putf and triggers dbNotifyCompletion at
        // this point — regardless of whether the alarm transition
        // fires — because a disabled record must not leave behind
        // pending reprocess requests or stranded put_notify completion
        // callbacks. Pre-fix Round 4 the Rust port only reset
        // nsta/nsev and updated the alarm state, leaking rpro/putf
        // into the next cycle and stalling CA WRITE_NOTIFY callers
        // (the put_notify_tx never fired so the CA dispatcher waited
        // until socket disconnect to release the operation).
        {
            let (sdis_link, disv, diss) = {
                let instance = rec.read().await;
                (
                    instance.parsed_sdis.clone(),
                    instance.common.disv,
                    instance.common.diss,
                )
            };

            if let crate::server::record::ParsedLink::Db(ref link) = sdis_link {
                let pv_name = if link.field == "VAL" {
                    link.record.clone()
                } else {
                    format!("{}.{}", link.record, link.field)
                };
                if let Ok(val) = self.get_pv(&pv_name).await {
                    let disa_val = val.to_f64().unwrap_or(0.0) as i16;
                    let mut instance = rec.write().await;
                    instance.common.disa = disa_val;
                }
            }

            let disa = rec.read().await.common.disa;
            if disa == disv {
                let put_notify_tx = {
                    let mut instance = rec.write().await;
                    // C `dbAccess.c:575-577` — clear rpro/putf and arm
                    // notifyCompletion BEFORE the alarm check. Disabled
                    // records skip processing entirely, so any pending
                    // reprocess request is dropped (the next non-
                    // disabled cycle will pick up fresh state) and the
                    // CA put_notify caller must be released.
                    instance.common.rpro = false;
                    instance.common.putf = false;
                    let tx = instance.put_notify_tx.take();

                    // Reset nsta/nsev so stale alarm state doesn't bleed
                    // into a subsequent (re-enabled) cycle. C resets
                    // them after the sevr/stat transition; doing it
                    // first here is observationally identical because
                    // the SDIS bail short-circuits any record-support
                    // path that could read them.
                    instance.common.nsta = 0;
                    instance.common.nsev = crate::server::record::AlarmSeverity::NoAlarm;

                    // C `dbAccess.c:580-581` — if already in
                    // DISABLE_ALARM, the alarm post is skipped entirely
                    // (the alarm cycle is debounced). The rpro/putf
                    // clear above still ran, matching C's pre-`goto
                    // all_done` ordering.
                    if instance.common.stat != crate::server::recgbl::alarm_status::DISABLE_ALARM {
                        use crate::server::recgbl::EventMask;
                        instance.common.sevr = diss;
                        instance.common.stat = crate::server::recgbl::alarm_status::DISABLE_ALARM;
                        // C `dbAccess.c:586-593` posts each field with
                        // its own mask:
                        //   db_post_events(&stat, DBE_VALUE);
                        //   db_post_events(&sevr, DBE_VALUE);
                        //   db_post_events(&val,  DBE_VALUE|DBE_ALARM);
                        // STAT/SEVR get DBE_VALUE only — a DBE_ALARM-only
                        // subscriber on `.STAT`/`.SEVR` must NOT receive
                        // this disable event. Only the value field
                        // carries DBE_ALARM.
                        instance.notify_field("STAT", EventMask::VALUE);
                        instance.notify_field("SEVR", EventMask::VALUE);
                        instance.notify_field("VAL", EventMask::VALUE | EventMask::ALARM);
                    }
                    tx
                };
                // Fire dbNotifyCompletion outside the record lock —
                // C `dbAccess.c:622-623` runs it at `all_done` after
                // the disable bail. Without this, a CA WRITE_NOTIFY
                // landing on a disabled record stalls until socket
                // disconnect.
                if let Some(tx) = put_notify_tx {
                    let _ = tx.send(());
                }
                return Ok(());
            }
        }

        // 0.3. TSEL link: C `recGblGetTimeStampSimm` (recGbl.c:310-323).
        //
        // When `TSEL` is a non-constant link, C distinguishes two
        // cases by the link target field:
        //   * the link points at another record's `.TIME` field
        //     (`DBLINK_FLAG_TSELisTIME`) — copy that record's
        //     timestamp directly into `prec->time`;
        //   * otherwise `dbGetLink(&tsel, DBR_SHORT, &prec->tse)` —
        //     load `TSE` from the link before the event lookup.
        {
            let tsel_link = {
                let instance = rec.read().await;
                instance.parsed_tsel.clone()
            };
            if let crate::server::record::ParsedLink::Db(ref link) = tsel_link {
                if link.field.eq_ignore_ascii_case("TIME") {
                    // TSEL points at a `.TIME` field — copy the source
                    // record's timestamp straight into `time`. The
                    // following `apply_timestamp` leaves it untouched
                    // for TSE=-2-equivalent (device-set) semantics; we
                    // mark TSE=-2 so it is not overwritten.
                    if let Some(src) = self.get_record(&link.record).await {
                        let src_time = src.read().await.common.time;
                        let mut instance = rec.write().await;
                        instance.common.time = src_time;
                        instance.common.tse = -2;
                    }
                } else {
                    let pv_name = if link.field == "VAL" {
                        link.record.clone()
                    } else {
                        format!("{}.{}", link.record, link.field)
                    };
                    if let Ok(val) = self.get_pv(&pv_name).await {
                        let tse_val = val.to_f64().unwrap_or(0.0) as i16;
                        let mut instance = rec.write().await;
                        instance.common.tse = tse_val;
                    }
                }
            }
        }

        // 0.5. Simulation mode check.
        //
        // C `aiRecord.c:151-168`: simulation is handled inside
        // `readValue()`, then `process()` ALWAYS runs `convert` /
        // `checkAlarms` / `monitor` / `recGblFwdLink(prec)`. A
        // simulated record therefore must still run the forward-link /
        // CP / RPRO tail — only the device read and record-support
        // body are replaced by the SIOL round-trip. Returning early
        // here would silently break every FLNK / CP chain downstream
        // of any record in SIMM mode.
        match self.check_simulation_mode(&rec).await {
            SimOutcome::NotSimulated => {}
            SimOutcome::Simulated => {
                self.run_forward_link_tail(name, &rec, visited, depth).await;
                return Ok(());
            }
        }

        // 1. Read INP link value and DOL link (outside lock)
        let (inp_parsed, is_soft, dol_info) = {
            let instance = rec.read().await;
            let rtype = instance.record.record_type();

            let inp = instance.parsed_inp.clone();
            let is_soft = crate::server::device_support::is_soft_dtyp(&instance.common.dtyp);

            // DOL link info for output records with OMSL=CLOSED_LOOP.
            //
            // C parity: every record type whose DBD declares both an
            // OMSL `menuOmsl` field AND a DOL link field must honour
            // the closed-loop binding. `dfanoutRecord.c:115-122` shows
            // dfanout doing this directly via `dbGetLink(&prec->dol,
            // DBR_DOUBLE, &prec->val, ...)` when `omsl ==
            // menuOmslclosed_loop`. The Rust port previously omitted
            // `dfanout`, so a dfanout configured with OMSL=closed_loop
            // never sourced VAL from DOL — every cycle silently used
            // the previously-cached VAL, breaking any cascaded
            // setpoint-distribution chain that relied on dfanout to
            // re-read the input.
            //
            // The `aao` (array analog output) record is the only other
            // OMSL-bearing C record; the Rust port does not implement
            // aao (confirmed: no `crates/epics-base-rs/src/server/records/aao*.rs`),
            // so it is a future gap, not a same-defect-not-fixed site.
            let dol = match rtype {
                "ao" | "longout" | "int64out" | "bo" | "mbbo" | "mbboDirect" | "stringout"
                | "lso" | "dfanout" => {
                    let omsl = instance
                        .record
                        .get_field("OMSL")
                        .and_then(|v| {
                            if let EpicsValue::Short(s) = v {
                                Some(s)
                            } else {
                                None
                            }
                        })
                        .unwrap_or(0);
                    let oif = instance
                        .record
                        .get_field("OIF")
                        .and_then(|v| {
                            if let EpicsValue::Short(s) = v {
                                Some(s)
                            } else {
                                None
                            }
                        })
                        .unwrap_or(0);
                    if omsl == 1 {
                        let dol_parsed = instance
                            .record
                            .get_field("DOL")
                            .and_then(|v| {
                                if let EpicsValue::String(s) = v {
                                    Some(s)
                                } else {
                                    None
                                }
                            })
                            .map(|s| crate::server::record::parse_link_v2(&s))
                            .unwrap_or(crate::server::record::ParsedLink::None);
                        Some((dol_parsed, oif))
                    } else {
                        None
                    }
                }
                _ => None,
            };

            (inp, is_soft, dol)
        };

        // 1.1. Pre-input-link actions: actions a record needs the
        // framework to execute BEFORE any input-link fetch this cycle.
        //
        // C `devEpidSoftCallback.c:120-151`: a DB-type readback-trigger
        // (TRIG) link is written with `dbPutLink` — which synchronously
        // processes the triggered source — and only then does
        // `dbGetLink(&pepid->inp, ...)` read CVAL. The trigger write
        // must land before the `INP -> CVAL` fetch, in the same pass.
        // `pre_process_actions` runs too late (after the input-link
        // fetch below), so `pre_input_link_actions` is a strictly
        // earlier hook. The record needs `dtyp` to decide whether the
        // callback DSET is active, so push the process context first.
        {
            let pre_input_actions = {
                let mut instance = rec.write().await;
                let ctx = instance.common.process_context();
                instance.record.set_process_context(&ctx);
                instance.record.pre_input_link_actions()
            };
            if !pre_input_actions.is_empty() {
                self.execute_process_actions(name, &rec, pre_input_actions, visited, depth)
                    .await;
            }
        }

        // Read INP value
        let inp_value = self
            .read_link_value_soft(&inp_parsed, is_soft, visited, depth)
            .await;

        // epics-base PR #d0cf47c: single-INP MS-class link must also
        // propagate the source record's STAT/SEVR/AMSG just like the
        // multi-input fetch loop below does. Previously the INPA..L
        // path (calc/sub/aSub/sel) propagated alarms but plain single
        // INP (ai/bi/longin/mbbi/stringin) silently dropped them —
        // downstream MSS readers saw NoAlarm even when the source was
        // INVALID. Only fires for soft-channel records: hardware-driver
        // alarms travel through device-support's own last_alarm path.
        //
        // B2: a soft INP that is an external `pva://` / `ca://` link
        // also propagates the lset's alarm. The link string carries
        // no `MonitorSwitch` (the `?sevr=MS` modifier is stripped by
        // the parser before epics-base-rs sees it), so the lset has
        // already applied the MS/NMS/MSI gate — a `Some` LinkAlarm
        // here is one the lset decided to propagate. We fold it in as
        // `MaximizeStatus` so the gated severity AND message both
        // reach `LINK_ALARM`, matching pvxs `pvalink_lset.cpp`
        // `recGblSetSevrMsg`.
        let inp_link_alarm: Option<(
            crate::server::record::MonitorSwitch,
            super::links::LinkAlarm,
        )> = if is_soft {
            match inp_parsed {
                crate::server::record::ParsedLink::Db(ref db) => {
                    let (_v, alarm) = self.read_link_with_alarm(&inp_parsed).await;
                    alarm.map(|a| (db.monitor_switch, a))
                }
                crate::server::record::ParsedLink::Pva(_)
                | crate::server::record::ParsedLink::Ca(_) => {
                    let (_v, alarm) = self.read_link_with_alarm(&inp_parsed).await;
                    alarm.map(|a| (crate::server::record::MonitorSwitch::MaximizeStatus, a))
                }
                _ => None,
            }
        } else {
            None
        };

        // BR-R19: if the single-INP link is an external `pva://` /
        // `ca://` link configured with `time=true`, the lset returns
        // the latched upstream NT timestamp here and we adopt it
        // into the owning record's `common.time`. The lset gates the
        // option internally (returns `None` unless `time=true`), so a
        // bare connected link without the flag still produces local
        // processing time. Mirrors pvxs `pvalink_lset.cpp:427`.
        let inp_link_remote_time: Option<(i64, i32)> = match inp_parsed {
            crate::server::record::ParsedLink::Pva(ref name)
            | crate::server::record::ParsedLink::Ca(ref name) => {
                self.external_link_time(name).await
            }
            _ => None,
        };

        // Read DOL value
        let dol_value = if let Some((ref dol_parsed, _oif)) = dol_info {
            self.read_link_value(dol_parsed, visited, depth).await
        } else {
            None
        };

        // 1.5. Multi-input link fetch (calc/calcout/sel/sub)
        // Also collect alarm info from source records for MS/NMS propagation.
        let multi_input_values: Vec<(String, EpicsValue)>;
        let mut link_alarms: Vec<(
            crate::server::record::MonitorSwitch,
            super::links::LinkAlarm,
        )> = Vec::new();
        // Link fields (the `multi_input_links` first element) whose
        // fetch actually produced a value this cycle — pushed to the
        // record via `set_resolved_input_links` so its `process()` can
        // observe link-fetch success (C `RTN_SUCCESS(dbGetLink(...))`).
        let mut resolved_link_fields: Vec<&'static str> = Vec::new();
        {
            let link_info: Vec<(String, &'static str, String)> = {
                let instance = rec.read().await;
                instance
                    .record
                    .multi_input_links()
                    .iter()
                    .map(|(lf, vf)| {
                        let link_str = instance
                            .record
                            .get_field(lf)
                            .and_then(|v| {
                                if let EpicsValue::String(s) = v {
                                    Some(s)
                                } else {
                                    None
                                }
                            })
                            .unwrap_or_default();
                        (link_str, *lf, vf.to_string())
                    })
                    .collect()
            }; // read lock dropped
            let mut results = Vec::new();
            for (link_str, link_field, val_field) in &link_info {
                if !link_str.is_empty() {
                    let parsed = crate::server::record::parse_link_v2(link_str);
                    // C `dbGetLink`: a `ProcessPassive` DB input link
                    // processes its passive source record before the
                    // value is read. `read_link_with_alarm` does a bare
                    // `get_pv`, so process the source here first —
                    // matching the single-INP `read_link_value_soft`
                    // path. Without this, calc/sel/sub/aSub INPA..INPL
                    // PP links read a stale source value.
                    if let crate::server::record::ParsedLink::Db(ref db) = parsed {
                        self.process_passive_db_source(db, visited, depth).await;
                    }
                    let (value, alarm) = self.read_link_with_alarm(&parsed).await;
                    if let Some(value) = value {
                        results.push((val_field.clone(), value));
                        resolved_link_fields.push(link_field);
                    }
                    // B2: multi-input alarm propagation covers external
                    // `pva://` / `ca://` links too — see the `inp_link_alarm`
                    // comment for why the switch is `MaximizeStatus`.
                    if let Some(alarm) = alarm {
                        match &parsed {
                            crate::server::record::ParsedLink::Db(db) => {
                                link_alarms.push((db.monitor_switch, alarm));
                            }
                            crate::server::record::ParsedLink::Pva(_)
                            | crate::server::record::ParsedLink::Ca(_) => {
                                link_alarms.push((
                                    crate::server::record::MonitorSwitch::MaximizeStatus,
                                    alarm,
                                ));
                            }
                            _ => {}
                        }
                    }
                }
            }
            multi_input_values = results;
        }
        // PR #d0cf47c continued: feed the INP alarm (if any) into the
        // same `link_alarms` list the lock-section iterates over. Order
        // doesn't matter — `rec_gbl_set_sevr_msg` takes the maximum
        // severity across all sources.
        if let Some(pair) = inp_link_alarm {
            link_alarms.push(pair);
        }

        // 1.6. Sel NVL link: resolve NVL -> SELN
        let sel_nvl_value: Option<EpicsValue> = {
            let instance = rec.read().await;
            if instance.record.record_type() == "sel" {
                let nvl_str = instance
                    .record
                    .get_field("NVL")
                    .and_then(|v| {
                        if let EpicsValue::String(s) = v {
                            Some(s)
                        } else {
                            None
                        }
                    })
                    .unwrap_or_default();
                if !nvl_str.is_empty() {
                    drop(instance); // release read lock before async read
                    let parsed = crate::server::record::parse_link_v2(&nvl_str);
                    self.read_link_value(&parsed, visited, depth).await
                } else {
                    None
                }
            } else {
                None
            }
        };

        // 2. Lock record, apply INP/DOL, process, evaluate alarms, build snapshot
        let (snapshot, out_info, flnk_name, process_actions, alarm_posts) = {
            let mut instance = rec.write().await;

            // Apply DOL value for output records (OMSL=CLOSED_LOOP)
            if let Some(dol_val) = dol_value {
                let oif = dol_info.as_ref().map(|(_, oif)| *oif).unwrap_or(0);
                if oif == 1 {
                    // Incremental: VAL += DOL value
                    if let (Some(cur), Some(dol_f)) = (
                        instance.record.val().and_then(|v| v.to_f64()),
                        dol_val.to_f64(),
                    ) {
                        let _ = instance.record.set_val(EpicsValue::Double(cur + dol_f));
                    }
                } else {
                    // Full: VAL = DOL value
                    let _ = instance.record.set_val(dol_val);
                }
            }

            // Apply INP value. "Soft Channel" sets VAL directly
            // (C `read_xxx` return 2, skip RVAL→VAL conversion).
            // "Raw Soft Channel" routes the value into RVAL and lets
            // the record's RVAL→VAL convert run (epics-base
            // f2fe9d12: devBiSoftRaw applies MASK after the read).
            // Records opt into the raw path via
            // `Record::accepts_raw_soft_input` so DTYPs on records
            // that haven't wired raw soft channel stay on the legacy
            // VAL-direct path.
            let is_raw_soft = instance.common.dtyp == "Raw Soft Channel"
                && instance.record.accepts_raw_soft_input();
            let soft_inp_applied = inp_value.is_some() && !is_raw_soft;
            if let Some(inp_val) = inp_value {
                if is_raw_soft {
                    let _ = instance.record.apply_raw_input(inp_val);
                } else {
                    let _ = instance.record.set_val(inp_val);
                }
            } else if is_soft
                && matches!(
                    inp_parsed,
                    crate::server::record::ParsedLink::Db(_)
                        | crate::server::record::ParsedLink::Ca(_)
                        | crate::server::record::ParsedLink::Pva(_)
                )
            {
                // epics-base PR #4737901: soft-channel `read_xxx` must
                // surface link-read failures via the alarm tree, not
                // silently succeed. When the INP link is a real
                // Db/Ca/Pva link (i.e. operator expected a value) and
                // the read returned None, attach LINK_ALARM/INVALID
                // so downstream consumers can react. ParsedLink::None
                // and Constant don't fall into this branch — the
                // former is "no link configured", the latter has its
                // own None-as-no-value semantics.
                use crate::server::recgbl::{alarm_status, rec_gbl_set_sevr};
                rec_gbl_set_sevr(
                    &mut instance.common,
                    alarm_status::LINK_ALARM,
                    crate::server::record::AlarmSeverity::Invalid,
                );
            }

            // Apply multi-input values (INPA..INPL -> A..L).
            //
            // Uses `put_field_internal`, not `put_field`: this is the
            // framework writing a resolved input-link value into a
            // record field, exactly like the `ReadDbLink` apply
            // (`execute_read_db_links` / `execute_process_actions`),
            // which already routes through `put_field_internal`. Some
            // records map an input link to a normally read-only field
            // — e.g. the epid record's `INP -> CVAL` — and `put_field`
            // rejects those with `ReadOnlyField`, silently dropping the
            // value. `put_field_internal` defaults to `put_field`, so
            // records with writable targets (calc/sub `A..L`) are
            // unaffected.
            for (val_field, value) in &multi_input_values {
                if let Some(f) = value.to_f64() {
                    let _ = instance
                        .record
                        .put_field_internal(val_field, EpicsValue::Double(f));
                }
            }

            // Tell the record which input link fields actually resolved
            // a value this cycle — the framework analogue of C device
            // support inspecting `RTN_SUCCESS(dbGetLink(...))`
            // (`epidRecord.c:191-193`).
            instance
                .record
                .set_resolved_input_links(&resolved_link_fields);

            // Apply sel NVL -> SELN
            if let Some(nvl_val) = sel_nvl_value {
                if let Some(f) = nvl_val.to_f64() {
                    let _ = instance
                        .record
                        .put_field("SELN", EpicsValue::Short(f as i16));
                }
            }

            // Device support read (input records only, not output records)
            let is_soft = instance.common.dtyp.is_empty() || instance.common.dtyp == "Soft Channel";
            let is_output = instance.record.can_device_write();
            let mut device_actions: Vec<crate::server::record::ProcessAction> = Vec::new();
            // C `devAiSoft.c:65` `read_ai` (and the other soft-channel
            // input `read_xxx`) ALWAYS returns 2 ("don't convert") for a
            // Soft-Channel input record — whether the value arrived via
            // an INP link or the INP link is constant/unset
            // (`dbLinkIsConstant` → `return 2`). Only `aiRecord.c:158`'s
            // `if (status==0) convert(prec)` runs RVAL→VAL conversion, so
            // for a plain Soft-Channel input record `convert()` must be
            // skipped unconditionally. Without this, a soft ai with no
            // INP would run `convert()` and clobber a preset VAL — e.g.
            // a preset NaN would be rewritten to 0.0, then the framework
            // UDF check (`value_is_undefined()`) would see a defined 0.0
            // and wrongly clear UDF (06-H-2 regression). `is_raw_soft`
            // (Raw Soft Channel, `devAiSoftRaw` returns 0) is excluded —
            // it deliberately wants the RVAL→VAL convert.
            //
            // Gated on `soft_channel_skips_convert()` so this only
            // suppresses an `RVAL → VAL` convert step. Records such as
            // `epid` also override `set_device_did_compute` but treat it
            // as "skip the whole built-in compute" (the PID loop); they
            // return `false` here so a Soft-Channel `epid` still runs
            // `do_pid()` in `process()`.
            let soft_input_skips_convert = is_soft
                && !is_output
                && !is_raw_soft
                && instance.record.soft_channel_skips_convert();
            let mut device_did_compute = (soft_inp_applied && is_soft) || soft_input_skips_convert;
            if !is_soft && !is_output {
                if let Some(mut dev) = instance.device.take() {
                    // Push framework-owned common state (PHAS/TSE/TSEL/
                    // UDF) so device support's read() can see it — C
                    // device support reads `dbCommon` directly
                    // (`devTimeOfDay.c:122` uses `psi->phas`).
                    dev.set_process_context(&instance.common.process_context());
                    match dev.read(&mut *instance.record) {
                        Ok(read_outcome) => {
                            device_did_compute = read_outcome.did_compute;
                            device_actions = read_outcome.actions;
                        }
                        Err(e) => {
                            eprintln!("device read error on {}: {e}", instance.name);
                            use crate::server::recgbl::{alarm_status, rec_gbl_set_sevr};
                            rec_gbl_set_sevr(
                                &mut instance.common,
                                alarm_status::READ_ALARM,
                                crate::server::record::AlarmSeverity::Invalid,
                            );
                        }
                    }
                    instance.device = Some(dev);
                }
            }

            // Pre-process actions: execute ReadDbLink from device support and
            // record's pre_process_actions() BEFORE process() so the values
            // are immediately available. Matches C dbGetLink() semantics.
            let mut pre_actions = instance.record.pre_process_actions();
            // Also collect ReadDbLink from device actions
            let mut deferred_device_actions = Vec::new();
            for action in device_actions {
                if matches!(
                    action,
                    crate::server::record::ProcessAction::ReadDbLink { .. }
                ) {
                    pre_actions.push(action);
                } else {
                    deferred_device_actions.push(action);
                }
            }
            if !pre_actions.is_empty() {
                let rec_name = instance.name.clone();
                drop(instance);
                self.execute_read_db_links(&rec_name, &rec, &pre_actions, visited, depth)
                    .await;
                instance = rec.write().await;
            }

            // Note: C EPICS LCNT prevents reentrant processing of the same
            // record within a single processing chain. In Rust, this is handled
            // by the `visited` HashSet (cycle detection) and the `processing`
            // AtomicBool guard. LCNT is not needed as a separate mechanism
            // because async processing with visited sets already prevents
            // the runaway loops that LCNT guards against in C.

            // Tell the record whether device support already computed.
            // Records that override set_device_did_compute() use this to
            // skip their built-in computation (e.g., ai skips RVAL->VAL).
            // Note: field_io.rs may have already called set_device_did_compute(true)
            // for CA puts to VAL. We only set true here, never reset to false.
            if device_did_compute {
                instance.record.set_device_did_compute(true);
            }

            // TPRO: trace processing (C EPICS dbProcess prints context when TPRO>0)
            if instance.common.tpro {
                eprintln!(
                    "[TPRO] {}: process (SCAN={:?}, PACT={})",
                    instance.name,
                    instance.common.scan,
                    instance
                        .processing
                        .load(std::sync::atomic::Ordering::Relaxed)
                );
            }

            // Push framework-owned common state (UDF/PHAS/TSE/TSEL) so
            // the record's process() can see it — C records read
            // `dbCommon` directly (`epidRecord.c:195` checks
            // `pepid->udf`, `timestampRecord.c:90` checks `tse`).
            {
                let ctx = instance.common.process_context();
                instance.record.set_process_context(&ctx);
            }

            // Process
            let mut outcome = instance.record.process()?;
            // Merge deferred device actions into process outcome actions
            outcome.actions.extend(deferred_device_actions);
            let process_result = outcome.result;
            let process_actions = outcome.actions;

            if process_result == crate::server::record::RecordProcessResult::AsyncPending {
                // C `dbProcess` contract: when device support / record body
                // signals "async pending", `pact` MUST be true so subsequent
                // dbProcess attempts on the same record bail at the entry
                // guard. Previous Rust port assumed `process_local` had
                // already set it via the swap-true at function entry, but
                // this main path bypasses `process_local` and calls
                // `record.process()` directly — leaving `processing=false`.
                // Mirrors `aiRecord.c:122` and similar: `prec->pact = TRUE;
                // return 0;` before async work.
                instance
                    .processing
                    .store(true, std::sync::atomic::Ordering::Release);

                // PACT stays set; skip alarm/timestamp/snapshot/OUT/FLNK.
                // But still execute any actions (e.g., ReprocessAfter for delayed re-entry).
                let rec_name = instance.name.clone();
                drop(instance);
                self.execute_process_actions(&rec_name, &rec, process_actions, visited, depth)
                    .await;
                return Ok(());
            }
            if let crate::server::record::RecordProcessResult::AsyncPendingNotify(fields) =
                process_result
            {
                // Intermediate notification (e.g. DMOV=0 at move start).
                // Execute device write first so the move command reaches the driver,
                // then flush DMOV=0 etc. to monitors.
                if !is_soft {
                    if let Some(mut dev) = instance.device.take() {
                        let _ = dev.write(&mut *instance.record);
                        instance.device = Some(dev);
                    }
                }
                apply_timestamp(&mut instance.common, is_soft);
                // Filter out fields that haven't changed, update MLST/last_posted.
                let mut changed_fields = Vec::new();
                for (name, val) in fields {
                    let changed = match instance.last_posted.get(&name) {
                        Some(prev) => prev != &val,
                        None => true,
                    };
                    if changed {
                        if name == "VAL" {
                            if let Some(f) = val.to_f64() {
                                instance.put_coerced("MLST", f);
                                instance.common.mlst = Some(f);
                            }
                        }
                        instance.last_posted.insert(name.clone(), val.clone());
                        changed_fields.push((name, val));
                    }
                }
                let event_mask = if changed_fields.is_empty() {
                    crate::server::recgbl::EventMask::NONE
                } else {
                    crate::server::recgbl::EventMask::VALUE
                        | crate::server::recgbl::EventMask::ALARM
                };
                let snapshot = crate::server::record::ProcessSnapshot {
                    changed_fields,
                    event_mask,
                };
                let rec_clone = rec.clone();
                drop(instance);
                {
                    let inst = rec_clone.read().await;
                    inst.notify_from_snapshot(&snapshot);
                }
                return Ok(());
            }

            // Async-completion PACT clear for the `ReprocessAfter`
            // continuation path. C parity `dbAccess.c:583` —
            // `prset->process(precord)` for a record whose first cycle
            // returned async-pending is the *completion* re-entry; the
            // record support clears `pact` itself inside `process()`
            // (e.g. `aiRecord.c` second pass sets `prec->pact = FALSE`).
            //
            // A record that returns `AsyncPending` AND emits a
            // `ProcessAction::ReprocessAfter` is re-entered here via
            // `process_record_continuation` (`is_continuation == true`,
            // PACT entry guard skipped). Reaching this point means the
            // continuation's `process()` did NOT return async-pending
            // again (both async branches above return early), so the
            // async cycle is genuinely complete. The non-continuation
            // async-device path clears `processing` in
            // `complete_async_record_inner`; the continuation path has
            // no such callback, so without this clear `processing`
            // stays `true` forever — every later foreign
            // `process_record_with_links` then trips the PACT entry
            // guard, counts to MAX_LOCK, and raises a spurious
            // SCAN_ALARM. Clearing here (record still write-locked,
            // before the OUT/FLNK tail) mirrors the C ordering where
            // `pact` is already `FALSE` when `recGblFwdLink` runs.
            if is_continuation {
                instance
                    .processing
                    .store(false, std::sync::atomic::Ordering::Release);
            }

            // MS-class alarm propagation from input links. Mirrors C
            // `recGblInheritSevrMsg` (recGbl.c::260):
            //
            // * NMS  — do nothing.
            // * MS   — DEST gets `LINK_ALARM` (NOT the source stat),
            //          max-raised sevr, NO amsg propagation.
            // * MSI  — same as MS, but only when source.sevr == INVALID.
            // * MSS  — DEST gets source stat, max-raised sevr, source amsg
            //          (PR d0cf47c is the only branch that propagates msg).
            //
            // Previous version treated Maximize and MaximizeStatus
            // identically, propagating source stat + amsg through both
            // — that matches MSS but is wrong for MS (and MSI), which
            // C says should always surface as LINK_ALARM with no msg.
            for (ms, alarm) in &link_alarms {
                use crate::server::recgbl::{alarm_status, rec_gbl_set_sevr, rec_gbl_set_sevr_msg};
                use crate::server::record::MonitorSwitch;
                match ms {
                    MonitorSwitch::Maximize => {
                        rec_gbl_set_sevr(
                            &mut instance.common,
                            alarm_status::LINK_ALARM,
                            alarm.sevr,
                        );
                    }
                    MonitorSwitch::MaximizeIfInvalid => {
                        if alarm.sevr == crate::server::record::AlarmSeverity::Invalid {
                            rec_gbl_set_sevr(
                                &mut instance.common,
                                alarm_status::LINK_ALARM,
                                alarm.sevr,
                            );
                        }
                    }
                    MonitorSwitch::MaximizeStatus => {
                        rec_gbl_set_sevr_msg(
                            &mut instance.common,
                            alarm.stat,
                            alarm.sevr,
                            alarm.amsg.clone(),
                        );
                    }
                    MonitorSwitch::NoMaximize => {} // NMS: do not propagate
                }
            }

            // UDF update — C parity (aiRecord.c:285, calcRecord.c
            // checkAlarms, int64inRecord.c:144): clear UDF only when
            // this cycle produced a *defined* value. A NaN computed
            // value (calc divide-by-zero) or a failed link read that
            // left VAL un-updated must keep UDF true so the following
            // `recGblCheckUDF` raises UDF_ALARM at severity UDFS.
            //
            // This MUST run before `evaluate_alarms()` (which calls
            // `rec_gbl_check_udf`): C records set `prec->udf` inside
            // `process()` before `checkAlarms()` runs.
            if instance.record.clears_udf() {
                instance.common.udf = instance.record.value_is_undefined();
            }

            // Per-record alarm hook — record-type-specific STATE / COS
            // / limit / SOFT alarms (C `checkAlarms()`). Records that
            // have migrated their alarm logic here raise into
            // `nsta`/`nsev`; the rest fall back to the framework's
            // centralised `evaluate_alarms` match below.
            {
                let inst = &mut *instance;
                inst.record.check_alarms(&mut inst.common);
            }

            // Evaluate alarms (accumulates into nsta/nsev)
            instance.evaluate_alarms();

            // Device support alarm/timestamp override
            if !is_soft {
                let (dev_alarm, dev_ts) = if let Some(ref dev) = instance.device {
                    (dev.last_alarm(), dev.last_timestamp())
                } else {
                    (None, None)
                };
                if let Some((stat, sevr)) = dev_alarm {
                    use crate::server::recgbl::rec_gbl_set_sevr;
                    rec_gbl_set_sevr(
                        &mut instance.common,
                        stat,
                        crate::server::record::AlarmSeverity::from_u16(sevr),
                    );
                }
                if let Some(ts) = dev_ts {
                    instance.common.time = ts;
                }
            }

            // BR-R19: pvalink `time=true` adopts the latched upstream
            // NT timestamp into the owning record. `external_link_time`
            // returned `None` unless the lset signalled the option, so
            // a `Some` here is the operator-requested remote timestamp.
            // Apply BEFORE `apply_timestamp` so the upstream value
            // survives the soft-channel TSE=0 default (`apply_timestamp`
            // would otherwise stamp wall-clock-now on top).
            if let Some((secs, ns)) = inp_link_remote_time {
                let secs = secs.max(0) as u64;
                let ns = ns.max(0) as u32;
                instance.common.time =
                    std::time::UNIX_EPOCH + std::time::Duration::new(secs, ns.min(999_999_999));
                // TSE=-2 marks "device-set time" — `apply_timestamp`
                // honours this by leaving `common.time` untouched,
                // mirroring the device-support timestamp branch above.
                instance.common.tse = -2;
            }

            // Transfer nsta/nsev -> sevr/stat, detect alarm change
            let alarm_result = crate::server::recgbl::rec_gbl_reset_alarms(&mut instance.common);

            // Apply timestamp based on TSE
            apply_timestamp(&mut instance.common, is_soft);
            // NOTE: UDF was already updated before `evaluate_alarms`
            // above — keyed on `value_is_undefined()` so a NaN result
            // keeps UDF true and UDF_ALARM is raised this cycle. Do
            // NOT clear UDF unconditionally here.

            // IVOA check for output records with INVALID alarm
            let skip_out = if instance.common.sevr == crate::server::record::AlarmSeverity::Invalid
            {
                let ivoa = instance
                    .record
                    .get_field("IVOA")
                    .and_then(|v| {
                        if let EpicsValue::Short(s) = v {
                            Some(s)
                        } else {
                            None
                        }
                    })
                    .unwrap_or(0);
                match ivoa {
                    1 => true, // Don't drive outputs
                    2 => {
                        // Set output to IVOV. Each record type knows
                        // which field its OUT writeback consumes — see
                        // [`Record::apply_invalid_output_value`]. The
                        // pre-Round-30C path special-cased `calcout`
                        // (OVAL) and fell back to `set_val` (VAL) for
                        // every other record. That hid a real bug:
                        // ao/lso/bo/mbbo/busy left their OVAL/RVAL
                        // staging field stale, so the OUT writeback —
                        // which reads `OVAL.or(VAL)` — sent the
                        // pre-IVOA value to the linked record. Per-type
                        // overrides now apply IVOV to the field that
                        // matches the C convention.
                        if let Some(ivov) = instance.record.get_field("IVOV") {
                            let _ = instance.record.apply_invalid_output_value(ivov);
                        }
                        false
                    }
                    _ => false, // Continue normally
                }
            } else {
                false
            };

            // OUT stage: soft channel -> link put, non-soft -> device.write()
            // Must run BEFORE check_deadband_ext so MLST is not prematurely
            // updated for async writes that return early.
            let can_dev_write = instance.record.can_device_write();
            let is_soft_out =
                instance.common.dtyp.is_empty() || instance.common.dtyp == "Soft Channel";
            let record_should_output = instance.record.should_output();
            let out_info = if skip_out {
                None
            } else if !can_dev_write {
                // Non-output records (calcout, etc.) may still have a
                // soft OUT link (DB or external ca://`/`pva://`).
                // Write OVAL to OUT when the record says should_output().
                if record_should_output && instance.parsed_out.is_writable_out_link() {
                    let oval = instance.record.get_field("OVAL");
                    let val = instance.record.val();
                    let out_val = oval.or(val);
                    out_val.map(|v| (instance.parsed_out.clone(), v))
                } else {
                    None
                }
            } else if is_soft_out {
                if !record_should_output {
                    // epics-base 7.0.8 OOPT: gate the soft OUT-link
                    // write on the record's `should_output()`. For
                    // longout/calcout with OOPT != 0 this lets a
                    // condition-not-met cycle silently skip the link
                    // write without disturbing alarms / monitors.
                    None
                } else if instance.parsed_out.is_writable_out_link() {
                    let out_val = instance
                        .record
                        .get_field("OVAL")
                        .or_else(|| instance.record.val());
                    out_val.map(|v| (instance.parsed_out.clone(), v))
                } else {
                    None
                }
            } else if !record_should_output {
                // OOPT gating for hardware outputs (longout DTYP=...).
                // Skip the device write when the OOPT predicate is
                // not satisfied; the record's val/timestamp/snapshot
                // path still runs so monitor consumers see the value
                // change even on a non-output cycle.
                None
            } else {
                if let Some(mut dev) = instance.device.take() {
                    // Try async write_begin() first
                    match dev.write_begin(&mut *instance.record) {
                        Ok(Some(completion)) => {
                            // Async write submitted -- set PACT, return early.
                            // complete_async_record will handle deadband, snapshot,
                            // notification, and FLNK when the write completes.
                            instance
                                .processing
                                .store(true, std::sync::atomic::Ordering::Release);
                            instance.device = Some(dev);
                            let rec_name = instance.name.clone();
                            let timeout = std::time::Duration::from_secs(5);
                            let db = self.clone();
                            tokio::spawn(async move {
                                let _ =
                                    tokio::task::spawn_blocking(move || completion.wait(timeout))
                                        .await;
                                let _ = db.complete_async_record(&rec_name).await;
                            });
                            return Ok(());
                        }
                        Ok(None) => {
                            // No async support -- fall back to synchronous write
                            if let Err(e) = dev.write(&mut *instance.record) {
                                eprintln!("device write error on {}: {e}", instance.name);
                                instance.common.stat =
                                    crate::server::recgbl::alarm_status::WRITE_ALARM;
                                instance.common.sevr =
                                    crate::server::record::AlarmSeverity::Invalid;
                            } else {
                                // OOPT 7.0.8: notify the record so it can
                                // latch transition state (e.g. longout.pval)
                                // for the next cycle.
                                instance.record.on_output_complete();
                            }
                        }
                        Err(e) => {
                            eprintln!("device write_begin error on {}: {e}", instance.name);
                            instance.common.stat = crate::server::recgbl::alarm_status::WRITE_ALARM;
                            instance.common.sevr = crate::server::record::AlarmSeverity::Invalid;
                        }
                    }
                    instance.device = Some(dev);
                }
                None
            };

            // Compute event mask (after OUT stage so async writes don't
            // update MLST/ALST prematurely before returning early)
            use crate::server::recgbl::EventMask;
            let mut event_mask = EventMask::NONE;

            let (include_val, include_archive) = if instance.record.uses_monitor_deadband() {
                instance.check_deadband_ext()
            } else {
                // Binary records (bi/bo/busy/mbbi/mbbo): always post monitors
                (true, true)
            };
            if include_val {
                event_mask |= EventMask::VALUE;
            }
            if include_archive {
                event_mask |= EventMask::LOG;
            }
            if alarm_result.alarm_changed || alarm_result.amsg_changed {
                // C `recGbl.c:194/203` — amsg-only OR sevr-changed sets
                // `stat_mask = DBE_ALARM`, which raises `val_mask =
                // DBE_ALARM` at line 212. Without this, an alarm whose
                // sevr/stat is unchanged but whose amsg shifted (e.g.
                // device re-flagging the same severity with a different
                // human-readable cause) silently drops the AMSG event
                // and any DBE_ALARM-only subscribers stay stale.
                event_mask |= EventMask::ALARM;
            }

            // Build snapshot
            let mut changed_fields = Vec::new();
            // C `recGblResetAlarms` returns `val_mask = DBE_ALARM`
            // when any alarm field moved — the record's VAL is posted
            // with DBE_ALARM even if the value deadband did not fire,
            // so a `DBE_ALARM`-only subscriber sees the value at the
            // moment the alarm changed.
            let val_on_alarm = alarm_result.alarm_changed || alarm_result.amsg_changed;
            if include_val || val_on_alarm {
                if let Some(val) = instance.record.val() {
                    changed_fields.push(("VAL".to_string(), val));
                }
            }
            // Add subscribed fields that actually changed since last notification.
            let mut sub_updates: Vec<(String, EpicsValue)> = Vec::new();
            for (field, subs) in &instance.subscribers {
                if !subs.is_empty()
                    && field != "VAL"
                    && field != "SEVR"
                    && field != "STAT"
                    && field != "AMSG"
                    && field != "UDF"
                {
                    if let Some(val) = instance.resolve_field(field) {
                        let changed = match instance.last_posted.get(field) {
                            Some(prev) => prev != &val,
                            None => true,
                        };
                        if changed {
                            sub_updates.push((field.clone(), val));
                        }
                    }
                }
            }
            if !sub_updates.is_empty() {
                for (field, val) in &sub_updates {
                    instance.last_posted.insert(field.clone(), val.clone());
                }
                changed_fields.extend(sub_updates);
                event_mask |= crate::server::recgbl::EventMask::VALUE;
            }
            // C `recGblResetAlarms` (recGbl.c:201-220) posts each
            // alarm field with its own per-field mask:
            //   * SEVR — DBE_VALUE, ONLY when `prev_sevr != new_sevr`.
            //   * STAT/AMSG — `stat_mask` = DBE_ALARM (on sevr- or
            //     amsg-change) | DBE_VALUE (on stat-change).
            //   * ACKS — DBE_VALUE when `stat_mask != 0`.
            // The pre-fix port pushed SEVR + STAT together on any
            // `alarm_changed`, over-posting SEVR on a stat-only
            // transition and collapsing the per-field mask into one
            // record-wide mask. Posting these via `notify_field` with
            // their individual masks restores C's granularity.
            let sevr_changed = instance.common.sevr != alarm_result.prev_sevr;
            let stat_changed = instance.common.stat != alarm_result.prev_stat;
            let stat_mask = {
                let mut m = EventMask::NONE;
                if sevr_changed || alarm_result.amsg_changed {
                    m |= EventMask::ALARM;
                }
                if stat_changed {
                    m |= EventMask::VALUE;
                }
                m
            };
            if !stat_mask.is_empty() {
                // C `val_mask = DBE_ALARM` — the value field carries
                // DBE_ALARM whenever any alarm field moved.
                event_mask |= EventMask::ALARM;
            }
            // Defer the SEVR/STAT/AMSG/ACKS posts to dedicated
            // `notify_field` calls (collected here, fired after the
            // snapshot notify below) so each gets its exact C mask.
            let mut alarm_posts: Vec<(&'static str, EventMask)> = Vec::new();
            if sevr_changed {
                alarm_posts.push(("SEVR", EventMask::VALUE));
            }
            if !stat_mask.is_empty() {
                alarm_posts.push(("STAT", stat_mask));
                alarm_posts.push(("AMSG", stat_mask));
            }
            // C parity (recGbl.c:216): ACKS is posted (DBE_VALUE) only
            // when `stat_mask != 0` AND recGblResetAlarms raised it.
            if alarm_result.acks_changed && !stat_mask.is_empty() {
                alarm_posts.push(("ACKS", EventMask::VALUE));
            }
            if !event_mask.is_empty() {
                changed_fields.push((
                    "UDF".to_string(),
                    EpicsValue::Char(if instance.common.udf { 1 } else { 0 }),
                ));
            }
            let snapshot = crate::server::record::ProcessSnapshot {
                changed_fields,
                event_mask,
            };

            let flnk_name = if instance.record.should_fire_forward_link() {
                if let crate::server::record::ParsedLink::Db(ref l) = instance.parsed_flnk {
                    Some(l.record.clone())
                } else {
                    None
                }
            } else {
                None
            };

            // Fire deferred put_notify completion when the record reports
            // that async work is done (e.g. motor: DMOV=1).
            if instance.put_notify_tx.is_some() && instance.record.is_put_complete() {
                if let Some(tx) = instance.put_notify_tx.take() {
                    let _ = tx.send(());
                }
            }

            (snapshot, out_info, flnk_name, process_actions, alarm_posts)
        };

        // 3. Notify subscribers (outside lock)
        {
            let instance = rec.read().await;
            instance.notify_from_snapshot(&snapshot);
            // Post the alarm fields (SEVR/STAT/AMSG/ACKS) with their
            // individual C masks — see recGblResetAlarms above.
            for &(field, mask) in &alarm_posts {
                instance.notify_field(field, mask);
            }
        }

        // Snapshot source PUTF for the C `processTarget` invariant (see
        // `write_db_link_value` doc). Captured once here so every OUT /
        // multi-OUT / FLNK dispatch in this cycle propagates the same bit.
        let src_putf = rec.read().await.common.putf;

        // 4. OUT link — DB *or* external `ca://`/`pva://`. C
        // `dbLink.c::dbPutLink` (dbLink.c:434-448) routes every link
        // write through the link set's `putValue`, so the OUTPUT side
        // dispatches by scheme exactly as the INPUT side does (B
        // `resolve_external_pv`). An external link with no registered
        // lset fails gracefully inside `write_out_link_value`.
        if let Some((ref link, ref out_val)) = out_info {
            self.write_out_link_value(link, out_val.clone(), src_putf, visited, depth)
                .await;
            // OOPT 7.0.8: latch the record's post-output state so the
            // next cycle's `should_output` sees the right pval.
            {
                let mut instance = rec.write().await;
                instance.record.on_output_complete();
            }
        }

        // 4.5 - 7. Multi-output / event / generic-multi-out / FLNK /
        // CP / RPRO tail. Shared with the simulation-mode path so a
        // simulated record runs the exact same `recGblFwdLink`
        // equivalent (C `aiRecord.c:168`).
        self.run_forward_link_tail_with_putf(
            name,
            &rec,
            flnk_name.as_deref(),
            src_putf,
            visited,
            depth,
        )
        .await;

        // 8. Execute ProcessActions from the record's process() outcome.
        // This handles WriteDbLink, ReadDbLink, and ReprocessAfter actions.
        self.execute_process_actions(name, &rec, process_actions, visited, depth)
            .await;

        // 9. C `recGbl.c::recGblFwdLink:302` clears `putf = FALSE` at the
        // tail of every synchronous process cycle, NOT just on the
        // foreign-entry path. When this record was driven through an
        // OUT-link propagation (write_db_link_value set our putf), the
        // target record's own process cycle must clear it before
        // returning — same lifecycle as the source record's PUTF
        // (which `put_record_field_from_ca` separately clears at the
        // foreign-entry boundary, and the async branch clears in
        // `complete_async_record_inner`). Async-pending records skip
        // this clear: their FLNK / putf-clear happens later in
        // `complete_async_record_inner` once the device round-trip
        // completes.
        {
            let guard = rec.read().await;
            if !guard.is_processing() {
                drop(guard);
                let mut guard = rec.write().await;
                guard.common.putf = false;
            }
        }

        Ok(())
    }

    /// Forward-link / CP / RPRO tail for the simulation-mode path.
    ///
    /// C `aiRecord.c:151-168`: a record in SIMM mode handles the value
    /// inside `readValue()`, then `process()` still runs `monitor` +
    /// `recGblFwdLink(prec)`. The simulation path in
    /// `process_record_with_links_inner` does its own monitor posting,
    /// so this drives the forward-link / CP / RPRO tail that
    /// `recGblFwdLink` would. `flnk_name` and `src_putf` are derived
    /// fresh from the record (a simulated cycle does not change FLNK,
    /// and SIOL reads/writes do not carry a foreign PUTF into the
    /// chain).
    async fn run_forward_link_tail(
        &self,
        name: &str,
        rec: &Arc<RwLock<RecordInstance>>,
        visited: &mut std::collections::HashSet<String>,
        depth: usize,
    ) {
        let (flnk_name, src_putf) = {
            let instance = rec.read().await;
            let flnk = if instance.record.should_fire_forward_link() {
                if let crate::server::record::ParsedLink::Db(ref l) = instance.parsed_flnk {
                    Some(l.record.clone())
                } else {
                    None
                }
            } else {
                None
            };
            (flnk, instance.common.putf)
        };
        self.run_forward_link_tail_with_putf(
            name,
            rec,
            flnk_name.as_deref(),
            src_putf,
            visited,
            depth,
        )
        .await;
    }

    /// Steps 4.5 - 7 of the process chain: multi-output dispatch,
    /// event-record posting, generic OUTA..OUTP links, FLNK forward
    /// link, CP-target dispatch, and RPRO reprocess. Shared by the
    /// main process path and the simulation-mode path so both run the
    /// identical `recGblFwdLink` equivalent.
    async fn run_forward_link_tail_with_putf(
        &self,
        name: &str,
        rec: &Arc<RwLock<RecordInstance>>,
        flnk_name: Option<&str>,
        src_putf: bool,
        visited: &mut std::collections::HashSet<String>,
        depth: usize,
    ) {
        // 4.5. Multi-output dispatch (fanout/dfanout/seq)
        self.dispatch_multi_output(rec, visited, depth).await;

        // 4.55. event record: post the named software event.
        self.dispatch_event_record(rec).await;

        // 4.6. Generic multi-output links (transform OUTA..OUTP -> A..P,
        // scalcout OUT->OVAL, epid OUTL).
        //
        // SINGLE-OWNER INVARIANT: a record type whose link groups are
        // dispatched by `dispatch_multi_output` (§4.5 above) MUST be
        // skipped here — otherwise its `LNKn`/`OUTn` would be written
        // twice per cycle. `sseq` previously also implemented the
        // `Record::multi_output_links` trait method, so this block
        // re-dispatched every selected `LNKn` after §4.5 already drove
        // it. The `multi_output_dispatch_owned` gate makes the
        // double-dispatch structurally impossible — not just removed
        // at the `SseqRecord` call site.
        {
            let multi_out = {
                let instance = rec.read().await;
                let links =
                    if super::links::multi_output_dispatch_owned(instance.record.record_type()) {
                        &[][..]
                    } else {
                        instance.record.multi_output_links()
                    };
                if links.is_empty() {
                    None
                } else {
                    let mut pairs = Vec::new();
                    for &(link_field, val_field) in links {
                        let link_str = instance
                            .record
                            .get_field(link_field)
                            .and_then(|v| {
                                if let EpicsValue::String(s) = v {
                                    Some(s)
                                } else {
                                    None
                                }
                            })
                            .unwrap_or_default();
                        if link_str.is_empty() {
                            continue;
                        }
                        if let Some(val) = instance.record.get_field(val_field) {
                            pairs.push((link_str, val));
                        }
                    }
                    if pairs.is_empty() { None } else { Some(pairs) }
                }
            };
            if let Some(pairs) = multi_out {
                for (link_str, val) in pairs {
                    // `multi_output_links` carries record OUT links
                    // (sseq `LNKn`, scalcout `OUTn` — all `DBF_OUTLINK`)
                    // driven via `dbPutLink` → `dbDbPutValue`
                    // (`dbDbLink.c:388`): a bare DB link is NPP, the
                    // value is written but the target is NOT processed.
                    // `parse_output_link_v2` applies the
                    // OUT-link-correct NPP default; `parse_link_v2` would
                    // wrongly default a bare link to ProcessPassive and
                    // re-process the target. An external `ca://`/`pva://`
                    // OUT link is routed through the link set's
                    // `putValue` (C `dbLink.c::dbPutLink`,
                    // dbLink.c:434-448).
                    let parsed = crate::server::record::parse_output_link_v2(&link_str);
                    self.write_out_link_value(&parsed, val, src_putf, visited, depth)
                        .await;
                }
            }
        }

        // 5. FLNK -- only process if target is Passive (like C dbScanFwdLink).
        // FLNK goes through C `dbScanPassive` -> `processTarget`, which
        // propagates `src_putf` to the target the same way OUT links do.
        if let Some(flnk) = flnk_name {
            if let Some(target_rec) = self.get_record(flnk).await {
                let (target_scan, should_process) = {
                    let mut tg = target_rec.write().await;
                    let pact = tg.is_processing();
                    let on_chain = visited.contains(flnk);
                    if !pact {
                        tg.common.putf = src_putf;
                    } else if src_putf && !on_chain {
                        tg.common.rpro = true;
                        tg.common.putf = false;
                    }
                    (tg.common.scan, !pact)
                };
                if should_process && target_scan == crate::server::record::ScanType::Passive {
                    // MR-R5: recursive FLNK within one chain — gate
                    // already held by the foreign entry record.
                    let _ = self
                        .process_record_with_links_recursive(flnk, visited, depth + 1)
                        .await;
                }
            }
        }

        // 6. CP link targets -- process records that have CP input links from this record
        self.dispatch_cp_targets(name, visited, depth).await;

        // 7. RPRO: if reprocess requested, clear flag and queue a
        // fresh process pass.
        //
        // C `recGblFwdLink` (recGbl.c:296-300) consumes RPRO via
        // `scanOnce(pdbc)` — the record is QUEUED on the scanOnce ring
        // buffer and reprocessed in a separate pass with a fresh lock
        // cycle AFTER the current process chain fully unwinds. It does
        // NOT recurse inline within the current link chain.
        //
        // Spawning a detached task is the Rust equivalent of the
        // scanOnce queue: the reprocess runs with a clean (empty)
        // `visited` set and starts at depth 0, so it cannot be
        // silently skipped by the current chain's cycle guard nor hit
        // the MAX_LINK_DEPTH / MAX_LINK_OPS budget the current chain
        // has already consumed.
        {
            let needs_rpro = {
                let mut instance = rec.write().await;
                if instance.common.rpro {
                    instance.common.rpro = false;
                    true
                } else {
                    false
                }
            };
            if needs_rpro {
                let db = self.clone();
                let rpro_name = name.to_string();
                crate::runtime::task::spawn(async move {
                    let mut fresh_visited = std::collections::HashSet::new();
                    let _ = db
                        .process_record_with_links(&rpro_name, &mut fresh_visited, 0)
                        .await;
                });
            }
        }
    }

    /// Execute ReadDbLink actions before process().
    /// Reads linked PV values and writes them into record fields via put_field_internal.
    async fn execute_read_db_links(
        &self,
        _record_name: &str,
        rec: &Arc<crate::runtime::sync::RwLock<RecordInstance>>,
        actions: &[crate::server::record::ProcessAction],
        visited: &mut HashSet<String>,
        depth: usize,
    ) {
        use crate::server::record::ProcessAction;
        for action in actions {
            if let ProcessAction::ReadDbLink {
                link_field,
                target_field,
            } = action
            {
                let link_str = {
                    let instance = rec.read().await;
                    instance
                        .record
                        .get_field(link_field)
                        .and_then(|v| {
                            if let EpicsValue::String(s) = v {
                                Some(s)
                            } else {
                                None
                            }
                        })
                        .unwrap_or_default()
                };
                if link_str.is_empty() {
                    continue;
                }
                let parsed = crate::server::record::parse_link_v2(&link_str);
                if let Some(value) = self.read_link_value(&parsed, visited, depth).await {
                    let mut instance = rec.write().await;
                    let _ = instance.record.put_field_internal(target_field, value);
                }
            }
        }
    }

    /// Execute ProcessActions returned by a record's process() call.
    ///
    /// Actions are executed in order:
    /// - ReadDbLink: reads a linked PV value and writes it into a record field
    ///   (bypasses read-only checks via put_field_internal)
    /// - WriteDbLink: writes a value to a linked PV
    /// - ReprocessAfter: schedules a delayed re-process via tokio::spawn
    async fn execute_process_actions(
        &self,
        record_name: &str,
        rec: &Arc<crate::runtime::sync::RwLock<RecordInstance>>,
        actions: Vec<crate::server::record::ProcessAction>,
        visited: &mut HashSet<String>,
        depth: usize,
    ) {
        use crate::server::record::ProcessAction;

        for action in actions {
            match action {
                ProcessAction::ReadDbLink {
                    link_field,
                    target_field,
                } => {
                    // 1. Get the link string from the record
                    let link_str = {
                        let instance = rec.read().await;
                        instance
                            .record
                            .get_field(link_field)
                            .and_then(|v| {
                                if let EpicsValue::String(s) = v {
                                    Some(s)
                                } else {
                                    None
                                }
                            })
                            .unwrap_or_default()
                    };
                    if link_str.is_empty() {
                        continue;
                    }
                    // 2. Parse and read the linked PV
                    let parsed = crate::server::record::parse_link_v2(&link_str);
                    if let Some(value) = self.read_link_value(&parsed, visited, depth).await {
                        // 3. Write into the record field (internal put bypasses read-only)
                        let mut instance = rec.write().await;
                        let _ = instance.record.put_field_internal(target_field, value);
                    }
                }
                ProcessAction::WriteDbLink { link_field, value } => {
                    // 1. Get the link string (record fields → common fields)
                    // and the source PUTF for processTarget propagation.
                    let (link_str, src_putf) = {
                        let instance = rec.read().await;
                        let link = instance
                            .resolve_field(link_field)
                            .and_then(|v| {
                                if let EpicsValue::String(s) = v {
                                    Some(s)
                                } else {
                                    None
                                }
                            })
                            .unwrap_or_default();
                        (link, instance.common.putf)
                    };
                    if link_str.is_empty() {
                        continue;
                    }
                    // 2. Parse and write to the linked PV — DB *or*
                    // external `ca://`/`pva://`. A record's `process()`
                    // emits `WriteDbLink` to drive an OUT-link field
                    // (transform `OUTn`, throttle/scaler `COUTP`, epid
                    // `TRIG`/`OUTL`); that field may resolve to a CA/PVA
                    // link, which C `dbPutLink` routes through the link
                    // set's `putValue` identically to a DB link
                    // (dbLink.c:434-448).
                    let parsed = crate::server::record::parse_link_v2(&link_str);
                    self.write_out_link_value(&parsed, value, src_putf, visited, depth)
                        .await;
                }
                ProcessAction::DeviceCommand { command, ref args } => {
                    let mut instance = rec.write().await;
                    if let Some(mut dev) = instance.device.take() {
                        // `handle_command` runs after the process snapshot
                        // was already built/notified, so any record field
                        // it mutated needs an explicit monitor post. The
                        // returned field names are posted with DBE_VALUE,
                        // mirroring the C record's `db_post_events` calls
                        // from inside `process()` (scalerRecord.c:425-430).
                        let changed = dev
                            .handle_command(&mut *instance.record, command, args)
                            .unwrap_or_default();
                        instance.device = Some(dev);
                        for field in changed {
                            instance.notify_field(field, crate::server::recgbl::EventMask::VALUE);
                        }
                    }
                }
                ProcessAction::ReprocessAfter(delay) => {
                    // Use generation counter for timer cancellation.
                    // Bump generation now; the spawned task only fires if
                    // the generation hasn't been bumped again (i.e., no newer
                    // timer replaced this one).
                    let (gen_counter, gen_val) = {
                        let instance = rec.read().await;
                        let val = instance
                            .reprocess_generation
                            .fetch_add(1, std::sync::atomic::Ordering::Relaxed)
                            + 1;
                        (instance.reprocess_generation.clone(), val)
                    };
                    let db = self.clone();
                    let rec_name = record_name.to_string();
                    tokio::spawn(async move {
                        tokio::time::sleep(delay).await;
                        // Only fire if no newer timer has been scheduled
                        let current = gen_counter.load(std::sync::atomic::Ordering::Relaxed);
                        if current == gen_val {
                            let mut visited = HashSet::new();
                            // Owner-driven continuation: bypass the PACT
                            // entry guard so the timer fire reaches the
                            // record's process() (which advances the
                            // async state machine — e.g. scaler DLY
                            // expiry, calc AFTC). Mirrors C
                            // `callbackRequestDelayed` dispatching to
                            // `(*prset->process)(prec)` directly.
                            let _ = db
                                .process_record_continuation(&rec_name, &mut visited, 0)
                                .await;
                        }
                    });
                }
            }
        }
    }

    /// Complete an asynchronous record's post-process steps.
    /// Call after device support signals completion (clears PACT, runs alarms, snapshot, OUT, FLNK).
    pub fn complete_async_record<'a>(
        &'a self,
        name: &'a str,
    ) -> std::pin::Pin<Box<dyn std::future::Future<Output = CaResult<()>> + Send + 'a>> {
        Box::pin(async move {
            let mut visited = HashSet::new();
            self.complete_async_record_inner(name, &mut visited, 0)
                .await
        })
    }

    async fn complete_async_record_inner(
        &self,
        name: &str,
        visited: &mut HashSet<String>,
        depth: usize,
    ) -> CaResult<()> {
        // Alias-aware entry — same pattern as
        // `process_record_with_links_inner`. `name` may arrive as an
        // alias from an async device-support callback that captured
        // the original record name; normalise to canonical so the
        // records-map lookup, the `visited` cycle set, and downstream
        // FLNK/OUT dispatches all see the same canonical name.
        let canonical_owned;
        let name: &str = if let Some(target) = self.resolve_alias(name).await {
            canonical_owned = target;
            &canonical_owned
        } else {
            name
        };

        let rec = {
            let records = self.inner.records.read().await;
            records
                .get(name)
                .cloned()
                .ok_or_else(|| CaError::ChannelNotFound(name.to_string()))?
        };

        // Seed the cycle guard with this record's own name — mirrors
        // the synchronous main path (`process_record_with_links_inner`
        // does `visited.insert(name)` before the body). Without this
        // the async-completion FLNK / OUT / CP dispatch can re-enter
        // the just-completed record: an async FLNK chain that loops
        // back (A async -> completes -> FLNK -> B -> FLNK -> A) would
        // re-process A unbounded, because PACT is cleared below before
        // the FLNK dispatch and nothing else blocks the re-entry.
        if !visited.insert(name.to_string()) {
            return Ok(()); // Cycle detected, skip
        }

        let (snapshot, out_info, flnk_name, alarm_posts) = {
            let mut instance = rec.write().await;

            // UDF update before alarm evaluation (C parity — see the
            // sync process path). A NaN/undefined value keeps UDF true
            // so `recGblCheckUDF` raises UDF_ALARM this cycle.
            if instance.record.clears_udf() {
                instance.common.udf = instance.record.value_is_undefined();
            }
            // Per-record alarm hook (C `checkAlarms()`).
            {
                let inst = &mut *instance;
                inst.record.check_alarms(&mut inst.common);
            }

            // Evaluate alarms
            instance.evaluate_alarms();

            let is_soft = instance.common.dtyp.is_empty() || instance.common.dtyp == "Soft Channel";

            // Device support alarm/timestamp override
            if !is_soft {
                let (dev_alarm, dev_ts) = if let Some(ref dev) = instance.device {
                    (dev.last_alarm(), dev.last_timestamp())
                } else {
                    (None, None)
                };
                if let Some((stat, sevr)) = dev_alarm {
                    crate::server::recgbl::rec_gbl_set_sevr(
                        &mut instance.common,
                        stat,
                        crate::server::record::AlarmSeverity::from_u16(sevr),
                    );
                }
                if let Some(ts) = dev_ts {
                    instance.common.time = ts;
                }
            }

            let alarm_result = crate::server::recgbl::rec_gbl_reset_alarms(&mut instance.common);

            apply_timestamp(&mut instance.common, is_soft);
            // UDF was already updated before `evaluate_alarms` above.

            // Clear PACT
            instance
                .processing
                .store(false, std::sync::atomic::Ordering::Release);

            // Fire put_notify completion (CA WRITE_NOTIFY response)
            if let Some(tx) = instance.put_notify_tx.take() {
                let _ = tx.send(());
            }

            use crate::server::recgbl::EventMask;
            let mut event_mask = EventMask::NONE;
            let (include_val, include_archive) = if instance.record.uses_monitor_deadband() {
                instance.check_deadband_ext()
            } else {
                // Binary records (bi/bo/busy/mbbi/mbbo): always post monitors
                (true, true)
            };
            if include_val {
                event_mask |= EventMask::VALUE;
            }
            if include_archive {
                event_mask |= EventMask::LOG;
            }
            if alarm_result.alarm_changed || alarm_result.amsg_changed {
                // C `recGbl.c:194/203` — same parity rule as the main
                // process path above (see comment there): amsg-only OR
                // sevr/stat change → DBE_ALARM.
                event_mask |= EventMask::ALARM;
            }

            let mut changed_fields = Vec::new();
            if include_val {
                if let Some(val) = instance.record.val() {
                    changed_fields.push(("VAL".to_string(), val));
                }
            }
            // C `recGblResetAlarms` (recGbl.c:201-220) posts each alarm
            // field with its OWN per-field mask, not the record-wide
            // `event_mask`. Mirror the synchronous link path
            // (`process_record_with_links_inner`) and `process_local`
            // exactly: SEVR=DBE_VALUE on a sevr change; STAT/AMSG share
            // `stat_mask` which carries DBE_ALARM when sevr OR amsg
            // moved and DBE_VALUE on a stat change; ACKS=DBE_VALUE only
            // when an alarm field moved AND recGblResetAlarms raised it.
            // Collapsing these into `changed_fields` would post them all
            // on one record-wide mask — losing C's per-field
            // granularity for `.SEVR`/`.STAT`-only subscribers.
            let sevr_changed = instance.common.sevr != alarm_result.prev_sevr;
            let stat_changed = instance.common.stat != alarm_result.prev_stat;
            let stat_mask = {
                let mut m = EventMask::NONE;
                if sevr_changed || alarm_result.amsg_changed {
                    m |= EventMask::ALARM;
                }
                if stat_changed {
                    m |= EventMask::VALUE;
                }
                m
            };
            let mut alarm_posts: Vec<(&'static str, EventMask)> = Vec::new();
            if sevr_changed {
                alarm_posts.push(("SEVR", EventMask::VALUE));
            }
            if !stat_mask.is_empty() {
                alarm_posts.push(("STAT", stat_mask));
                alarm_posts.push(("AMSG", stat_mask));
                // C `val_mask = DBE_ALARM` — the value field carries
                // DBE_ALARM whenever any alarm field moved.
                event_mask |= EventMask::ALARM;
            }
            // C parity (recGbl.c:216): ACKS is posted (DBE_VALUE) only
            // when an alarm field moved AND recGblResetAlarms raised it.
            if alarm_result.acks_changed && !stat_mask.is_empty() {
                alarm_posts.push(("ACKS", EventMask::VALUE));
            }
            if !event_mask.is_empty() {
                changed_fields.push((
                    "UDF".to_string(),
                    EpicsValue::Char(if instance.common.udf { 1 } else { 0 }),
                ));
            }
            // Add subscribed non-{VAL,SEVR,STAT,AMSG,UDF} fields that
            // actually changed since last notification — mirrors the
            // main-path snapshot gate (process_record_with_links_inner
            // L794-820). Without this, every async-completion cycle
            // re-sends every subscribed auxiliary field even when its
            // value is unchanged, multiplying the monitor traffic for
            // any record that pairs an async write with a sticky
            // metadata field.
            let mut sub_updates: Vec<(String, EpicsValue)> = Vec::new();
            for (field, subs) in &instance.subscribers {
                if !subs.is_empty()
                    && field != "VAL"
                    && field != "SEVR"
                    && field != "STAT"
                    && field != "AMSG"
                    && field != "UDF"
                {
                    if let Some(val) = instance.resolve_field(field) {
                        let changed = match instance.last_posted.get(field) {
                            Some(prev) => prev != &val,
                            None => true,
                        };
                        if changed {
                            sub_updates.push((field.clone(), val));
                        }
                    }
                }
            }
            if !sub_updates.is_empty() {
                for (field, val) in &sub_updates {
                    instance.last_posted.insert(field.clone(), val.clone());
                }
                changed_fields.extend(sub_updates);
                event_mask |= crate::server::recgbl::EventMask::VALUE;
            }
            let snapshot = crate::server::record::ProcessSnapshot {
                changed_fields,
                event_mask,
            };

            // IVOA check
            let skip_out = if instance.common.sevr == crate::server::record::AlarmSeverity::Invalid
            {
                let ivoa = instance
                    .record
                    .get_field("IVOA")
                    .and_then(|v| {
                        if let EpicsValue::Short(s) = v {
                            Some(s)
                        } else {
                            None
                        }
                    })
                    .unwrap_or(0);
                match ivoa {
                    1 => true,
                    2 => {
                        // See Round-30C comment in
                        // `process_record_with_links_inner` — IVOA=2
                        // delegates to the per-record
                        // `apply_invalid_output_value` so OVAL/RVAL/VAL
                        // get the C-convention values.
                        if let Some(ivov) = instance.record.get_field("IVOV") {
                            let _ = instance.record.apply_invalid_output_value(ivov);
                        }
                        false
                    }
                    _ => false,
                }
            } else {
                false
            };

            let can_dev_write = instance.record.can_device_write();
            let is_soft_out =
                instance.common.dtyp.is_empty() || instance.common.dtyp == "Soft Channel";
            let record_should_output = instance.record.should_output();
            let out_info = if skip_out {
                None
            } else if !can_dev_write {
                // Non-output records (calcout, etc.) with soft OUT link
                // (DB or external `ca://`/`pva://`).
                if record_should_output && instance.parsed_out.is_writable_out_link() {
                    let out_val = instance
                        .record
                        .get_field("OVAL")
                        .or_else(|| instance.record.val());
                    out_val.map(|v| (instance.parsed_out.clone(), v))
                } else {
                    None
                }
            } else if is_soft_out {
                if instance.parsed_out.is_writable_out_link() {
                    let out_val = instance
                        .record
                        .get_field("OVAL")
                        .or_else(|| instance.record.val());
                    out_val.map(|v| (instance.parsed_out.clone(), v))
                } else {
                    None
                }
            } else {
                // Non-soft output: the async device write already completed
                // (that's why we're in complete_async_record). Don't re-do
                // write_begin -- it would start another async cycle.
                None
            };

            let flnk_name = if instance.record.should_fire_forward_link() {
                if let crate::server::record::ParsedLink::Db(ref l) = instance.parsed_flnk {
                    Some(l.record.clone())
                } else {
                    None
                }
            } else {
                None
            };

            (snapshot, out_info, flnk_name, alarm_posts)
        };

        // Notify subscribers
        {
            let instance = rec.read().await;
            instance.notify_from_snapshot(&snapshot);
            // Post the alarm fields (SEVR/STAT/AMSG/ACKS) with their
            // individual C masks — see recGblResetAlarms above.
            for &(field, mask) in &alarm_posts {
                instance.notify_field(field, mask);
            }
        }

        // Snapshot source PUTF for processTarget propagation (see
        // `write_db_link_value` doc). For the async-completion path PUTF
        // would have been set when the put landed on the record; it must
        // propagate through the (now-completing) OUT / FLNK chain.
        let src_putf = rec.read().await.common.putf;

        // OUT link — DB *or* external `ca://`/`pva://`. Same scheme
        // dispatch as the sync path (C `dbLink.c::dbPutLink`,
        // dbLink.c:434-448).
        if let Some((link, out_val)) = out_info {
            self.write_out_link_value(&link, out_val, src_putf, visited, depth)
                .await;
        }

        // Multi-output dispatch (fanout/dfanout/seq/sseq)
        self.dispatch_multi_output(&rec, visited, depth).await;

        // event record: post the named software event.
        self.dispatch_event_record(&rec).await;

        // Generic multi-output links (transform OUTA..OUTP -> A..P,
        // scalcout OUT->OVAL, epid OUTL).
        //
        // SINGLE-OWNER INVARIANT: skip any record type owned by
        // `dispatch_multi_output` (called above) so its `LNKn`/`OUTn`
        // is not dispatched twice — see the sync-path twin in
        // `run_forward_link_tail_with_putf` §4.6.
        {
            let multi_out = {
                let instance = rec.read().await;
                let links =
                    if super::links::multi_output_dispatch_owned(instance.record.record_type()) {
                        &[][..]
                    } else {
                        instance.record.multi_output_links()
                    };
                if links.is_empty() {
                    None
                } else {
                    let mut pairs = Vec::new();
                    for &(link_field, val_field) in links {
                        let link_str = instance
                            .record
                            .get_field(link_field)
                            .and_then(|v| {
                                if let EpicsValue::String(s) = v {
                                    Some(s)
                                } else {
                                    None
                                }
                            })
                            .unwrap_or_default();
                        if link_str.is_empty() {
                            continue;
                        }
                        if let Some(val) = instance.record.get_field(val_field) {
                            pairs.push((link_str, val));
                        }
                    }
                    if pairs.is_empty() { None } else { Some(pairs) }
                }
            };
            if let Some(pairs) = multi_out {
                for (link_str, val) in pairs {
                    // `multi_output_links` carries record OUT links
                    // (sseq `LNKn`, scalcout `OUTn` — all `DBF_OUTLINK`):
                    // a bare DB link is NPP (`dbDbLink.c:388`).
                    // `parse_output_link_v2` applies the OUT-link-correct
                    // NPP default; an external `ca://`/`pva://` link is
                    // routed through the link set's `putValue` — see the
                    // sync-path twin above.
                    let parsed = crate::server::record::parse_output_link_v2(&link_str);
                    self.write_out_link_value(&parsed, val, src_putf, visited, depth)
                        .await;
                }
            }
        }

        // FLNK -- only process if target is Passive (C `dbScanFwdLink` ->
        // `dbScanPassive` -> `processTarget` propagates PUTF the same way
        // OUT links do).
        if let Some(ref flnk) = flnk_name {
            if let Some(target_rec) = self.get_record(flnk).await {
                let (target_scan, should_process) = {
                    let mut tg = target_rec.write().await;
                    let pact = tg.is_processing();
                    let on_chain = visited.contains(flnk);
                    if !pact {
                        tg.common.putf = src_putf;
                    } else if src_putf && !on_chain {
                        tg.common.rpro = true;
                        tg.common.putf = false;
                    }
                    (tg.common.scan, !pact)
                };
                if should_process && target_scan == crate::server::record::ScanType::Passive {
                    // MR-R5: recursive FLNK within one chain — gate
                    // already held by the foreign entry record.
                    let _ = self
                        .process_record_with_links_recursive(flnk, visited, depth + 1)
                        .await;
                }
            }
        }

        // CP link targets
        self.dispatch_cp_targets(name, visited, depth).await;

        // RPRO: C `recGblFwdLink` consumes a pending reprocess via
        // `scanOnce` — queued, not recursed. Mirror the synchronous
        // path: spawn a fresh process pass (clean `visited`, depth 0).
        {
            let needs_rpro = {
                let mut guard = rec.write().await;
                if guard.common.rpro {
                    guard.common.rpro = false;
                    true
                } else {
                    false
                }
            };
            if needs_rpro {
                let db = self.clone();
                let rpro_name = name.to_string();
                crate::runtime::task::spawn(async move {
                    let mut fresh_visited = std::collections::HashSet::new();
                    let _ = db
                        .process_record_with_links(&rpro_name, &mut fresh_visited, 0)
                        .await;
                });
            }
        }

        // C `recGbl.c::recGblFwdLink:302` clears `putf = FALSE` after
        // the forward-link dispatch. The same clearing must happen
        // at the tail of the async-completion path (this is the moral
        // equivalent of the synchronous completion path in
        // `put_record_field_from_ca` which clears after
        // `process_record_with_links` returns). Without this, a
        // record that completed an async write triggered by a
        // CA put would keep `putf=1` forever, leaking into every
        // subsequent scan-driven process cycle.
        {
            let mut guard = rec.write().await;
            guard.common.putf = false;
        }

        Ok(())
    }

    /// Dispatch CP-link targets that take a CP/CPP input link from `name`.
    ///
    /// C parity (a4bc0db): the CP-driven dispatch is the moral equivalent of
    /// dbCaTask's CA_DBPROCESS handler invoking `db_process(prec)`. Before
    /// processing each target, set PUTF=true; if the target is already
    /// processing (async record mid-flight), set RPRO=true instead so the
    /// in-flight pass reprocesses on completion. Already-visited targets
    /// (current process chain) are skipped via the `visited` cycle guard.
    async fn dispatch_cp_targets(
        &self,
        name: &str,
        visited: &mut std::collections::HashSet<String>,
        depth: usize,
    ) {
        let cp_targets = self.get_cp_targets(name).await;
        for target in cp_targets {
            if visited.contains(&target) {
                continue;
            }
            let target_rec = {
                let records = self.inner.records.read().await;
                records.get(&target).cloned()
            };
            let already_active = if let Some(ref t) = target_rec {
                let mut tg = t.write().await;
                if tg.processing.load(std::sync::atomic::Ordering::Acquire) {
                    tg.common.rpro = true;
                    true
                } else {
                    // epics-base PR #3fb10b6: PUTF must remain
                    // false on CP-driven targets — only the record
                    // directly receiving the dbPut should report
                    // PUTF=1 to dbNotify/onChange observers. The
                    // pre-fix C path (and this Rust port until now)
                    // wrongly propagated PUTF=true onto every CP
                    // target, so a downstream OPI reading PUTF on
                    // chained records saw the put attribution
                    // smeared across the entire chain.
                    false
                }
            } else {
                false
            };
            if already_active {
                continue;
            }
            // MR-R5: recursive CP-target fan-out within one chain —
            // gate already held by the foreign entry record.
            let _ = self
                .process_record_with_links_recursive(&target, visited, depth + 1)
                .await;
        }
    }

    /// Check simulation mode for a record. Returns
    /// `SimOutcome::Simulated` when simulation handled the value (the
    /// caller must still run the forward-link tail), or
    /// `SimOutcome::NotSimulated` when normal processing should proceed.
    async fn check_simulation_mode(&self, rec: &Arc<RwLock<RecordInstance>>) -> SimOutcome {
        // Read SIML, SIMM, SIOL, SIMS from the record
        let (siml_link, siol_link, sims, _rtype, is_input) = {
            let instance = rec.read().await;
            let rtype = instance.record.record_type().to_string();
            // Every input record whose DBD declares SIML/SIOL/SIMM/SIMS.
            // `mbbi`/`mbbiDirect` are input records: `mbbiRecord.c:125-126`
            // (and mbbiDirectRecord.c) declare SIML+SIOL, and
            // `mbbiRecord.c:388-394` reads `dbGetLink(&prec->siol,
            // DBR_ULONG, &prec->sval)` then `rval = sval` — input
            // semantics. Omitting them sent a simulated mbbi down the
            // OUTPUT branch, which writes VAL out to SIOL instead of
            // reading the value in from it.
            let is_input = matches!(
                rtype.as_str(),
                "ai" | "bi"
                    | "mbbi"
                    | "mbbiDirect"
                    | "longin"
                    | "int64in"
                    | "stringin"
                    | "lsi"
                    | "event"
            );

            let siml = instance
                .record
                .get_field("SIML")
                .and_then(|v| {
                    if let EpicsValue::String(s) = v {
                        Some(s)
                    } else {
                        None
                    }
                })
                .unwrap_or_default();
            let siol = instance
                .record
                .get_field("SIOL")
                .and_then(|v| {
                    if let EpicsValue::String(s) = v {
                        Some(s)
                    } else {
                        None
                    }
                })
                .unwrap_or_default();
            let sims = instance
                .record
                .get_field("SIMS")
                .and_then(|v| {
                    if let EpicsValue::Short(s) = v {
                        Some(s)
                    } else {
                        None
                    }
                })
                .unwrap_or(0);

            if siml.is_empty() && siol.is_empty() {
                return SimOutcome::NotSimulated; // No simulation configured
            }

            let siml_parsed = crate::server::record::parse_link_v2(&siml);
            let siol_parsed = crate::server::record::parse_link_v2(&siol);

            (siml_parsed, siol_parsed, sims, rtype, is_input)
        };

        // Read SIML -> update SIMM
        if let crate::server::record::ParsedLink::Db(ref link) = siml_link {
            let pv_name = if link.field == "VAL" {
                link.record.clone()
            } else {
                format!("{}.{}", link.record, link.field)
            };
            if let Ok(val) = self.get_pv(&pv_name).await {
                let simm_val = val.to_f64().unwrap_or(0.0) as i16;
                let mut instance = rec.write().await;
                let _ = instance
                    .record
                    .put_field("SIMM", EpicsValue::Short(simm_val));
            }
        }

        // Check SIMM
        let simm = {
            let instance = rec.read().await;
            instance
                .record
                .get_field("SIMM")
                .and_then(|v| {
                    if let EpicsValue::Short(s) = v {
                        Some(s)
                    } else {
                        None
                    }
                })
                .unwrap_or(0)
        };

        if simm == 0 {
            return SimOutcome::NotSimulated; // NO simulation, proceed normally
        }

        // epics-base 7.0.7 (SIMM menu):
        //   1 = YES — read/write via SIOL using the cooked VAL
        //   2 = RAW — read/write via SIOL using the raw RVAL when the
        //             record carries one (ai/ao only); falls back to
        //             VAL when no RVAL is present. Mirrors the C
        //             implementation, which treats records lacking
        //             a raw value as "YES" since there's nothing
        //             else to copy.
        let raw_mode = simm == 2;
        let raw_field = if raw_mode { "RVAL" } else { "VAL" };

        // SIMM=YES(1) / SIMM=RAW(2): handle simulation
        if let crate::server::record::ParsedLink::Db(ref link) = siol_link {
            let pv_name = if link.field == "VAL" {
                link.record.clone()
            } else {
                format!("{}.{}", link.record, link.field)
            };

            if is_input {
                // Input record: read from SIOL -> set VAL/RVAL.
                if let Ok(siol_val) = self.get_pv(&pv_name).await {
                    let mut instance = rec.write().await;
                    let target_supports_raw =
                        raw_mode && instance.record.get_field("RVAL").is_some();
                    if target_supports_raw {
                        // PR #ac92e3e follow-up: SIMM=RAW on records
                        // with RVAL (ai/ao/etc.) writes the raw value
                        // into RVAL and runs the record's own
                        // process() so the LINR / ESLO / EOFF / ASLO
                        // / AOFF conversion chain computes VAL. The
                        // pre-fix path additionally called set_val
                        // here, which overwrote VAL with the raw
                        // count and silently bypassed conversion —
                        // the visible failure mode was "SIMM=RAW
                        // simulation returns counts instead of EGU".
                        //
                        // Coerce to RVAL's native DBR type before
                        // put_field — ai.RVAL is Long, but SIOL on a
                        // soft channel typically yields Double. Without
                        // the coerce step the put_field rejects with
                        // TypeMismatch and leaves RVAL at 0, so
                        // process() computes VAL = 0*ESLO + EOFF
                        // (the offset only), not the intended
                        // RAW*ESLO + EOFF.
                        let rval_type = instance
                            .record
                            .field_list()
                            .iter()
                            .find(|f| f.name == "RVAL")
                            .map(|f| f.dbf_type)
                            .unwrap_or(crate::types::DbFieldType::Long);
                        // C parity (aiRecord.c:495): `rval = (long)floor(sval)`.
                        // Rust `convert_to(Long)` truncates toward zero,
                        // diverging for negative bipolar-ADC raw values
                        // (sval=-1.5 → C: -2, Rust as-cast: -1).
                        // Floor explicitly when narrowing a float to
                        // an integer RVAL.
                        let coerced = match (&siol_val, rval_type) {
                            (EpicsValue::Double(d), crate::types::DbFieldType::Long) => {
                                EpicsValue::Long(d.floor() as i32)
                            }
                            (EpicsValue::Double(d), crate::types::DbFieldType::Int64) => {
                                EpicsValue::Int64(d.floor() as i64)
                            }
                            (EpicsValue::Float(d), crate::types::DbFieldType::Long) => {
                                EpicsValue::Long((*d as f64).floor() as i32)
                            }
                            (EpicsValue::Float(d), crate::types::DbFieldType::Int64) => {
                                EpicsValue::Int64((*d as f64).floor() as i64)
                            }
                            _ if siol_val.db_field_type() != rval_type => {
                                siol_val.convert_to(rval_type)
                            }
                            _ => siol_val,
                        };
                        let _ = instance.record.put_field("RVAL", coerced);
                        let ctx = instance.common.process_context();
                        instance.record.set_process_context(&ctx);
                        let _ = instance.record.process();
                    } else {
                        // Records without RVAL fall back to SIMM=YES
                        // semantics: the SIOL value goes straight into
                        // VAL; no conversion to run.
                        let _ = instance.record.set_val(siol_val);
                    }
                    apply_timestamp(&mut instance.common, true);
                    instance.common.udf = false;

                    // Set simulation alarm
                    let sev = crate::server::record::AlarmSeverity::from_u16(sims as u16);
                    if sev != crate::server::record::AlarmSeverity::NoAlarm {
                        instance.common.sevr = sev;
                        instance.common.stat = crate::server::recgbl::alarm_status::SIMM_ALARM;
                    }

                    // Build snapshot and notify
                    let mut changed_fields = Vec::new();
                    if let Some(val) = instance.record.val() {
                        changed_fields.push(("VAL".to_string(), val));
                    }
                    changed_fields.push((
                        "SEVR".to_string(),
                        EpicsValue::Short(instance.common.sevr as i16),
                    ));
                    changed_fields.push((
                        "STAT".to_string(),
                        EpicsValue::Short(instance.common.stat as i16),
                    ));
                    let snapshot = crate::server::record::ProcessSnapshot {
                        changed_fields,
                        event_mask: crate::server::recgbl::EventMask::VALUE
                            | crate::server::recgbl::EventMask::ALARM,
                    };
                    instance.notify_from_snapshot(&snapshot);
                }
            } else {
                // Output record: write VAL (or RVAL for SIMM=RAW) to
                // SIOL (skip device write).
                let out_val = {
                    let instance = rec.read().await;
                    if raw_mode {
                        // RAW path: prefer RVAL when the record has
                        // one. Otherwise fall through to VAL.
                        instance
                            .record
                            .get_field(raw_field)
                            .or_else(|| instance.record.val())
                    } else {
                        instance.record.val()
                    }
                };
                if let Some(val) = out_val {
                    // MR-R5: writing VAL to the SIOL target is an
                    // internal step of this record's processing chain,
                    // which already holds the entry record's advisory
                    // write gate. Use the `_already_locked` write so a
                    // SIOL that points back at a chain record cannot
                    // dead-lock on a non-reentrant gate (same reasoning
                    // as the OUT-link write in `write_db_link_value`).
                    let _ = self.put_pv_already_locked(&pv_name, val).await;
                }

                let mut instance = rec.write().await;
                apply_timestamp(&mut instance.common, true);
                instance.common.udf = false;

                let sev = crate::server::record::AlarmSeverity::from_u16(sims as u16);
                if sev != crate::server::record::AlarmSeverity::NoAlarm {
                    instance.common.sevr = sev;
                    instance.common.stat = crate::server::recgbl::alarm_status::SIMM_ALARM;
                }

                // Notify subscribers of simulation output
                let mut changed_fields = Vec::new();
                if let Some(val) = instance.record.val() {
                    changed_fields.push(("VAL".to_string(), val));
                }
                changed_fields.push((
                    "SEVR".to_string(),
                    EpicsValue::Short(instance.common.sevr as i16),
                ));
                changed_fields.push((
                    "STAT".to_string(),
                    EpicsValue::Short(instance.common.stat as i16),
                ));
                let snapshot = crate::server::record::ProcessSnapshot {
                    changed_fields,
                    event_mask: crate::server::recgbl::EventMask::VALUE
                        | crate::server::recgbl::EventMask::ALARM,
                };
                instance.notify_from_snapshot(&snapshot);
            }
        }

        SimOutcome::Simulated
    }
}