Offline-first autonomy platform for weak-network edge devices
curl -fsSL https://cli.moonbitlang.com/install/unix.sh | bash -s latest
export PATH="$HOME/.moon/bin:$PATH"
git clone https://github.com/zyyaa125/EdgeWeaver.git
cd EdgeWeaver
moon version --allmoon fmt --check
moon check --deny-warn
moon test --deny-warn
moon run cmd/mainmoon run cmd/recovery -- write recovery-demo.json
moon run cmd/recovery -- replay recovery-demo.jsonpub(all) suberror ControlError {
EmptyNodeId
InvalidTimestamp
UnsupportedProtocol(ProtocolVersion)
DuplicateNode(NodeId)
UnknownNode(NodeId)
NodeNotEnrolled(NodeId)
RetiredNode(NodeId)
StaleHeartbeat
StaleConfig(ConfigVersion)
DuplicateCommand(CommandId)
UnknownCommand(CommandId)
CommandTargetMismatch(CommandId)
CommandTransition(CommandError)
} derive(Eq, Debug)pub(all) suberror EdgeError {
EmptyAllowlistedAction
InvalidActionName
InvalidExecutionCheckpoint
PersistenceFailed
OutputTooLarge
InvalidTimestamp
NodeNotReady(NodeLifecycle)
RetiredNode(NodeId)
CommandTargetMismatch(CommandId)
CommandConflict(CommandId)
ActionNotAllowed(CommandAction)
MissingCapability(NodeCapability)
ExpiredCommand(CommandId)
CommandNotExecutable(CommandState)
ExecutionInProgress(CommandId)
EmptyResultMessage
CommandTransition(CommandError)
} derive(Eq, Debug)pub(all) suberror PolicyError {
EmptyRuleId
EmptyAction
InvalidWindow
InvalidMaxTriggers
InvalidOfflineDuration
InvalidTimestamp
InvalidCheckpoint
UnsupportedPolicyVersion
PersistenceFailed
DuplicateRuleId(String)
ActionNotAllowed(CommandAction)
} derive(Eq, Debug)pub(all) suberror ProtocolError {
InvalidVersion
UnsupportedVersion(ProtocolVersion)
EmptyMessageId
EmptyCorrelationId
EmptyPayload
} derive(Eq, Debug)pub(all) suberror TransportError {
InvalidBaseDelay
InvalidMaxDelay
InvalidAttemptLimit
InvalidTimestamp
TimestampOverflow
InvalidState(ConnectionState)
RetryNotReady(Timestamp)
RetryExhausted
UnsupportedProtocol(ProtocolVersion)
NoSharedCapability
SyncNotReady
Store(StoreError)
} derive(Eq, Debug)fn ActuatorAdapter::execute(self : ActuatorAdapter, action~ : String, at~ : Timestamp) -> ActuatorResult raise AdapterErrorfn ActuatorAdapter::new(id~ : String, allowed_actions~ : Array[String], initial_state~ : String) -> ActuatorAdapter raise AdapterErrorpub(all) struct AutonomyContext {
lifecycle : NodeLifecycle
health : NodeHealth
offline_for : Int
} derive(Eq, Debug)fn AutonomyContext::new(lifecycle~ : NodeLifecycle, health~ : NodeHealth, offline_for~ : Int) -> AutonomyContext raise PolicyErrorpub(all) struct AutonomyEngine {
allowlist : CommandAllowlist
rules : Array[AutonomyRule]
} derive(Eq, Debug)fn AutonomyEngine::evaluate(self : AutonomyEngine, context~ : AutonomyContext, at~ : Timestamp) -> PolicyDecision raise PolicyErrorfn AutonomyEngine::evaluate_durable(self : AutonomyEngine, context~ : AutonomyContext, at~ : Timestamp, persist~ : (Array[AutonomyRule]) -> Bool) -> PolicyDecision raise PolicyErrorfn AutonomyEngine::new(allowlist~ : CommandAllowlist, rules~ : Array[AutonomyRule]) -> AutonomyEngine raise PolicyErrorfn AutonomyEngine::restore(allowlist~ : CommandAllowlist, rules~ : Array[AutonomyRule]) -> AutonomyEngine raise PolicyErrorpub(all) struct AutonomyRule {
id : String
condition : PolicyCondition
action : CommandAction
valid_from : Timestamp
valid_until : Timestamp
max_triggers : Int
fired_count : Int
policy_version : Int
last_evaluated_at : Timestamp?
} derive(Eq, ToJson, Debug, FromJson)fn AutonomyRule::new(id~ : String, condition~ : PolicyCondition, action~ : CommandAction, valid_from~ : Timestamp, valid_until~ : Timestamp, max_triggers~ : Int) -> AutonomyRule raise PolicyErrorfn BackoffPolicy::new(base_delay~ : Int, max_delay~ : Int, max_attempts~ : Int) -> BackoffPolicy raise TransportErrorpub(all) struct CapabilityAdvertisement {
protocol : ProtocolVersion
capabilities : Array[TransportCapability]
} derive(Eq, Debug)fn CapabilityAdvertisement::new(protocol~ : ProtocolVersion, capabilities~ : Array[TransportCapability]) -> CapabilityAdvertisementfn CapabilityAdvertisement::not_equal(x : CapabilityAdvertisement, y : CapabilityAdvertisement) -> Boolpub(all) struct Command {
id : CommandId
node_id : NodeId
action : CommandAction
issued_at : Timestamp
expires_at : Timestamp
updated_at : Timestamp
state : CommandState
result : CommandResult?
} derive(Eq, ToJson, Debug, FromJson)fn Command::new(id~ : CommandId, node_id~ : NodeId, action~ : CommandAction, issued_at~ : Timestamp, expires_at~ : Timestamp) -> Command raise CommandErrorfn Command::succeed(self : Command, at~ : Timestamp, message~ : String) -> Command raise CommandErrorpub(all) enum CommandAction {
ApplyConfig(ConfigVersion)
ReadTelemetry
Execute(String)
} derive(Eq, ToJson, Debug, FromJson)fn CommandResult::new(state~ : CommandState, message~ : String, completed_at~ : Timestamp) -> CommandResult raise CommandErrorpub(all) struct ConfigDelivery {
node_id : NodeId
snapshot : ConfigSnapshot
delivered_at : Timestamp
} derive(Eq, Debug)fn ConfigSnapshot::new(version~ : ConfigVersion, content~ : String, created_at~ : Timestamp) -> ConfigSnapshot raise ConfigErrorfn ControlPlane::deliver_command(self : ControlPlane, node_id~ : NodeId, command~ : Command) -> Command raise ControlErrorfn ControlPlane::deliver_config(self : ControlPlane, node_id~ : NodeId, snapshot~ : ConfigSnapshot, at~ : Timestamp) -> ConfigDelivery raise ControlErrorfn ControlPlane::enroll(self : ControlPlane, request~ : EnrollmentRequest) -> Node raise ControlErrorfn ControlPlane::mark_offline(self : ControlPlane, node_id~ : NodeId, at~ : Timestamp) -> Node raise ControlErrorfn ControlPlane::receive_heartbeat(self : ControlPlane, heartbeat~ : Heartbeat) -> Node raise ControlErrorfn ControlPlane::update_command(self : ControlPlane, command_id~ : CommandId, update~ : CommandUpdate, at~ : Timestamp) -> Command raise ControlErrorfn DeviceObservation::new(node_id~ : NodeId, device_id~ : String, health~ : NodeHealth, metric~ : String, value~ : Int, unit~ : String, captured_at~ : Timestamp) -> DeviceObservation raise AdapterErrorfn DeviceObservation::to_envelope(self : DeviceObservation, message_id~ : String) -> MessageEnvelope raise ProtocolErrorpub(all) struct DeviceSimulator {
node_id : NodeId
sensors : Array[SensorAdapter]
actuators : Array[ActuatorAdapter]
} derive(Eq, Debug)fn DeviceSimulator::collect_sensor(self : DeviceSimulator, sensor_id~ : String, at~ : Timestamp) -> DeviceObservation raise AdapterErrorfn DeviceSimulator::execute_actuator(self : DeviceSimulator, actuator_id~ : String, action~ : String, at~ : Timestamp) -> ActuatorResult raise AdapterErrorfn DeviceSimulator::new(node_id~ : NodeId, sensors~ : Array[SensorAdapter], actuators~ : Array[ActuatorAdapter]) -> DeviceSimulator raise AdapterErrorfn DeviceSimulator::read_all(self : DeviceSimulator, at~ : Timestamp) -> Array[SensorReading] raise AdapterErrorfn DeviceSimulator::read_sensor(self : DeviceSimulator, sensor_id~ : String, at~ : Timestamp) -> SensorReading raise AdapterErrorpub(all) struct DeviceSnapshot {
node_id : NodeId
sensors : Array[SensorAdapter]
actuators : Array[ActuatorAdapter]
} derive(Eq, Debug)pub(all) struct DiagnosticsCollector {
max_logs : Int
logs : Array[DiagnosticEvent]
sync_failures : Int
last_sync_error : String?
last_policy_decision : PolicyDecision
} derive(Eq, Debug)fn DiagnosticsCollector::record_event(self : DiagnosticsCollector, at~ : Timestamp, level~ : LogLevel, component~ : String, message~ : String) -> Unit raise DiagnosticsErrorfn DiagnosticsCollector::record_policy_decision(self : DiagnosticsCollector, at~ : Timestamp, decision~ : PolicyDecision) -> Unit raise DiagnosticsErrorfn DiagnosticsCollector::record_sync_failure(self : DiagnosticsCollector, at~ : Timestamp, message~ : String) -> Unit raise DiagnosticsErrorfn DiagnosticsCollector::snapshot(self : DiagnosticsCollector, node~ : Node, store~ : LocalStore, transport~ : TransportManager, executor~ : EdgeExecutor, at~ : Timestamp) -> LocalDiagnostics raise DiagnosticsErrorpub(all) struct EdgeExecutor {
node : Node
allowlist : CommandAllowlist
records : Array[ExecutionRecord]
} derive(Eq, Debug)fn EdgeExecutor::execute(self : EdgeExecutor, command~ : Command, at~ : Timestamp, runner~ : (CommandAction) -> ExecutionResult) -> ExecutionRecord raise EdgeErrorfn EdgeExecutor::execute_durable(self : EdgeExecutor, command~ : Command, at~ : Timestamp, persist~ : (ExecutionCheckpoint) -> Bool, runner~ : (CommandAction) -> ExecutionResult) -> ExecutionRecord raise EdgeErrorfn EdgeExecutor::restore(node~ : Node, allowlist~ : CommandAllowlist, checkpoint~ : ExecutionCheckpoint) -> EdgeExecutorfn EdgeExecutor::restore_checked(node~ : Node, allowlist~ : CommandAllowlist, checkpoint~ : ExecutionCheckpoint) -> EdgeExecutor raise EdgeErrorpub(all) struct EnrollmentRequest {
version : ProtocolVersion
node_id : String
capabilities : Array[NodeCapability]
at : Timestamp
} derive(Eq, Debug)fn EnrollmentRequest::new(version~ : ProtocolVersion, node_id~ : String, capabilities~ : Array[NodeCapability], at~ : Timestamp) -> EnrollmentRequest raise ControlErrorpub(all) struct Heartbeat {
version : ProtocolVersion
node_id : String
health : NodeHealth
at : Timestamp
} derive(Eq, Debug)fn Heartbeat::new(version~ : ProtocolVersion, node_id~ : String, health~ : NodeHealth, at~ : Timestamp) -> Heartbeat raise ControlErrorpub(all) struct JournalConfig {
capacity : Int
full_policy : QueueFullPolicy
} derive(Eq, ToJson, Debug, FromJson)fn JournalConfig::new(capacity~ : Int, full_policy~ : QueueFullPolicy) -> JournalConfig raise StoreErrorpub(all) struct JournalEntry {
sequence : Int
envelope : MessageEnvelope
status : JournalStatus
attempts : Int
} derive(Eq, ToJson, Debug, FromJson)pub(all) struct LaunchOptions {
kind : RunProfileKind
data_dir : String?
config_path : String?
} derive(Eq, Debug)pub(all) struct LocalDiagnostics {
node_id : NodeId
lifecycle : NodeLifecycle
health : NodeHealth
captured_at : Timestamp
pending_events : Int
dropped_events : Int
sync_failures : Int
last_sync_error : String?
sync_cursor : SyncCursor
connection_state : ConnectionState
last_connected_at : Timestamp?
last_error_category : String?
negotiated_capabilities : Array[TransportCapability]
next_retry_at : Timestamp?
policy_decision : PolicyDecision
execution_records : Int
uncertain_executions : Int
failed_executions : Int
succeeded_executions : Int
events : Array[DiagnosticEvent]
} derive(Eq, Debug)pub(all) struct LocalStore {
config : JournalConfig
state : LocalStateSnapshot?
journal : Array[JournalEntry]
next_sequence : Int
dropped_count : Int
}pub(all) struct MessageEnvelope {
version : ProtocolVersion
message_id : String
node_id : NodeId
created_at : Timestamp
correlation_id : String?
payload : MessagePayload
} derive(Eq, ToJson, Debug, FromJson)fn MessageEnvelope::new(version~ : ProtocolVersion, message_id~ : String, node_id~ : NodeId, created_at~ : Timestamp, correlation_id~ : String?, payload~ : MessagePayload) -> MessageEnvelope raise ProtocolErrorpub(all) struct NegotiatedLink {
protocol : ProtocolVersion
capabilities : Array[TransportCapability]
} derive(Eq, Debug)pub(all) struct Node {
id : NodeId
capabilities : Array[NodeCapability]
lifecycle : NodeLifecycle
health : NodeHealth
created_at : Timestamp
updated_at : Timestamp
} derive(Eq, ToJson, Debug, FromJson)pub(all) enum PolicyCondition {
OfflineFor(Int)
HealthIs(NodeHealth)
} derive(Eq, ToJson, Debug, FromJson)pub(all) struct ReplayItem {
sequence : Int
message_id : String
outcome : ReplayOutcome
attempts : Int
} derive(Eq, Debug)pub(all) struct ReplayReport {
cursor : SyncCursor
inspected : Int
acknowledged : Int
deduplicated : Int
retryable_failures : Int
permanent_failures : Int
exhausted : Int
items : Array[ReplayItem]
} derive(Eq, Debug)fn ResourceLimits::new(max_pending_events~ : Int, max_log_entries~ : Int, max_reconnect_attempts~ : Int) -> ResourceLimits raise DeployErrorpub(all) struct RunProfile {
name : String
kind : RunProfileKind
target : DeploymentTarget
network : NetworkMode
data_dir : String
config_path : String
journal_dir : String
diagnostics_dir : String
resources : ResourceLimits
} derive(Eq, Debug)fn SensorAdapter::new(id~ : String, unit~ : String, initial_value~ : Int, step~ : Int) -> SensorAdapter raise AdapterErrorpub(all) struct StoreCheckpoint {
config : JournalConfig
state : LocalStateSnapshot?
journal : Array[JournalEntry]
next_sequence : Int
dropped_count : Int
} derive(Eq, ToJson, Debug, FromJson)pub(all) struct TransportAdapter {
connect : () -> TransportIoResult
close : () -> Unit
exchange : (CapabilityAdvertisement) -> TransportHandshakeResult
send_batch : (Array[MessageEnvelope]) -> TransportBatchResult
}pub(all) enum TransportBatchResult {
Receipts(Array[TransportReceipt])
Timeout
Disconnected
Failed(String)
}pub(all) enum TransportHandshakeResult {
Peer(CapabilityAdvertisement)
Timeout
Disconnected
Failed(String)
}pub(all) enum TransportIoResult {
Ready
Timeout
Disconnected
Failed(String)
}pub(all) struct TransportManager {
protocol : ProtocolVersion
capabilities : Array[TransportCapability]
backoff : BackoffPolicy
state : ConnectionState
attempts : Int
next_retry_at : Timestamp?
last_connect_attempt_at : Timestamp?
last_connected_at : Timestamp?
negotiated : NegotiatedLink?
cursor : SyncCursor
} derive(Eq, Debug)fn TransportManager::begin_connect(self : TransportManager, at~ : Timestamp) -> Unit raise TransportErrorfn TransportManager::close_with(self : TransportManager, adapter~ : TransportAdapter, at~ : Timestamp) -> Unit raise TransportErrorfn TransportManager::connect_with(self : TransportManager, adapter~ : TransportAdapter, at~ : Timestamp) -> NegotiatedLink raise TransportErrorfn TransportManager::disconnect(self : TransportManager, at~ : Timestamp) -> Unit raise TransportErrorfn TransportManager::handshake(self : TransportManager, peer~ : CapabilityAdvertisement) -> NegotiatedLink raise TransportErrorfn TransportManager::has_capability(self : TransportManager, capability~ : TransportCapability) -> Boolfn TransportManager::new(protocol~ : ProtocolVersion, capabilities~ : Array[TransportCapability], backoff~ : BackoffPolicy) -> TransportManager raise TransportErrorfn TransportManager::replay_pending(self : TransportManager, store : LocalStore, config~ : SyncConfig, sender~ : (MessageEnvelope) -> SyncSendResult) -> ReplayReport raise TransportErrorfn TransportManager::replay_with(self : TransportManager, store : LocalStore, config~ : SyncConfig, adapter~ : TransportAdapter) -> ReplayReport raise TransportErrorpub(all) struct WireFrame {
version : ProtocolVersion
message : WireMessage
} derive(ToJson, FromJson)pub(all) enum WireMessage {
Handshake(Array[TransportCapability])
Enroll(String, Array[NodeCapability], Timestamp)
Heartbeat(String, NodeHealth, Timestamp)
ConfigUpdate(String, ConfigSnapshot)
Command(String, String, String, Timestamp)
EventReplay(MessageEnvelope)
Receipt(String, WireAckStatus)
ErrorResponse(String, String)
} derive(ToJson, FromJson)fn replay_pending(store : LocalStore, config~ : SyncConfig, cursor~ : SyncCursor, sender~ : (MessageEnvelope) -> SyncSendResult) -> ReplayReport raise StoreErrorfn wire_max_bytes() -> IntInstall
Download zipOffline-first autonomy platform for weak-network edge devices