bot-forge 1.0.2

//! Ordered lifecycle facts with process-wide session routing and JSONL encoding.
//!
//! Execution, backends, and infrastructure publish facts through this neutral output capability.
//! Final report persistence remains in [`crate::reporting`], while this module owns run routing,
//! redaction, sequence assignment, and stream serialization.

use std::io::{self, Write};
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Mutex, OnceLock, Weak};
use std::time::{SystemTime, UNIX_EPOCH};

use serde::Serialize;

use crate::redaction::mask_secrets;

#[derive(Debug, Clone, Serialize)]
/// Metadata and payload for one ordered lifecycle event.
pub(crate) struct EventEnvelope {
    pub(crate) run_id: String,
    pub(crate) sequence: u64,
    pub(crate) timestamp_ms: u128,
    pub(crate) node: Option<String>,
    pub(crate) component: Option<String>,
    pub(crate) backend: Option<String>,
    #[serde(flatten)]
    pub(crate) payload: LifecycleEvent,
}

#[derive(Debug, Clone, Serialize)]
#[serde(tag = "event", rename_all = "kebab-case")]
/// Stable machine-readable events emitted across a run and its nodes.
pub(crate) enum LifecycleEvent {
    Planned,
    Queued,
    ResourceWait {
        resource: String,
    },
    ResourceAcquired {
        resource: String,
        wait_ms: u128,
    },
    RetryScheduled {
        attempt: u32,
        delay_ms: u64,
    },
    Progress {
        elapsed_ms: u128,
        message: String,
    },
    Completed {
        elapsed_ms: u128,
    },
    Failed {
        code: String,
        message: String,
    },
    Blocked {
        dependency: String,
    },
    Skipped {
        reason: String,
    },
    Cancelled {
        forced: bool,
    },
    RunCompleted {
        outcome: String,
        duration_ms: u128,
        log_path: String,
    },
}

/// Creates monotonically sequenced envelopes for one run.
pub(crate) struct EventFactory {
    run_id: String,
    sequence: AtomicU64,
}

struct ActiveEventStream {
    factory: EventFactory,
}

static ACTIVE_EVENT_STREAM: OnceLock<Mutex<Weak<ActiveEventStream>>> = OnceLock::new();

/// RAII scope that routes global event emission to one active run.
///
/// The process supports one active lifecycle session at a time. Starting an overlapping session
/// replaces the global route; callers that need concurrent independent runs must provide explicit
/// run-scoped routing instead of sharing this coordinator.
pub(crate) struct LifecycleSession {
    _stream: Arc<ActiveEventStream>,
}

impl LifecycleSession {
    /// Start an event session and replace the process-wide active stream.
    pub(crate) fn begin(run_id: impl Into<String>) -> Self {
        let stream = Arc::new(ActiveEventStream {
            factory: EventFactory::new(run_id),
        });
        let active = ACTIVE_EVENT_STREAM.get_or_init(|| Mutex::new(Weak::new()));
        if let Ok(mut current) = active.lock() {
            *current = Arc::downgrade(&stream);
        }
        Self { _stream: stream }
    }

    /// Emit an event through this session, attaching optional execution context.
    ///
    /// Payload text is redacted before serialization. Output failures are intentionally not
    /// surfaced because lifecycle reporting must not replace the operation's primary result.
    pub(crate) fn emit(
        &self,
        node: Option<&str>,
        component: Option<&str>,
        backend: Option<&str>,
        event: LifecycleEvent,
    ) {
        write_event(&self._stream, node, component, backend, event);
    }
}

impl Drop for LifecycleSession {
    fn drop(&mut self) {
        if let Some(active) = ACTIVE_EVENT_STREAM.get()
            && let Ok(mut current) = active.lock()
        {
            *current = Weak::new();
        }
    }
}

/// Emit through the active lifecycle session, or do nothing when no session is active.
pub(crate) fn emit(
    node: Option<&str>,
    component: Option<&str>,
    backend: Option<&str>,
    event: LifecycleEvent,
) {
    let stream = ACTIVE_EVENT_STREAM
        .get()
        .and_then(|active| active.lock().ok())
        .and_then(|active| active.upgrade());
    if let Some(stream) = stream {
        write_event(&stream, node, component, backend, event);
    }
}

/// Return whether a live lifecycle session currently receives global events.
pub(crate) fn is_active() -> bool {
    ACTIVE_EVENT_STREAM
        .get()
        .and_then(|active| active.lock().ok())
        .and_then(|active| active.upgrade())
        .is_some()
}

fn write_event(
    stream: &ActiveEventStream,
    node: Option<&str>,
    component: Option<&str>,
    backend: Option<&str>,
    event: LifecycleEvent,
) {
    let envelope = stream.factory.envelope(node, component, backend, event);
    let _ = write_jsonl_event(io::stdout().lock(), &envelope);
}

/// Serialize one lifecycle envelope as one newline-terminated JSON object.
///
/// Broken pipes are treated as successful early consumer termination so a producer can exit
/// cleanly when a downstream JSONL reader stops consuming.
pub(crate) fn write_jsonl_event(mut writer: impl Write, event: &EventEnvelope) -> io::Result<()> {
    let mut text = serde_json::to_string(event)
        .map_err(|error| io::Error::new(io::ErrorKind::InvalidData, error))?;
    text.push('\n');
    match writer.write_all(text.as_bytes()) {
        Ok(()) => writer.flush(),
        Err(error) if error.kind() == io::ErrorKind::BrokenPipe => Ok(()),
        Err(error) => Err(error),
    }
}

impl EventFactory {
    /// Create a factory whose first envelope has sequence number one.
    pub(crate) fn new(run_id: impl Into<String>) -> Self {
        Self {
            run_id: run_id.into(),
            sequence: AtomicU64::new(0),
        }
    }

    /// Build the next envelope and redact its payload before returning it.
    pub(crate) fn envelope(
        &self,
        node: Option<&str>,
        component: Option<&str>,
        backend: Option<&str>,
        payload: LifecycleEvent,
    ) -> EventEnvelope {
        EventEnvelope {
            run_id: self.run_id.clone(),
            sequence: self.sequence.fetch_add(1, Ordering::SeqCst) + 1,
            timestamp_ms: SystemTime::now()
                .duration_since(UNIX_EPOCH)
                .unwrap_or_default()
                .as_millis(),
            node: node.map(str::to_string),
            component: component.map(str::to_string),
            backend: backend.map(str::to_string),
            payload: redact_event(payload),
        }
    }
}

/// Redact secret-like text carried by a lifecycle event.
pub(crate) fn redact_event(event: LifecycleEvent) -> LifecycleEvent {
    match event {
        LifecycleEvent::Progress {
            elapsed_ms,
            message,
        } => LifecycleEvent::Progress {
            elapsed_ms,
            message: mask_secrets(&message),
        },
        LifecycleEvent::ResourceWait { resource } => LifecycleEvent::ResourceWait {
            resource: mask_secrets(&resource),
        },
        LifecycleEvent::ResourceAcquired { resource, wait_ms } => {
            LifecycleEvent::ResourceAcquired {
                resource: mask_secrets(&resource),
                wait_ms,
            }
        }
        LifecycleEvent::Failed { code, message } => LifecycleEvent::Failed {
            code,
            message: mask_secrets(&message),
        },
        LifecycleEvent::Skipped { reason } => LifecycleEvent::Skipped {
            reason: mask_secrets(&reason),
        },
        other => other,
    }
}

#[cfg(test)]
mod tests {
    use crate::events::{EventFactory, LifecycleEvent};

    #[test]
    fn envelopes_have_monotonic_sequences_and_redact_secrets() {
        let factory = EventFactory::new("run");
        let first = factory.envelope(
            None,
            Some("demo"),
            None,
            LifecycleEvent::Failed {
                code: "probe".into(),
                message: "token=secret".into(),
            },
        );
        let second = factory.envelope(None, None, None, LifecycleEvent::Planned);
        assert_eq!(first.sequence, 1);
        assert_eq!(second.sequence, 2);
        let encoded = serde_json::to_string(&first).unwrap();
        assert!(encoded.contains("token=***"));
        assert!(!encoded.contains("schema_version"));
    }

    #[test]
    fn resource_acquired_keeps_reason_and_duration_together() {
        let factory = EventFactory::new("run");
        let event = factory.envelope(
            Some("node"),
            Some("demo"),
            None,
            LifecycleEvent::ResourceAcquired {
                resource: "registry lock".into(),
                wait_ms: 42,
            },
        );
        let value = serde_json::to_value(event).unwrap();
        assert_eq!(value["event"], "resource-acquired");
        assert_eq!(value["resource"], "registry lock");
        assert_eq!(value["wait_ms"], 42);
    }
}