feat(daemoneye-agent): implement agent loading state, heartbeat detection, and recovery

[unclesp1d3r]

Summary

Loading State Machine

• Implement agent loading state machine (Loading → Ready → SteadyState) with error states
• Add collectors configuration module for /etc/daemoneye/collectors.json with JSON schema and validation
• Add collector readiness tracking via registration - collectors marked ready when they register
• Implement startup timeout handling with configurable per-collector timeouts
• Implement "begin monitoring" broadcast on control.collector.lifecycle topic
• Add privilege dropping stub for future platform-specific implementation
• Integrate loading state coordination into main.rs startup sequence

Heartbeat Detection & Recovery

• Add HeartbeatStatus enum (Healthy, Degraded, Failed) with needs_recovery() helper
• Implement missed heartbeat tracking in CollectorRegistry with check_heartbeats()
• Create escalating RecoveryAction chain: HealthCheck → GracefulShutdown → ForceKill → Restart
• Add CollectorRecoveryState for tracking recovery attempts per collector
• Implement execute_recovery() function for automated escalating recovery

Test Plan

• 51 unit tests pass for broker_manager and collector_config modules
• 10 integration tests pass for loading state workflow
• 21 integration tests pass for heartbeat detection and recovery
• 10 unit tests pass for recovery module
• All existing integration tests continue to pass
• cargo clippy -- -D warnings passes with zero warnings
• Pre-commit hooks (fmt, clippy, check, audit) all pass

🤖 Generated with Claude Code

[coderabbitai]

Caution

Review failed

Failed to post review comments

Summary by CodeRabbit

• New Features

  • Collector configuration loading and validation; startup sequencing that waits for collectors, exposes agent state/readiness, and broadcasts begin-monitoring; heartbeat health statuses and an escalating recovery workflow.

• Documentation

  • Added coding guidelines and pre-commit gotchas (naming, intentional-cast annotation, tracing macro await rules, formatter note).

• Tests

  • Extensive unit & integration tests covering config, startup flow, readiness, heartbeat detection, and recovery.

• Chores

  • Minor spec status update.

Walkthrough

Adds collector configuration loading/validation, an agent startup state machine with collector readiness tracking, missed‑heartbeat detection and an escalating recovery workflow, plus public re-exports and comprehensive unit/integration tests.

Changes

Cohort / File(s)	Summary
Configuration daemoneye-agent/src/collector_config.rs, daemoneye-agent/src/lib.rs	New CollectorsConfig and CollectorEntry types, platform defaults, serde support, file loading (load_from_file/load_or_default), validation, builder helpers, and public re-exports.
Broker & Startup State daemoneye-agent/src/broker_manager.rs, daemoneye-agent/src/main.rs, daemoneye-agent/tests/loading_state_integration.rs	Introduces AgentState state machine and CollectorReadinessTracker; readiness APIs and transitions (wait_for_collectors_ready, transition_to_ready/transition_to_steady_state, mark_startup_failed, transition_to_shutting_down), collectors config setters/getters, privilege-drop stub, begin‑monitoring broadcast, and lifecycle integration tests.
Heartbeat & Recovery daemoneye-agent/src/collector_registry.rs, daemoneye-agent/src/recovery.rs, daemoneye-agent/tests/heartbeat_detection_integration.rs	Adds missed_heartbeats to records, MAX_MISSED_HEARTBEATS, HeartbeatStatus and query/reset APIs, recovery module with RecoveryAction escalation, RecoveryResult/RecoveryError, CollectorRecoveryState, execute_recovery orchestration, action executors, and extensive tests.
Public Surface & Re-exports daemoneye-agent/src/lib.rs	Exports new collector_config and recovery modules; re-exports CollectorConfigError/CollectorEntry/CollectorsConfig; adds AgentState and HeartbeatStatus to the public API.
Tests & Integration daemoneye-agent/tests/*, daemoneye-agent/src/*	Extensive unit and integration tests for config parsing/validation, readiness/startup flows, heartbeat status behavior, recovery escalation, and lifecycle scenarios.
Docs & Misc AGENTS.md, spec/procmond/index.md, codecov.yml	Docs: coding guidelines and pre-commit gotchas; spec checkbox update; coverage targets/ignore-list adjusted.

Sequence Diagram(s)

sequenceDiagram
    participant Agent as Agent (main)
    participant Config as ConfigLoader
    participant Broker as BrokerManager
    participant Registry as CollectorRegistry
    participant Collector as Collector

    Agent->>Config: load_collectors_config()
    Config-->>Agent: CollectorsConfig
    Agent->>Broker: set_collectors_config(config)
    Agent->>Broker: wait_for_collectors_ready(timeout, poll)
    Broker->>Registry: set_expected_collectors(...)

    loop registrations / polling
        Collector->>Registry: register()
        Registry-->>Broker: registration_event(id)
        Broker->>Broker: mark_collector_ready(id)
        Broker-->>Agent: readiness_update
    end

    Broker-->>Agent: all_collectors_ready -> transition_to_ready()
    Agent->>Broker: transition_to_steady_state()
    Agent->>Broker: broadcast_begin_monitoring()
    Agent->>Broker: drop_privileges()

Loading

Estimated code review effort

🎯 4 (Complex) | ⏱️ ~60 minutes

Possibly related PRs

• feat(procmond): implement WAL and EventBus connector integration #127: Overlaps changes to broker_manager.rs (agent state / startup readiness and related fields).

Poem

🐰 I nibbled configs, checked each tiny part,
I nudged the broker with a hopeful heart,
Collectors chimed in, one, two, then all,
Heartbeats guarded, recoveries on call,
Hooray — the agent wakes; I jump and start! 🥕

🚥 Pre-merge checks | ✅ 3

✅ Passed checks (3 passed)

Check name	Status	Explanation
Title check	✅ Passed	The title accurately captures the main features implemented: agent loading state, heartbeat detection, and recovery mechanism for daemoneye-agent.
Description check	✅ Passed	The description is directly related to the changeset, covering the three major components (loading state machine, heartbeat detection, recovery) with detailed subsections and test results.
Docstring Coverage	✅ Passed	Docstring coverage is 100.00% which is sufficient. The required threshold is 80.00%.

_{✏️ Tip: You can configure your own custom pre-merge checks in the settings.}

✨ Finishing touches

• 📝 Generate docstrings

🧪 Generate unit tests (beta)

• Create PR with unit tests
• Post copyable unit tests in a comment
• Commit unit tests in branch implement-agent-loading-state-and-heartbeat-detection

---

_{Comment @coderabbitai help to get the list of available commands and usage tips.}

[dosubot]

Related Documentation

Checked 21 published document(s) in 1 knowledge base(s). No updates required.

^{How did I do? Any feedback?}  

[qltysh]

1 new issue

Tool	Category	Rule	Count
qlty	Structure	Function with many returns (count = 8): run	1

[Copilot]

Pull request overview

This PR implements an agent loading state machine for coordinating startup between daemoneye-agent and its collectors. The state machine ensures all collectors are ready before transitioning to normal operation.

Changes:

• Implemented agent loading state machine (Loading → Ready → SteadyState) with error states in broker_manager.rs
• Added collectors configuration module in collector_config.rs for loading and validating /etc/daemoneye/collectors.json
• Integrated loading state coordination into main.rs startup sequence with timeout handling
• Added 10 integration tests and 51 unit tests for comprehensive coverage
• Updated spec to mark Ticket 3 as complete

Reviewed changes

Copilot reviewed 7 out of 7 changed files in this pull request and generated 6 comments.

Show a summary per file

File	Description
spec/procmond/index.md	Marked Ticket 3 (RPC Service and Registration Manager) as complete
daemoneye-agent/tests/loading_state_integration.rs	New integration tests for loading state workflow including timeout, registration, and full lifecycle
daemoneye-agent/src/main.rs	Integrated loading state coordination with configuration loading, collector waiting, privilege dropping stub, and state transitions
daemoneye-agent/src/lib.rs	Exported new public API items for collector configuration
daemoneye-agent/src/collector_config.rs	New module for loading, validating, and managing collector configurations from JSON
daemoneye-agent/src/broker_manager.rs	Implemented AgentState enum, CollectorReadinessTracker, state transition methods, and integrated with registration
AGENTS.md	Added coding guidelines for ignored variables in closures and as_conversions

[Copilot]

The "begin monitoring" broadcast is being performed twice: once explicitly here (line 204) and again inside transition_to_steady_state() (broker_manager.rs line 951). This duplication means collectors will receive the BeginMonitoring message twice.

Either remove the explicit broadcast call here and let transition_to_steady_state() handle it, or remove the broadcast from inside transition_to_steady_state() and keep it here. The recommended approach is to remove the explicit call here since the transition method's documentation states it broadcasts "begin monitoring" during the transition.

[Copilot]

State transition and broadcast are not atomic. If broadcast_begin_monitoring() fails, the state will already have been changed to SteadyState (line 947), but the error will propagate (line 951 with ?). This leaves the agent in SteadyState even though the broadcast failed.

Consider one of these approaches:

1. Perform the broadcast BEFORE changing state, so failure prevents the transition
2. Revert the state on broadcast failure
3. Make broadcast failure non-fatal (remove ? and log error) since collectors might still receive the message through other means

The current main.rs treats broadcast failure as non-fatal (line 204-208), so option 3 aligns with that design.

Suggested change

	self.broadcast_begin_monitoring().await?;
	if let Err(error) = self.broadcast_begin_monitoring().await {
	warn!(%error, "Failed to broadcast begin monitoring; continuing in SteadyState");
	}

[Copilot]

This test broadcasts "begin monitoring" explicitly at line 288, and then calls transition_to_steady_state() at line 291. However, transition_to_steady_state() also broadcasts "begin monitoring" internally (broker_manager.rs line 951). This means the broadcast happens twice in this test.

This test should match the actual usage pattern. Remove line 288 to let transition_to_steady_state() handle the broadcast, or adjust the test to match the corrected implementation once the duplicate broadcast issue in main.rs is resolved.

Suggested change

	// 6. Broadcast begin monitoring
	manager.broadcast_begin_monitoring().await?;
	// 7. Transition to SteadyState
	// 6. Transition to SteadyState

[Copilot]

Using unwrap_or_default() here means if the system time is before Unix epoch (an extremely rare but possible scenario), the timestamp will silently be 0 milliseconds instead of propagating an error. This could cause confusion in logs or collector behavior.

Consider either:

1. Using a proper error with context: .context("System time is before Unix epoch")?
2. Adding a warning log when the fallback is used
3. Documenting this behavior if it's intentional

Given that this is for a "BeginMonitoring" message, option 1 (failing fast) is recommended since it indicates a serious system misconfiguration.

Suggested change

	let message = serde_json::json!({
	"type": "BeginMonitoring",
	"timestamp": std::time::SystemTime::now()
	.duration_since(std::time::SystemTime::UNIX_EPOCH)
	.unwrap_or_default()
	.as_millis(),
	let timestamp = std::time::SystemTime::now()
	.duration_since(std::time::SystemTime::UNIX_EPOCH)
	.context("System time is before Unix epoch when creating BeginMonitoring timestamp")?
	.as_millis();
	let message = serde_json::json!({
	"type": "BeginMonitoring",
	"timestamp": timestamp,

[Copilot]

The documentation states that in the Ready state "privileges dropped", but this is inaccurate. Looking at main.rs lines 196-201, privilege dropping occurs AFTER the transition to Ready state. The Ready state only indicates that all collectors have registered.

Update the documentation to clarify: "All collectors registered and reported 'ready'. Waiting for privilege dropping before transitioning to steady state." or similar wording that doesn't imply privileges have already been dropped.

Suggested change

	//! - **Ready**: All collectors registered and reported "ready", privileges dropped
	//! - **Ready**: All collectors registered and reported "ready". Awaiting privilege
	//! dropping before transitioning to `SteadyState`.

[Copilot]

This documentation states "The agent has dropped privileges (if configured)" in the Ready state, but according to main.rs (lines 196-201), privilege dropping occurs AFTER transitioning to Ready. The Ready state should only indicate that all collectors have registered.

Update this to: "All collectors have registered and reported 'ready'. Privilege dropping will occur before transitioning to steady state." to match the actual implementation.

Suggested change

	/// The agent has dropped privileges (if configured).
	/// Privilege dropping will occur before transitioning to steady state.

[Copilot]

Pull request overview

Copilot reviewed 10 out of 10 changed files in this pull request and generated 10 comments.

[Copilot]

The AgentState enum is used in integration tests (line 19 in loading_state_integration.rs) but is not re-exported in lib.rs. While the tests currently access it via daemoneye_agent::broker_manager::AgentState, it would be more consistent with the codebase pattern to re-export it in lib.rs alongside BrokerManager, similar to how HeartbeatStatus is exported alongside CollectorRegistry.

This would make the public API more discoverable and consistent.

Suggested change

	pub use broker_manager::{BrokerHealth, BrokerManager};
	pub use broker_manager::{AgentState, BrokerHealth, BrokerManager};

[Copilot]

In the execute_recovery function, the attempts counter is incremented at the start of each loop iteration (line 240), but the check for exhaustion happens after that (line 242). This means the function will attempt recovery MAX_RECOVERY_ATTEMPTS + 1 times instead of MAX_RECOVERY_ATTEMPTS times.

For example, with MAX_RECOVERY_ATTEMPTS = 3:

• Iteration 1: attempts becomes 1, check passes (1 <= 3), execute action
• Iteration 2: attempts becomes 2, check passes (2 <= 3), execute action
• Iteration 3: attempts becomes 3, check passes (3 <= 3), execute action
• Iteration 4: attempts becomes 4, check fails (4 > 3), return exhausted

The check should use >= instead of > to ensure exactly MAX_RECOVERY_ATTEMPTS attempts are made.

Suggested change

	if attempts > MAX_RECOVERY_ATTEMPTS {
	if attempts >= MAX_RECOVERY_ATTEMPTS {

[Copilot]

The broadcast_begin_monitoring call on line 204 is made before transitioning to SteadyState, but according to the AgentState documentation (lines 103-105 in broker_manager.rs), the "begin monitoring" broadcast should happen when entering SteadyState. The current implementation contradicts this design.

Additionally, the transition_to_steady_state method (lines 938-980 in broker_manager.rs) already calls broadcast_begin_monitoring internally (line 951), so calling it twice could result in collectors receiving duplicate "begin monitoring" messages. The explicit call on line 204 should be removed.

Suggested change

	// Broadcast "begin monitoring" to all collectors
	if let Err(e) = broker_manager.broadcast_begin_monitoring().await {
	error!(error = %e, "Failed to broadcast begin monitoring");
	// Continue anyway - broadcast failure is not fatal for the main loop
	warn!("Collectors may not have received begin monitoring signal");
	}
	// Transition to SteadyState
	// Transition to SteadyState (this will broadcast "begin monitoring")

[Copilot]

The binary path validation on line 270 uses std::fs::metadata(path)? which will propagate an IoError if the file exists but metadata cannot be read (e.g., due to permission issues). This IoError will be converted to CollectorConfigError::IoError via the #[from] attribute, but the error message won't clearly indicate whether the issue is with reading metadata vs. reading the config file itself.

Consider catching this specific error and returning a more descriptive CollectorConfigError::ValidationError that indicates the binary exists but metadata couldn't be read.

Suggested change

	let metadata = std::fs::metadata(path)?;
	let metadata = match std::fs::metadata(path) {
	Ok(metadata) => metadata,
	Err(e) => {
	// If the file disappeared between the exists() check and metadata(),
	// treat it as a missing binary; otherwise surface a clear validation error.
	if e.kind() == std::io::ErrorKind::NotFound {
	warn!(
	collector_id = %collector_id,
	path = %path.display(),
	"Collector binary vanished before metadata check"
	);
	return Err(CollectorConfigError::BinaryNotFound {
	path: path.to_path_buf(),
	});
	}
	warn!(
	collector_id = %collector_id,
	path = %path.display(),
	error = %e,
	"Collector binary exists but metadata could not be read"
	);
	return Err(CollectorConfigError::ValidationError {
	message: format!(
	"collector '{}' binary at '{}' exists but metadata could not be read: {}",
	collector_id,
	path.display(),
	e
	),
	});
	}
	};

[Copilot]

The check_heartbeats method uses .unwrap_or(Duration::ZERO) on line 117 when duration_since fails. This can happen if the system clock is adjusted backwards, making now earlier than record.last_heartbeat. In this case, Duration::ZERO will be used, which means the heartbeat will be considered current even though the clock may have jumped backwards significantly.

This could prevent detection of failed collectors if the system clock is adjusted backwards during operation. Consider handling clock adjustments explicitly, either by detecting backwards time jumps and logging a warning, or by resetting heartbeat tracking when this occurs.

Suggested change

	let elapsed = now
	.duration_since(record.last_heartbeat)
	.unwrap_or(Duration::ZERO);
	let elapsed = match now.duration_since(record.last_heartbeat) {
	Ok(elapsed) => elapsed,
	Err(_e) => {
	// System clock appears to have moved backwards relative to the last
	// recorded heartbeat. Reset the heartbeat timestamp to `now` and
	// skip this collector for this check cycle to avoid incorrectly
	// treating the heartbeat as either stale or fresh.
	record.last_heartbeat = now;
	continue;
	}
	};

[Copilot]

In the execute_force_kill function, line 392 passes graceful: true to stop_collector, which contradicts the function's purpose of performing a force kill. This should be graceful: false to properly send a kill signal (SIGKILL on Unix) rather than a graceful termination signal (SIGTERM).

The timeout wrapping also creates a potential issue: the outer timeout is FORCE_KILL_TIMEOUT (10s) and the inner timeout passed to stop_collector is also FORCE_KILL_TIMEOUT (10s). If the inner operation times out, the outer timeout will also time out simultaneously, making it ambiguous which timeout occurred.

[Copilot]

In the execute_restart function, the restart operation is wrapped with RESTART_TIMEOUT (60s) for both the outer tokio::timeout and the inner restart_collector timeout parameter. This double timeout creates the same ambiguity issue as in execute_force_kill: if the inner operation times out at 60s, the outer timeout will also trigger simultaneously, making it unclear which timeout occurred.

Additionally, after a successful restart, the collector will be in an initial state and needs time to re-register and become ready. The recovery should potentially wait for re-registration or at least reset the heartbeat tracking for this collector.

[Copilot]

The broadcast_begin_monitoring method uses .unwrap_or_default() on line 1171 which will return Duration::ZERO (epoch time 0ms) if the system clock is set before Unix epoch. This could cause collectors to receive a BeginMonitoring message with timestamp 0, potentially causing confusion or incorrect behavior.

Consider using .unwrap_or_else() with a logged error or returning an error if the system time cannot be determined, as a system with an invalid clock may have other issues that should be surfaced.

Suggested change

	// Create the lifecycle message
	let message = serde_json::json!({
	"type": "BeginMonitoring",
	"timestamp": std::time::SystemTime::now()
	.duration_since(std::time::SystemTime::UNIX_EPOCH)
	.unwrap_or_default()
	.as_millis(),
	// Compute a monotonic-ish timestamp; log if the system clock is invalid
	let timestamp_ms = match std::time::SystemTime::now()
	.duration_since(std::time::SystemTime::UNIX_EPOCH)
	{
	Ok(duration) => duration.as_millis(),
	Err(err) => {
	warn!(
	?err,
	"System clock is before UNIX_EPOCH; using timestamp 0 for BeginMonitoring"
	);
	0
	}
	};
	// Create the lifecycle message
	let message = serde_json::json!({
	"type": "BeginMonitoring",
	"timestamp": timestamp_ms,

[Copilot]

The heartbeat_status method has the same clock adjustment issue as check_heartbeats, using .unwrap_or(Duration::ZERO) on line 170 when the system clock may have been adjusted backwards. This will incorrectly report a collector as Healthy even if significant time has passed since the last heartbeat.

Suggested change

	let elapsed = now
	.duration_since(record.last_heartbeat)
	.unwrap_or(Duration::ZERO);
	let elapsed = match now.duration_since(record.last_heartbeat) {
	Ok(duration) => duration,
	Err(_) => {
	// System clock moved backwards; cannot reliably determine elapsed time.
	return None;
	}
	};

[Copilot]

The spec update marks Ticket 3 as complete (line 28 changes from [ ] to [x]), but according to the PR description and code changes in this PR, this PR implements Ticket 4 (Agent Loading State and Heartbeat Detection), not Ticket 3 (RPC Service and Registration Manager for procmond).

The correct ticket to mark complete would be Ticket 4 on line 36, not Ticket 3. Ticket 3 relates to procmond's RPC implementation, while this PR implements daemoneye-agent's loading state machine and heartbeat detection.

[codecov]

Codecov Report

❌ Patch coverage is 75.85693% with 324 lines in your changes missing coverage. Please review.

Files with missing lines	Patch %	Lines
daemoneye-agent/src/recovery.rs	49.60%	127 Missing ⚠️
daemoneye-agent/src/main.rs	0.00%	66 Missing ⚠️
daemoneye-agent/src/broker_manager.rs	92.44%	46 Missing ⚠️
daemoneye-agent/src/collector_config.rs	87.24%	44 Missing ⚠️
daemoneye-agent/src/collector_registry.rs	41.42%	41 Missing ⚠️

📢 Thoughts on this report? Let us know!