Merge branch 'main' of https://github.com/100monkeys-ai/aegis-orchestrator

Author: Theaxiom

cli/src/daemon/server.rs

@@ -19,9 +19,6 @@ use axum::{
     Json, Router,
 };
 
-/// Maximum number of retries when attempting to establish the Temporal connection.
-/// Previously this was an inline magic number (`30`) in the connection retry loop.
-const TEMPORAL_CONNECTION_MAX_RETRIES: i32 = 30;
 use std::sync::Arc;
 
 // Type alias for repository tuple to avoid clippy "very complex type" lint
@@ -80,11 +77,7 @@ fn default_local_host_mount_point() -> String {
         return path;
     }
 
-    let default_path = PathBuf::from("/")
-        .join("var")
-        .join("lib")
-        .join("aegis")
-        .join("local-host-volumes");
+    let default_path = PathBuf::from("/var/lib/aegis/local-host-volumes");
     default_path.to_string_lossy().into_owned()
 }
 
@@ -618,6 +611,10 @@ pub async fn start_daemon(config_path: Option<PathBuf>, port: u16) -> Result<()>
     let temporal_address_clone = temporal_address.clone();
     let worker_http_endpoint_clone = worker_http_endpoint.clone();
 
+    /// Maximum number of retries when attempting to establish the Temporal connection.
+    /// Previously this was an inline magic number (`30`) in the connection retry loop.
+    const TEMPORAL_CONNECTION_MAX_RETRIES: i32 = 30;
+
     // Spawn background task to connect
     tokio::spawn(async move {
         let mut retries: i32 = 0;
@@ -959,7 +956,7 @@ pub async fn start_daemon(config_path: Option<PathBuf>, port: u16) -> Result<()>
                         Arc::new(manager)
                     }
                     Err(e) => {
-                        return Err(anyhow::anyhow!("Failed to initialise OpenBao secrets manager: {}", e));
+                        return Err(anyhow::anyhow!("Failed to initialize OpenBao secrets manager: {}", e));
                     }
                 }
             }
@@ -1651,16 +1648,26 @@ async fn stream_events_handler(
 
             // Execution Terminal State
             if let Some(ended_at) = execution.ended_at {
+                /// Extract the final output from the last iteration of an execution.
+                ///
+                /// Returns `None` if the execution has no iterations or if the last iteration
+                /// produced no output.
+                fn final_output_from_execution(
+                    execution: &aegis_orchestrator_core::domain::execution::Execution,
+                ) -> Option<String> {
+                    execution
+                        .iterations()
+                        .last()
+                        .and_then(|i| i.output.clone())
+                }
+
                 match execution.status {
                     aegis_orchestrator_core::domain::execution::ExecutionStatus::Completed => {
                          // NOTE: The Execution struct does not expose an explicit `final_output` field.
                          // By convention, we treat the last iteration's `output` (if any) as the final result.
                          // If no such output exists, we surface `null` and keep the semantics explicit instead of
                          // silently defaulting to an empty string.
-                        let final_output = execution
-                            .iterations()
-                            .last()
-                            .and_then(|i| i.output.clone());
+                        let final_output = final_output_from_execution(&execution);
 
                         let exec_end = serde_json::json!({
                             "event_type": "ExecutionCompleted",
@@ -1932,7 +1939,7 @@ async fn stream_agent_events_handler(
                 if let Some(ended_at) = execution.ended_at {
                     match execution.status {
                         aegis_orchestrator_core::domain::execution::ExecutionStatus::Completed => {
-                            let result = execution.iterations().last().and_then(|i| i.output.clone()).unwrap_or_default();
+                            let result = execution.iterations().last().and_then(|i| i.output.clone());
                             let exec_end = serde_json::json!({
                                 "event_type": "ExecutionCompleted",
                                 "execution_id": execution.id.0,

docker/runtime-entrypoint.sh

@@ -11,9 +11,17 @@ else
 fi
 if [ -n "$SOCK_GID" ]; then
   if ! getent group "$SOCK_GID" >/dev/null 2>&1; then
-    groupadd -f -g "$SOCK_GID" hostdocker >/dev/null 2>&1 || true
+    if ! groupadd -f -g "$SOCK_GID" hostdocker >/dev/null 2>&1; then
+      # If group creation failed, re-check whether a group with this GID now exists.
+      if ! getent group "$SOCK_GID" >/dev/null 2>&1; then
+        echo "Failed to create group with GID $SOCK_GID for Docker socket access" >&2
+        exit 1
+      fi
+    fi
+  fi
+  if ! usermod -aG "$SOCK_GID" aegis >/dev/null 2>&1; then
+    echo "Warning: failed to add user 'aegis' to group GID ${SOCK_GID}; Docker socket access may not work." >&2
   fi
-  usermod -aG "$SOCK_GID" aegis >/dev/null 2>&1 || true
 fi
 
 if [ "$#" -eq 0 ]; then

orchestrator/core/src/application/tool_invocation_service.rs

@@ -118,7 +118,7 @@ impl ToolInvocationService {
         envelope: &impl EnvelopeVerifier,
     ) -> Result<Value, SmcpSessionError> {
         // 1. Get active session for agent
-        let session = self
+        let mut session = self
             .smcp_session_repo
             .find_active_by_agent(agent_id)
             .await
@@ -130,7 +130,7 @@ impl ToolInvocationService {
             ))?;
 
         // 2. Middleware verifies signature and evaluates against SecurityContext
-        let args = self.smcp_middleware.verify_and_unwrap(&session, envelope)?;
+        let args = self.smcp_middleware.verify_and_unwrap(&mut session, envelope)?;
         let tool_name = envelope
             .extract_tool_name()
             .ok_or(SmcpSessionError::MalformedPayload(

orchestrator/core/src/domain/smcp_session.rs

@@ -2,6 +2,10 @@
 // SPDX-License-Identifier: AGPL-3.0
 //! # SMCP Session Aggregate (BC-12, ADR-035)
 //!
+//! `BC-12` refers to the bounded context that owns SMCP session lifecycle logic, and
+//! `ADR-035` is the architecture decision record that defines the design of this
+//! aggregate and its invariants (see the project's architecture decision records).
+//!
 //! Domain model for the **Secure Model Context Protocol** session lifecycle.
 //! Each agent execution that uses MCP tools goes through an attestation handshake
 //! (see [`crate::application::attestation_service`]) to receive a [`SmcpSession`],
@@ -53,6 +57,11 @@ pub struct SessionId(pub Uuid);
 
 impl SessionId {
     /// Generate a new random session ID.
+    ///
+    /// This uses a UUID v4 and relies on its statistical uniqueness; the probability of
+    /// collision is negligible for realistic volumes of sessions. Any additional collision
+    /// handling (for example, enforcing a unique constraint at the persistence layer) is
+    /// expected to be performed outside this constructor.
     pub fn new() -> Self {
         Self(Uuid::new_v4())
     }
@@ -206,7 +215,7 @@ impl SmcpSession {
             security_context,
             status: SessionStatus::Active,
             created_at: now,
-            expires_at: now + chrono::Duration::hours(SESSION_TTL_HOURS),
+            expires_at: now + chrono::TimeDelta::hours(SESSION_TTL_HOURS),
         }
     }
 
@@ -232,7 +241,7 @@ impl SmcpSession {
     /// This is the **single enforcement point** for all SMCP policy checks. Every
     /// tool call from any agent must pass through this method before being forwarded
     /// to the MCP server. See ADR-035 §4 (Enforcement Architecture).
-    pub fn evaluate_call(&self, envelope: &impl EnvelopeVerifier) -> Result<(), SmcpSessionError> {
+    pub fn evaluate_call(&mut self, envelope: &impl EnvelopeVerifier) -> Result<(), SmcpSessionError> {
         let now = Utc::now();
 
         // 1. Check session is active
@@ -242,6 +251,9 @@ impl SmcpSession {
 
         // 2. Check not expired
         if now > self.expires_at {
+            // Once the session is past its expiry time, transition it to a terminal
+            // `Expired` state so that future calls observe a consistent status.
+            self.status = SessionStatus::Expired;
             return Err(SmcpSessionError::SessionExpired);
         }
 

orchestrator/core/src/infrastructure/repositories/postgres_execution.rs

@@ -206,10 +206,13 @@ impl ExecutionRepository for PostgresExecutionRepository {
                     // NOTE: Only `parent_execution_id` is persisted, so we cannot
                     // reliably reconstruct full multi-level ancestry here without
                     // additional queries. We therefore preserve the direct parent link
-                    // but treat depth/path as local to this execution.
+                    // and use a minimal non-zero depth to indicate that this execution
+                    // is at least one level below some root. The true depth and full
+                    // path may be greater and cannot be reconstructed without further
+                    // queries.
                     ExecutionHierarchy {
                         parent_execution_id: Some(ExecutionId(parent_id)),
-                        depth: 0,
+                        depth: 1,
                         path: vec![ExecutionId(id)],
                     }
                 }
@@ -301,7 +304,12 @@ impl ExecutionRepository for PostgresExecutionRepository {
             }?;
 
             let input: ExecutionInput =
-                serde_json::from_value(input_val).map_err(RepositoryError::from)?;
+                serde_json::from_value(input_val).map_err(|e| {
+                    RepositoryError::Serialization(format!(
+                        "Failed to deserialize execution input: {}",
+                        e
+                    ))
+                })?;
             let iterations: Vec<Iteration> =
                 serde_json::from_value(iterations_val).map_err(|e| {
                     RepositoryError::Serialization(format!(
@@ -320,7 +328,7 @@ impl ExecutionRepository for PostgresExecutionRepository {
             let hierarchy = match parent_execution_id {
                 Some(parent_id) => ExecutionHierarchy {
                     parent_execution_id: Some(ExecutionId(parent_id)),
-                    depth: 0,
+                    depth: 1,
                     path: vec![ExecutionId(id)],
                 },
                 None => ExecutionHierarchy::root(ExecutionId(id)),
@@ -404,7 +412,12 @@ impl ExecutionRepository for PostgresExecutionRepository {
             }?;
 
             let input: ExecutionInput =
-                serde_json::from_value(input_val).map_err(RepositoryError::from)?;
+                serde_json::from_value(input_val).map_err(|e| {
+                    RepositoryError::Serialization(format!(
+                        "Failed to deserialize execution input: {}",
+                        e
+                    ))
+                })?;
             let iterations: Vec<Iteration> =
                 serde_json::from_value(iterations_val).map_err(|e| {
                     RepositoryError::Serialization(format!(
@@ -423,7 +436,7 @@ impl ExecutionRepository for PostgresExecutionRepository {
             let hierarchy = match parent_execution_id {
                 Some(parent_id) => ExecutionHierarchy {
                     parent_execution_id: Some(ExecutionId(parent_id)),
-                    depth: 0,
+                    depth: 1,
                     path: vec![ExecutionId(id)],
                 },
                 None => ExecutionHierarchy::root(ExecutionId(id)),

orchestrator/core/src/infrastructure/smcp/middleware.rs

@@ -57,7 +57,7 @@ impl SmcpMiddleware {
     /// the tool server, preserving credential isolation (ADR-033).
     pub fn verify_and_unwrap(
         &self,
-        session: &SmcpSession,
+        session: &mut SmcpSession,
         envelope: &impl EnvelopeVerifier,
     ) -> Result<Value, SmcpSessionError> {
         info!("Verifying SMCP envelope for session {}", session.id);

orchestrator/core/src/infrastructure/temporal_event_listener.rs

@@ -66,6 +66,10 @@ use serde::{Deserialize, Serialize};
 use std::sync::Arc;
 use uuid::Uuid;
 
+/// Canonical file name used to store validation feedback artifacts produced by refinement
+/// workflows. This is consumed by the `RefinementApplied` event handler, which expects any
+/// validation feedback generated during a refinement iteration to be written under this
+/// name within the associated artifact set or storage location.
 const VALIDATION_FEEDBACK_FILE_NAME: &str = "validation_feedback";
 
 /// External event payload from Temporal worker
@@ -294,6 +298,47 @@ impl TemporalEventListener {
         }
     }
 
+    /// Persist an execution-scoped event to the repository and publish it to the event bus.
+    ///
+    /// This helper encapsulates the two-step pattern used for all execution events:
+    /// 1. Serialise the raw payload and append it to the event log via the repository.
+    /// 2. Publish the mapped domain event to the in-process event bus for subscribers.
+    ///
+    /// # Arguments
+    ///
+    /// * `execution_id` - The execution this event belongs to.
+    /// * `temporal_sequence_number` - The Temporal sequence number for ordering.
+    /// * `event_type` - The string event type name.
+    /// * `raw_payload` - The original Temporal payload to persist as JSON.
+    /// * `iteration_number` - Optional iteration number associated with this event.
+    /// * `domain_event` - The mapped domain event to publish after persistence.
+    async fn persist_and_publish_execution_event(
+        &self,
+        execution_id: ExecutionId,
+        temporal_sequence_number: i64,
+        event_type: String,
+        raw_payload: &TemporalEventPayload,
+        iteration_number: Option<u8>,
+        domain_event: crate::domain::events::ExecutionEvent,
+    ) -> Result<()> {
+        let serialized_payload = serde_json::to_value(raw_payload)
+            .context("Failed to serialize TemporalEventPayload for persistence")?;
+
+        self.execution_repository
+            .append_event(
+                execution_id,
+                temporal_sequence_number,
+                event_type,
+                serialized_payload,
+                iteration_number,
+            )
+            .await
+            .context("Failed to persist execution event")?;
+
+        self.event_bus.publish_execution_event(domain_event);
+        Ok(())
+    }
+
     /// Process incoming event from Temporal worker
     ///
     /// # Arguments
@@ -331,18 +376,22 @@ impl TemporalEventListener {
                 .ok_or_else(|| anyhow!("Missing code_diff for RefinementApplied event"))?;
             let diff_str = match diff_val {
                 serde_json::Value::String(s) => s,
-                _ => diff_val.to_string(),
+                other => {
+                    return Err(anyhow!(
+                        "Invalid code_diff format for RefinementApplied event: expected string, got {}",
+                        other
+                    ));
+                }
             };
 
             let code_diff = crate::domain::execution::CodeDiff {
                 file_path: VALIDATION_FEEDBACK_FILE_NAME.to_string(),
                 diff: diff_str,
             };
 
-            let applied_at = match DateTime::parse_from_rfc3339(&payload.timestamp) {
-                Ok(dt) => dt.with_timezone(&Utc),
-                Err(_) => Utc::now(),
-            };
+            let applied_at = DateTime::parse_from_rfc3339(&payload.timestamp)
+                .context("Failed to parse timestamp as RFC3339 for RefinementApplied event")?
+                .with_timezone(&Utc);
 
             let domain_event = crate::domain::events::ExecutionEvent::RefinementApplied {
                 execution_id,
@@ -352,18 +401,16 @@ impl TemporalEventListener {
                 applied_at,
             };
 
-            self.execution_repository
-                .append_event(
-                    execution_id,
-                    payload.temporal_sequence_number,
-                    payload.event_type.clone(),
-                    serde_json::to_value(&payload)?,
-                    Some(iteration_number),
-                )
-                .await
-                .context("Failed to persist execution event")?;
-
-            self.event_bus.publish_execution_event(domain_event);
+            self.persist_and_publish_execution_event(
+                execution_id,
+                payload.temporal_sequence_number,
+                payload.event_type.clone(),
+                &payload,
+                Some(iteration_number),
+                domain_event,
+            )
+            .await?;
+
             return Ok(payload.execution_id.clone());
         }
 
@@ -389,7 +436,12 @@ impl TemporalEventListener {
             | WorkflowEvent::WorkflowExecutionCompleted { execution_id, .. }
             | WorkflowEvent::WorkflowExecutionFailed { execution_id, .. }
             | WorkflowEvent::WorkflowExecutionCancelled { execution_id, .. } => *execution_id,
-            WorkflowEvent::WorkflowRegistered { .. } => unreachable!("handled above"),
+            WorkflowEvent::WorkflowRegistered { .. } => {
+                return Err(anyhow!(
+                    "WorkflowRegistered event unexpectedly reached execution-scoped handling; \
+                     this variant should be handled before deriving an execution_id"
+                ));
+            }
         };
 
         let execution_id_str = execution_id_obj.0.to_string();